X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Ftapi%2Ftest03_tapi_device_change_notifications.py;h=5d2364702dd5c64b9960e03623e4baf23d7de5b6;hb=db87ba733190c2e59997e00e0fe9199c9c74234d;hp=e36e3052da5adb9c10abce500da47f243f60d76f;hpb=90ecc9a5431eec82b8f4e7cb7d66cd29884e8f9a;p=transportpce.git diff --git a/tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py b/tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py index e36e3052d..5d2364702 100644 --- a/tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py +++ b/tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py @@ -11,7 +11,6 @@ # pylint: disable=no-member # pylint: disable=too-many-public-methods import os -import json import unittest import time import requests @@ -32,6 +31,12 @@ class UuidServices: self.odu = None self.dsr = None self.eth = None +# XPDR-A1-XPDR1+XPONDER Uuid is: +# 4378fc29-6408-39ec-8737-5008c3dc49e5 +# XPDR-C1-XPDR1+XPONDER Uuid is: +# 1770bea4-b1da-3b20-abce-7d182c0ec0df +# ROADM-A1+PHOTONIC_MEDIA Uuid is: +# 3b726367-6f2d-3e3f-9033-d99b61459075 class TransportPCEFulltesting(unittest.TestCase): @@ -76,29 +81,41 @@ class TransportPCEFulltesting(unittest.TestCase): } ], "connectivity-constraint": { - "service-layer": "ETH", + # "service-layer": "ETH", "service-type": "POINT_TO_POINT_CONNECTIVITY", "service-level": "Some service-level", "requested-capacity": { "total-size": { "value": "100", - "unit": "GB" + "unit": "tapi-common:CAPACITY_UNIT_GBPS" } } }, - "state": "Some state" + "topology-constraint": [ + { + "local-id": "localIdTopoConstraint", + "name": [ + { + "value-name": "Dumb constraint", + "value": "for debug1" + } + ] + } + ], + "state": "LOCKED", + "layer-protocol-name": "ETH" } - del_serv_input_data = {"service-id-or-name": "TBD"} + del_serv_input_data = {"uuid": "TBD"} - tapi_topo = {"topology-id-or-name": "TBD"} + tapi_topo = {"topology-id": "TBD"} node_details = { - "topology-id-or-name": "TBD", - "node-id-or-name": "TBD" + "topology-id": "TBD", + "node-id": "TBD" } - tapi_serv_details = {"service-id-or-name": "TBD"} + tapi_serv_details = {"uuid": "TBD"} processes = [] uuid_services = UuidServices() @@ -113,7 +130,7 @@ class TransportPCEFulltesting(unittest.TestCase): os.environ['JAVA_MAX_MEM'] = '4096M' cls.processes = test_utils.start_tpce() # TAPI feature is not installed by default in Karaf - if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True': + if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True': print("installing tapi feature...") result = test_utils.install_karaf_feature("odl-transportpce-tapi") if result.returncode != 0: @@ -128,7 +145,6 @@ class TransportPCEFulltesting(unittest.TestCase): print("tapi installation feature failed...") test_utils.shutdown_process(cls.processes[0]) sys.exit(2) - cls.processes = test_utils.start_tpce() cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221), ('roadma', cls.NODE_VERSION_221), ('roadmc', cls.NODE_VERSION_221), @@ -162,7 +178,7 @@ class TransportPCEFulltesting(unittest.TestCase): response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) - def test_05_connect_xprdA_N1_to_roadmA_PP1(self): + def test_05_connect_xpdrA_N1_to_roadmA_PP1(self): response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', @@ -180,7 +196,7 @@ class TransportPCEFulltesting(unittest.TestCase): self.assertIn('Roadm Xponder links created successfully', response["output"]["result"]) time.sleep(2) - def test_07_connect_xprdC_N1_to_roadmC_PP1(self): + def test_07_connect_xpdrC_N1_to_roadmC_PP1(self): response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', @@ -241,9 +257,11 @@ class TransportPCEFulltesting(unittest.TestCase): input_dict_1 = {'administrative-state': 'LOCKED', 'lifecycle-state': 'PLANNED', 'operational-state': 'DISABLED', - 'service-type': 'POINT_TO_POINT_CONNECTIVITY', - 'service-layer': 'ETH', - 'connectivity-direction': 'BIDIRECTIONAL' + # 'service-type': 'POINT_TO_POINT_CONNECTIVITY', + # 'service-layer': 'ETH', + 'layer-protocol-name': 'ETH', + # 'connectivity-direction': 'BIDIRECTIONAL' + 'direction': 'BIDIRECTIONAL' } input_dict_2 = {'value-name': 'OpenROADM node id', 'value': 'XPDR-C1-XPDR1'} @@ -269,7 +287,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_13_get_connectivity_service_Ethernet(self): - self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth) + self.tapi_serv_details["uuid"] = str(self.uuid_services.eth) response = test_utils.transportpce_api_rpc_request( 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -279,17 +297,13 @@ class TransportPCEFulltesting(unittest.TestCase): self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED') def test_14_change_status_line_port_xpdrc(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1', + { "port-name": "1", "port-type": "CFP2", "administrative-state": "outOfService", - "port-qual": "xpdr-network"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "xpdr-network" + })) time.sleep(2) def test_15_check_update_portmapping(self): @@ -341,11 +355,12 @@ class TransportPCEFulltesting(unittest.TestCase): self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService') self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService') self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified") - time.sleep(1) + time.sleep(10) def test_17_check_update_tapi_neps(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi" + self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -363,33 +378,34 @@ class TransportPCEFulltesting(unittest.TestCase): "Operational State should be 'ENABLED'") self.assertEqual(nep['administrative-state'], 'UNLOCKED', "Administrative State should be 'UNLOCKED'") - self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR" - response = test_utils.transportpce_api_rpc_request( - 'tapi-topology', 'get-node-details', self.node_details) - self.assertEqual(response['status_code'], requests.codes.ok) - nep_list = response['output']['node']['owned-node-edge-point'] - for nep in nep_list: - if 'XPDR1-NETWORK1' in nep['name'][0]['value']: - self.assertEqual(nep['operational-state'], 'DISABLED', - "Operational State should be 'DISABLED'") - self.assertEqual(nep['administrative-state'], 'LOCKED', - "Administrative State should be 'LOCKED'") - nb_updated_neps += 1 - else: - self.assertEqual(nep['operational-state'], 'ENABLED', - "Operational State should be 'ENABLED'") - self.assertEqual(nep['administrative-state'], 'UNLOCKED', - "Administrative State should be 'UNLOCKED'") - self.assertEqual(nb_updated_neps, 4, "Only two xponder neps should have been modified") + self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR" +# response = test_utils.transportpce_api_rpc_request( +# 'tapi-topology', 'get-node-details', self.node_details) +# self.assertEqual(response['status_code'], requests.codes.ok) +# nep_list = response['output']['node']['owned-node-edge-point'] +# for nep in nep_list: +# if 'XPDR1-NETWORK1' in nep['name'][0]['value']: +# self.assertEqual(nep['operational-state'], 'DISABLED', +# "Operational State should be 'DISABLED'") +# self.assertEqual(nep['administrative-state'], 'LOCKED', +# "Administrative State should be 'LOCKED'") +# nb_updated_neps += 1 +# else: +# self.assertEqual(nep['operational-state'], 'ENABLED', +# "Operational State should be 'ENABLED'") +# self.assertEqual(nep['administrative-state'], 'UNLOCKED', +# "Administrative State should be 'UNLOCKED'") + self.assertEqual(nb_updated_neps, 3, "3 xponder neps (OTS, OTSI_MC, eODU) should have been modified") time.sleep(1) def test_18_check_update_tapi_links(self): - self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO + self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-topology-details', self.tapi_topo) time.sleep(2) self.assertEqual(response['status_code'], requests.codes.ok) link_list = response['output']['topology']['link'] + print(response['output']['topology']['link']) nb_updated_link = 0 for link in link_list: if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']): @@ -399,8 +415,8 @@ class TransportPCEFulltesting(unittest.TestCase): else: self.assertEqual(link['operational-state'], 'ENABLED') self.assertEqual(link['administrative-state'], 'UNLOCKED') - self.assertEqual(nb_updated_link, 2, - "Only two xponder-output/input & xponder-transi links should have been modified") + self.assertEqual(nb_updated_link, 1, + "Only one xponder-output/input bidirectional link should have been modified") time.sleep(1) def test_19_check_update_service_Ethernet(self): @@ -410,7 +426,8 @@ class TransportPCEFulltesting(unittest.TestCase): self.assertEqual(response['services'][0]['administrative-state'], 'inService') def test_20_check_update_connectivity_service_Ethernet(self): - self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth) + self.tapi_serv_details["uuid"] = str(self.uuid_services.eth) + print(str(self.uuid_services.eth)) response = test_utils.transportpce_api_rpc_request( 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -419,17 +436,13 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_21_restore_status_line_port_xpdrc(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1', + { "port-name": "1", "port-type": "CFP2", "administrative-state": "inService", - "port-qual": "xpdr-network"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "xpdr-network" + })) time.sleep(2) def test_22_check_update_portmapping_ok(self): @@ -462,8 +475,9 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_24_check_update_tapi_neps_ok(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi" + self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -474,20 +488,20 @@ class TransportPCEFulltesting(unittest.TestCase): self.assertEqual(nep['administrative-state'], 'UNLOCKED', "Administrative State should be 'UNLOCKED'") - self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR" - response = test_utils.transportpce_api_rpc_request( - 'tapi-topology', 'get-node-details', self.node_details) - self.assertEqual(response['status_code'], requests.codes.ok) - nep_list = response['output']['node']['owned-node-edge-point'] - for nep in nep_list: - self.assertEqual(nep['operational-state'], 'ENABLED', - "Operational State should be 'ENABLED'") - self.assertEqual(nep['administrative-state'], 'UNLOCKED', - "Administrative State should be 'UNLOCKED'") +# self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR" +# response = test_utils.transportpce_api_rpc_request( +# 'tapi-topology', 'get-node-details', self.node_details) +# self.assertEqual(response['status_code'], requests.codes.ok) +# nep_list = response['output']['node']['owned-node-edge-point'] +# for nep in nep_list: +# self.assertEqual(nep['operational-state'], 'ENABLED', +# "Operational State should be 'ENABLED'") +# self.assertEqual(nep['administrative-state'], 'UNLOCKED', +# "Administrative State should be 'UNLOCKED'") time.sleep(1) def test_25_check_update_tapi_links_ok(self): - self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO + self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-topology-details', self.tapi_topo) time.sleep(2) @@ -504,19 +518,15 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_13_get_connectivity_service_Ethernet() def test_28_change_status_port_roadma_srg(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1', + { "port-name": "C1", "logical-connection-point": "SRG1-PP1", "port-type": "client", "circuit-id": "SRG1", "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "roadm-external" + })) time.sleep(2) def test_29_check_update_portmapping(self): @@ -571,8 +581,9 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_31_check_update_tapi_neps(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -590,11 +601,11 @@ class TransportPCEFulltesting(unittest.TestCase): "Operational State should be 'ENABLED'") self.assertEqual(nep['administrative-state'], 'UNLOCKED', "Administrative State should be 'UNLOCKED'") - self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified") + self.assertEqual(nb_updated_neps, 2, "Only 2 roadm SRG-PP nep (OTS/MC)should have been modified") time.sleep(1) def test_32_check_update_tapi_links(self): - self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO + self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-topology-details', self.tapi_topo) time.sleep(2) @@ -619,19 +630,15 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_20_check_update_connectivity_service_Ethernet() def test_35_restore_status_port_roadma_srg(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1', + { "port-name": "C1", "logical-connection-point": "SRG1-PP1", "port-type": "client", "circuit-id": "SRG1", "administrative-state": "inService", - "port-qual": "roadm-external"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "roadm-external" + })) time.sleep(2) def test_36_check_update_portmapping_ok(self): @@ -649,8 +656,9 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_23_check_update_openroadm_topo_ok() def test_38_check_update_tapi_neps_ok(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -673,19 +681,15 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_13_get_connectivity_service_Ethernet() def test_42_change_status_line_port_roadma_deg(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1', + { "port-name": "L1", "logical-connection-point": "DEG2-TTP-TXRX", "port-type": "LINE", "circuit-id": "1", "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "roadm-external" + })) time.sleep(2) def test_43_check_update_portmapping(self): @@ -740,8 +744,9 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_45_check_update_tapi_neps(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -759,11 +764,11 @@ class TransportPCEFulltesting(unittest.TestCase): "Operational State should be 'ENABLED'") self.assertEqual(nep['administrative-state'], 'UNLOCKED', "Administrative State should be 'UNLOCKED'") - self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified") + self.assertEqual(nb_updated_neps, 4, "4 roadm NEPS should have been modified (OTS/OMS/MC/OTSI_MC") time.sleep(1) def test_46_check_update_tapi_links(self): - self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO + self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-topology-details', self.tapi_topo) time.sleep(2) @@ -789,19 +794,15 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_20_check_update_connectivity_service_Ethernet() def test_49_restore_status_line_port_roadma_deg(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1', + { "port-name": "L1", "logical-connection-point": "DEG2-TTP-TXRX", "port-type": "LINE", "circuit-id": "1", "administrative-state": "inService", - "port-qual": "roadm-external"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "roadm-external" + })) time.sleep(2) def test_50_check_update_portmapping_ok(self): @@ -823,19 +824,15 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_13_get_connectivity_service_Ethernet() def test_56_change_status_port_roadma_srg(self): - url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2" - body = {"ports": [{ + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2', + { "port-name": "C2", "logical-connection-point": "SRG1-PP2", "port-type": "client", "circuit-id": "SRG1", "administrative-state": "outOfService", - "port-qual": "roadm-external"}]} - response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"), - data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON, - auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD), - timeout=test_utils.REQUEST_TIMEOUT) - self.assertEqual(response.status_code, requests.codes.ok) + "port-qual": "roadm-external" + })) time.sleep(2) def test_57_check_update_portmapping(self): @@ -883,8 +880,9 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_59_check_update_tapi_neps(self): - self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO - self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA" + self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID +# self.node_details["node-id"] = uuid.UUID(bytes="ROADM-A1+PHOTONIC_MEDIA".bytes) + self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075" response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-node-details', self.node_details) self.assertEqual(response['status_code'], requests.codes.ok) @@ -902,11 +900,11 @@ class TransportPCEFulltesting(unittest.TestCase): "Operational State should be 'ENABLED'") self.assertEqual(nep['administrative-state'], 'UNLOCKED', "Administrative State should be 'UNLOCKED'") - self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified") + self.assertEqual(nb_updated_neps, 1, "Only 1 roadm neps OTS should have been modified for SRG2PP") time.sleep(1) def test_60_check_update_tapi_links(self): - self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO + self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID response = test_utils.transportpce_api_rpc_request( 'tapi-topology', 'get-topology-details', self.tapi_topo) time.sleep(2) @@ -932,7 +930,7 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_13_get_connectivity_service_Ethernet() def test_63_delete_connectivity_service_Ethernet(self): - self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.eth) + self.del_serv_input_data["uuid"] = str(self.uuid_services.eth) response = test_utils.transportpce_api_rpc_request( 'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data) self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content)) @@ -964,6 +962,27 @@ class TransportPCEFulltesting(unittest.TestCase): response = test_utils.unmount_device("ROADM-C1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) + def test_69_restore_status_port_roadma_srg(self): + self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2', + { + "port-name": "C2", + "logical-connection-point": "SRG1-PP2", + "port-type": "client", + "circuit-id": "SRG1", + "administrative-state": "inService", + "port-qual": "roadm-external" + })) + time.sleep(2) + + def test_70_clean_openroadm_topology(self): + response = test_utils.get_ietf_network_request('openroadm-topology', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) + links = response['network'][0]['ietf-network-topology:link'] + for link in links: + if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT', 'ROADM-TO-ROADM'): + response = test_utils.del_ietf_network_link_request('openroadm-topology', link['link-id'], 'config') + self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) + if __name__ == "__main__": unittest.main(verbosity=2)