X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest15_otn_end2end_with_intermediate_switch.py;fp=tests%2Ftransportpce_tests%2F2.2.1%2Ftest15_otn_end2end_with_intermediate_switch.py;h=8a2f924fdebc99f7a0c116ea38dd5a92e13c1691;hb=654a9fb915061ac8eb5ec52cacbc3c76a8455aa8;hp=fc4c1ba63ee916026fdcf4aaf7392c1c32ef2344;hpb=6233872e577c6abc704af61bc26362202017c103;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test15_otn_end2end_with_intermediate_switch.py b/tests/transportpce_tests/2.2.1/test15_otn_end2end_with_intermediate_switch.py index fc4c1ba63..8a2f924fd 100644 --- a/tests/transportpce_tests/2.2.1/test15_otn_end2end_with_intermediate_switch.py +++ b/tests/transportpce_tests/2.2.1/test15_otn_end2end_with_intermediate_switch.py @@ -21,7 +21,7 @@ import sys sys.path.append('transportpce_tests/common/') # pylint: disable=wrong-import-position # pylint: disable=import-error -import test_utils_rfc8040 # nopep8 +import test_utils # nopep8 class TransportPCEtesting(unittest.TestCase): @@ -135,56 +135,56 @@ class TransportPCEtesting(unittest.TestCase): @classmethod def setUpClass(cls): - cls.processes = test_utils_rfc8040.start_tpce() - cls.processes = test_utils_rfc8040.start_sims([('spdra', cls.NODE_VERSION), - ('spdrb', cls.NODE_VERSION), - ('spdrc', cls.NODE_VERSION), - ('roadma', cls.NODE_VERSION), - ('roadmb', cls.NODE_VERSION), - ('roadmc', cls.NODE_VERSION)]) + cls.processes = test_utils.start_tpce() + cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION), + ('spdrb', cls.NODE_VERSION), + ('spdrc', cls.NODE_VERSION), + ('roadma', cls.NODE_VERSION), + ('roadmb', cls.NODE_VERSION), + ('roadmc', cls.NODE_VERSION)]) @classmethod def tearDownClass(cls): # pylint: disable=not-an-iterable for process in cls.processes: - test_utils_rfc8040.shutdown_process(process) + test_utils.shutdown_process(process) print("all processes killed") def setUp(self): time.sleep(2) def test_001_connect_spdrA(self): - response = test_utils_rfc8040.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) + response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_002_connect_spdrB(self): - response = test_utils_rfc8040.mount_device("SPDR-SB1", ('spdrb', self.NODE_VERSION)) + response = test_utils.mount_device("SPDR-SB1", ('spdrb', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_003_connect_spdrC(self): - response = test_utils_rfc8040.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) + response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_004_connect_rdmA(self): - response = test_utils_rfc8040.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION)) + response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_005_connect_rdmB(self): - response = test_utils_rfc8040.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION)) + response = test_utils.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_006_connect_rdmC(self): - response = test_utils_rfc8040.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) + response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) self.assertEqual(response.status_code, - requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_007_connect_sprdA_1_N1_to_roadmA_PP1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -193,7 +193,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_008_connect_roadmA_PP1_to_spdrA_1_N1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-rdm-xpdr-links', {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -202,7 +202,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_009_connect_sprdC_1_N1_to_roadmC_PP1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -211,7 +211,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_010_connect_roadmC_PP1_to_spdrC_1_N1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-rdm-xpdr-links', {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -220,7 +220,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_011_connect_sprdB_2_N1_to_roadmB_PP1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'SPDR-SB1', 'xpdr-num': '2', 'network-num': '1', 'rdm-node': 'ROADM-B1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -229,7 +229,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_012_connect_roadmB_PP1_to_spdrB_2_N1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-rdm-xpdr-links', {'links-input': {'xpdr-node': 'SPDR-SB1', 'xpdr-num': '2', 'network-num': '1', 'rdm-node': 'ROADM-B1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -238,7 +238,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_013_connect_sprdB_2_N2_to_roadmB_PP2(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', {'links-input': {'xpdr-node': 'SPDR-SB1', 'xpdr-num': '2', 'network-num': '2', 'rdm-node': 'ROADM-B1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -247,7 +247,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_014_connect_roadmB_PP2_to_spdrB_2_N2(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-rdm-xpdr-links', {'links-input': {'xpdr-node': 'SPDR-SB1', 'xpdr-num': '2', 'network-num': '2', 'rdm-node': 'ROADM-B1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -267,8 +267,8 @@ class TransportPCEtesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) def test_016_add_omsAttributes_ROADMB_ROADMA(self): @@ -283,8 +283,8 @@ class TransportPCEtesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) def test_017_add_omsAttributes_ROADMB_ROADMC(self): @@ -299,8 +299,8 @@ class TransportPCEtesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) def test_018_add_omsAttributes_ROADMC_ROADMB(self): @@ -315,12 +315,12 @@ class TransportPCEtesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) def test_019_create_OTS_ROADMA_DEG1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-device-renderer', 'create-ots-oms', { 'node-id': 'ROADM-A1', @@ -331,7 +331,7 @@ class TransportPCEtesting(unittest.TestCase): response["output"]["result"]) def test_020_create_OTS_ROADMB_DEG1(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-device-renderer', 'create-ots-oms', { 'node-id': 'ROADM-B1', @@ -342,7 +342,7 @@ class TransportPCEtesting(unittest.TestCase): response["output"]["result"]) def test_021_create_OTS_ROADMB_DEG2(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-device-renderer', 'create-ots-oms', { 'node-id': 'ROADM-B1', @@ -353,7 +353,7 @@ class TransportPCEtesting(unittest.TestCase): response["output"]["result"]) def test_022_create_OTS_ROADMC_DEG2(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-device-renderer', 'create-ots-oms', { 'node-id': 'ROADM-C1', @@ -364,7 +364,7 @@ class TransportPCEtesting(unittest.TestCase): response["output"]["result"]) def test_023_calculate_span_loss_base_all(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'transportpce-olm', 'calculate-spanloss-base', { 'src-type': 'all' @@ -398,7 +398,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(5) def test_024_check_otn_topology(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['node']), 9, 'There should be 9 nodes') self.assertNotIn('ietf-network-topology:link', response['network'][0], @@ -406,7 +406,7 @@ class TransportPCEtesting(unittest.TestCase): # test service-create for OCH-OTU4 service from spdrA to spdrB def test_025_create_OCH_OTU4_service_AB(self): - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -415,7 +415,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_026_get_OCH_OTU4_service_AB(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service-OCH-OTU4-AB") + response = test_utils.get_ordm_serv_list_attr_request("services", "service-OCH-OTU4-AB") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service-OCH-OTU4-AB') @@ -424,7 +424,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_027_check_otn_topo_otu4_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 2) listLinkId = ['OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SB1-XPDR2-XPDR2-NETWORK1', @@ -457,7 +457,7 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR1-NETWORK1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] = "SPDR-SC1-XPDR1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-NETWORK1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -466,7 +466,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_029_get_OCH_OTU4_service_BC(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service-OCH-OTU4-BC") + response = test_utils.get_ordm_serv_list_attr_request("services", "service-OCH-OTU4-BC") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service-OCH-OTU4-BC') @@ -475,7 +475,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_030_check_otn_topo_otu4_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 4) listLinkId = ['OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SB1-XPDR2-XPDR2-NETWORK1', @@ -498,6 +498,7 @@ class TransportPCEtesting(unittest.TestCase): # test service-create for ODU4 service from spdrA to spdrC via spdrB + def test_031_create_ODU4_service(self): self.cr_serv_input_data["service-name"] = "service-ODU4-ABC" self.cr_serv_input_data["service-a-end"]["node-id"] = "SPDR-SA1" @@ -519,7 +520,7 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] = "SPDR-SC1-XPDR1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-NETWORK1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -528,7 +529,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_032_get_ODU4_service_ABC(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service-ODU4-ABC") + response = test_utils.get_ordm_serv_list_attr_request("services", "service-ODU4-ABC") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service-ODU4-ABC') @@ -537,7 +538,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_033_check_interface_ODU4_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', @@ -561,7 +562,7 @@ class TransportPCEtesting(unittest.TestCase): {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute2_request( + response2 = test_utils.check_node_attribute2_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU4', 'org-openroadm-otn-odu-interfaces:odu') self.assertEqual(response2['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], response2['org-openroadm-otn-odu-interfaces:odu']['tx-dapi']) @@ -570,7 +571,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(input_dict_2['tx-dapi'], response2['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_034_check_interface_ODU4_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', @@ -593,7 +594,7 @@ class TransportPCEtesting(unittest.TestCase): {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute2_request( + response2 = test_utils.check_node_attribute2_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU4', 'org-openroadm-otn-odu-interfaces:odu') self.assertEqual(response2['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], response2['org-openroadm-otn-odu-interfaces:odu']['tx-dapi']) @@ -602,7 +603,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(input_dict_2['tx-dapi'], response2['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_035_check_interface_ODU4_NETWORK1_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', @@ -623,7 +624,7 @@ class TransportPCEtesting(unittest.TestCase): dict.keys(response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])) def test_036_check_interface_ODU4_NETWORK2_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', @@ -645,7 +646,7 @@ class TransportPCEtesting(unittest.TestCase): dict.keys(response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])) def test_037_check_ODU4_connection_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'odu-connection', 'XPDR2-NETWORK1-ODU4-x-XPDR2-NETWORK2-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { @@ -661,7 +662,7 @@ class TransportPCEtesting(unittest.TestCase): response['odu-connection'][0]['source']) def test_038_check_otn_topo_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 6) for link in response['network'][0]['ietf-network-topology:link']: @@ -686,7 +687,7 @@ class TransportPCEtesting(unittest.TestCase): self.fail("this link should not exist") def test_039_check_otn_topo_tp(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) for node in response['network'][0]['node']: if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1': @@ -703,6 +704,7 @@ class TransportPCEtesting(unittest.TestCase): # test service-create for 10GE service from spdr to spdr + def test_040_create_10GE_service(self): self.cr_serv_input_data["service-name"] = "service1-10GE" self.cr_serv_input_data["connection-type"] = "service" @@ -716,7 +718,7 @@ class TransportPCEtesting(unittest.TestCase): del self.cr_serv_input_data["service-z-end"]["odu-service-rate"] self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -725,7 +727,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_041_get_10GE_service1(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service1-10GE") + response = test_utils.get_ordm_serv_list_attr_request("services", "service1-10GE") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service1-10GE') @@ -734,7 +736,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_042_check_interface_10GE_CLIENT_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-CLIENT1-ETHERNET10G') self.assertEqual(response['status_code'], requests.codes.ok) input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G', @@ -748,7 +750,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(10000, response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']['speed']) def test_043_check_interface_ODU2E_CLIENT_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-CLIENT1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e', @@ -776,7 +778,7 @@ class TransportPCEtesting(unittest.TestCase): {'payload-type': '03', 'exp-payload-type': '03'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-CLIENT1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -789,7 +791,7 @@ class TransportPCEtesting(unittest.TestCase): response2['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_044_check_interface_ODU2E_NETWORK_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e', @@ -818,7 +820,7 @@ class TransportPCEtesting(unittest.TestCase): ['trib-slots']) def test_045_check_ODU2E_connection_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'odu-connection', 'XPDR1-CLIENT1-ODU2e-x-XPDR1-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { @@ -834,7 +836,7 @@ class TransportPCEtesting(unittest.TestCase): response['odu-connection'][0]['source']) def test_046_check_interface_10GE_CLIENT_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-CLIENT1-ETHERNET10G') self.assertEqual(response['status_code'], requests.codes.ok) input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G', @@ -848,7 +850,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(10000, response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']['speed']) def test_047_check_interface_ODU2E_CLIENT_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-CLIENT1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e', @@ -876,7 +878,7 @@ class TransportPCEtesting(unittest.TestCase): {'payload-type': '03', 'exp-payload-type': '03'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-CLIENT1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -889,7 +891,7 @@ class TransportPCEtesting(unittest.TestCase): response2['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_048_check_interface_ODU2E_NETWORK_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e', @@ -918,7 +920,7 @@ class TransportPCEtesting(unittest.TestCase): ['trib-slots']) def test_049_check_ODU2E_connection_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'odu-connection', 'XPDR1-CLIENT1-ODU2e-x-XPDR1-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { @@ -934,7 +936,7 @@ class TransportPCEtesting(unittest.TestCase): response['odu-connection'][0]['source']) def test_050_check_otn_topo_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 6) for link in response['network'][0]['ietf-network-topology:link']: @@ -947,7 +949,7 @@ class TransportPCEtesting(unittest.TestCase): link['org-openroadm-otn-network-topology:used-bandwidth'], 10000) def test_051_check_otn_topo_tp(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) for node in response['network'][0]['node']: if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1': @@ -966,7 +968,7 @@ class TransportPCEtesting(unittest.TestCase): def test_052_delete_10GE_service(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1-10GE" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -975,34 +977,34 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_053_check_service_list(self): - response = test_utils_rfc8040.get_ordm_serv_list_request() + response = test_utils.get_ordm_serv_list_request() self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['service-list']['services']), 3) time.sleep(1) def test_054_check_no_ODU2e_connection_spdra(self): - response = test_utils_rfc8040.check_node_request("SPDR-SA1") + response = test_utils.check_node_request("SPDR-SA1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn(['odu-connection'][0], response['org-openroadm-device']) time.sleep(1) def test_055_check_no_interface_ODU2E_NETWORK_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.conflict) def test_056_check_no_interface_ODU2E_CLIENT_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-CLIENT1-ODU2e') self.assertEqual(response['status_code'], requests.codes.conflict) def test_057_check_no_interface_10GE_CLIENT_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-CLIENT1-ETHERNET10G') self.assertEqual(response['status_code'], requests.codes.conflict) def test_058_check_otn_topo_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 6) for link in response['network'][0]['ietf-network-topology:link']: @@ -1015,7 +1017,7 @@ class TransportPCEtesting(unittest.TestCase): link['org-openroadm-otn-network-topology:used-bandwidth'], 0) def test_059_check_otn_topo_tp(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) for node in response['network'][0]['node']: if (node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1'): @@ -1029,7 +1031,7 @@ class TransportPCEtesting(unittest.TestCase): def test_060_delete_ODU4_service(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service-ODU4-ABC" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1038,32 +1040,32 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_061_check_service_list(self): - response = test_utils_rfc8040.get_ordm_serv_list_request() + response = test_utils.get_ordm_serv_list_request() self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['service-list']['services']), 2) time.sleep(1) def test_062_check_no_interface_ODU4_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.conflict) def test_063_check_no_interface_ODU4_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.conflict) - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU4') self.assertEqual(response['status_code'], requests.codes.conflict) def test_064_check_no_ODU4_connection_spdrb(self): - response = test_utils_rfc8040.check_node_request("SPDR-SB1") + response = test_utils.check_node_request("SPDR-SB1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn(['odu-connection'][0], response['org-openroadm-device']) time.sleep(1) def test_065_check_no_interface_ODU4_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.conflict) @@ -1071,7 +1073,7 @@ class TransportPCEtesting(unittest.TestCase): self.test_030_check_otn_topo_otu4_links() def test_067_check_otn_topo_tp(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) for node in response['network'][0]['node']: if node['node-id'] == 'SPDR-SA1-XPDR1' or node['node-id'] == 'SPDR-SC1-XPDR1': @@ -1104,7 +1106,7 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] = "SPDR-SB1-XPDR2" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR2-NETWORK1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1113,7 +1115,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_069_get_ODU4_service_AB(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service-ODU4-AB") + response = test_utils.get_ordm_serv_list_attr_request("services", "service-ODU4-AB") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service-ODU4-AB') @@ -1122,7 +1124,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_070_check_interface_ODU4_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', @@ -1148,7 +1150,7 @@ class TransportPCEtesting(unittest.TestCase): {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -1161,7 +1163,7 @@ class TransportPCEtesting(unittest.TestCase): response2['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_071_check_interface_ODU4_spdrb_N1(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', @@ -1186,7 +1188,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual( {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -1216,7 +1218,7 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] = "SPDR-SC1-XPDR1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-NETWORK1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1225,7 +1227,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_073_get_ODU4_service_AB(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service-ODU4-BC") + response = test_utils.get_ordm_serv_list_attr_request("services", "service-ODU4-BC") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service-ODU4-BC') @@ -1234,7 +1236,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_074_check_interface_ODU4_spdrb_N2(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', @@ -1259,7 +1261,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual( {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -1272,7 +1274,7 @@ class TransportPCEtesting(unittest.TestCase): response2['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['expected-dapi']) def test_075_check_interface_ODU4_spdrc(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SC1', 'interface', 'XPDR1-NETWORK1-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', @@ -1296,7 +1298,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual( {'payload-type': '21', 'exp-payload-type': '21'}, response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) - response2 = test_utils_rfc8040.check_node_attribute_request( + response2 = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU4') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(input_dict_2['tx-sapi'], @@ -1311,6 +1313,7 @@ class TransportPCEtesting(unittest.TestCase): # test service-create for 10GE service from spdr to spdr + def test_076_create_10GE_service_ABC(self): self.cr_serv_input_data["service-name"] = "service1-10GE" self.cr_serv_input_data["connection-type"] = "service" @@ -1332,7 +1335,7 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] = "SPDR-SC1-XPDR1" self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-create', self.cr_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1351,7 +1354,7 @@ class TransportPCEtesting(unittest.TestCase): self.test_049_check_ODU2E_connection_spdrc() def test_078_check_interface_ODU2E_NETWORK1_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU2e', @@ -1379,7 +1382,7 @@ class TransportPCEtesting(unittest.TestCase): ['trib-slots']) def test_079_check_interface_ODU2E_NETWORK2_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR2-NETWORK2-ODU2e', @@ -1407,7 +1410,7 @@ class TransportPCEtesting(unittest.TestCase): ['trib-slots']) def test_080_check_ODU2E_connection_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'odu-connection', 'XPDR2-NETWORK1-ODU2e-x-XPDR2-NETWORK2-ODU2e') self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { @@ -1424,7 +1427,7 @@ class TransportPCEtesting(unittest.TestCase): response['odu-connection'][0]['source']) def test_081_check_otn_topo_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 8) for link in response['network'][0]['ietf-network-topology:link']: @@ -1443,7 +1446,7 @@ class TransportPCEtesting(unittest.TestCase): def test_083_delete_10GE_service(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1-10GE" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1452,7 +1455,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_084_check_service_list(self): - response = test_utils_rfc8040.get_ordm_serv_list_request() + response = test_utils.get_ordm_serv_list_request() self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['service-list']['services']), 4) time.sleep(1) @@ -1464,23 +1467,23 @@ class TransportPCEtesting(unittest.TestCase): self.test_057_check_no_interface_10GE_CLIENT_spdra() def test_086_check_no_ODU2e_connection_spdrb(self): - response = test_utils_rfc8040.check_node_request("SPDR-SB1") + response = test_utils.check_node_request("SPDR-SB1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn(['odu-connection'][0], response['org-openroadm-device']) time.sleep(1) def test_087_check_no_interface_ODU2E_NETWORK1_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK1-ODU2e') self.assertEqual(response['status_code'], requests.codes.conflict) def test_088_check_no_interface_ODU2E_NETWORK2_spdrb(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SB1', 'interface', 'XPDR2-NETWORK2-ODU2e') self.assertEqual(response['status_code'], requests.codes.conflict) def test_089_check_otn_topo_links(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 8) for link in response['network'][0]['ietf-network-topology:link']: @@ -1499,7 +1502,7 @@ class TransportPCEtesting(unittest.TestCase): def test_091_delete_ODU4_service_AB(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service-ODU4-AB" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1509,7 +1512,7 @@ class TransportPCEtesting(unittest.TestCase): def test_092_delete_ODU4_service_BC(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service-ODU4-BC" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1528,7 +1531,7 @@ class TransportPCEtesting(unittest.TestCase): def test_094_delete_OCH_OTU4_service_AB(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service-OCH-OTU4-AB" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1538,7 +1541,7 @@ class TransportPCEtesting(unittest.TestCase): def test_095_delete_OCH_OTU4_service_BC(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service-OCH-OTU4-BC" - response = test_utils_rfc8040.transportpce_api_rpc_request( + response = test_utils.transportpce_api_rpc_request( 'org-openroadm-service', 'service-delete', self.del_serv_input_data) self.assertEqual(response['status_code'], requests.codes.ok) @@ -1547,7 +1550,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_096_get_no_service(self): - response = test_utils_rfc8040.get_ordm_serv_list_request() + response = test_utils.get_ordm_serv_list_request() self.assertEqual(response['status_code'], requests.codes.conflict) self.assertIn(response['service-list'], ( { @@ -1566,58 +1569,58 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(1) def test_097_check_no_interface_OTU4_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-OTU') self.assertEqual(response['status_code'], requests.codes.conflict) def test_098_check_no_interface_OCH_spdra(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( 'SPDR-SA1', 'interface', 'XPDR1-NETWORK1-1') self.assertEqual(response['status_code'], requests.codes.conflict) def test_099_getLinks_OtnTopology(self): - response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config') + response = test_utils.get_ietf_network_request('otn-topology', 'config') self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn('ietf-network-topology:link', response['network'][0]) def test_100_disconnect_xponders_from_roadm(self): - response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config') + 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'): - response = test_utils_rfc8040.del_ietf_network_link_request( + 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)) def test_101_check_openroadm_topology(self): - response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config') + 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'] self.assertEqual(28, len(links), 'Topology should contain 28 links') def test_102_disconnect_spdrA(self): - response = test_utils_rfc8040.unmount_device("SPDR-SA1") + response = test_utils.unmount_device("SPDR-SA1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_103_disconnect_spdrC(self): - response = test_utils_rfc8040.unmount_device("SPDR-SC1") + response = test_utils.unmount_device("SPDR-SC1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_104_disconnect_spdrB(self): - response = test_utils_rfc8040.unmount_device("SPDR-SB1") + response = test_utils.unmount_device("SPDR-SB1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_105_disconnect_roadmA(self): - response = test_utils_rfc8040.unmount_device("ROADM-A1") + response = test_utils.unmount_device("ROADM-A1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_106_disconnect_roadmB(self): - response = test_utils_rfc8040.unmount_device("ROADM-B1") + response = test_utils.unmount_device("ROADM-B1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_107_disconnect_roadmC(self): - response = test_utils_rfc8040.unmount_device("ROADM-C1") + response = test_utils.unmount_device("ROADM-C1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))