X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest12_end2end.py;fp=tests%2Ftransportpce_tests%2F2.2.1%2Ftest12_end2end.py;h=c60c2d2a8c0d6f43800d10e4381ebb0c628447ca;hb=654a9fb915061ac8eb5ec52cacbc3c76a8455aa8;hp=8d047bb58cdc0dc2c4dc697e645043e5e8b978b7;hpb=6233872e577c6abc704af61bc26362202017c103;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test12_end2end.py b/tests/transportpce_tests/2.2.1/test12_end2end.py index 8d047bb58..c60c2d2a8 100644 --- a/tests/transportpce_tests/2.2.1/test12_end2end.py +++ b/tests/transportpce_tests/2.2.1/test12_end2end.py @@ -20,7 +20,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 TransportPCEFulltesting(unittest.TestCase): @@ -134,17 +134,17 @@ class TransportPCEFulltesting(unittest.TestCase): @classmethod def setUpClass(cls): - cls.processes = test_utils_rfc8040.start_tpce() - cls.processes = test_utils_rfc8040.start_sims([('xpdra', cls.NODE_VERSION), - ('roadma', cls.NODE_VERSION), - ('roadmc', cls.NODE_VERSION), - ('xpdrc', cls.NODE_VERSION)]) + cls.processes = test_utils.start_tpce() + cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION), + ('roadma', cls.NODE_VERSION), + ('roadmc', cls.NODE_VERSION), + ('xpdrc', 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): # instruction executed before each test method @@ -152,23 +152,23 @@ class TransportPCEFulltesting(unittest.TestCase): print("execution of {}".format(self.id().split(".")[-1])) def test_01_connect_xpdrA(self): - response = test_utils_rfc8040.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION)) - self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION)) + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_02_connect_xpdrC(self): - response = test_utils_rfc8040.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION)) - self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION)) + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_03_connect_rdmA(self): - response = test_utils_rfc8040.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION)) - self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION)) + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_04_connect_rdmC(self): - response = test_utils_rfc8040.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) - self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) + response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_05_connect_xprdA_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': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -177,7 +177,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_06_connect_roadmA_PP1_to_xpdrA_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': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -186,7 +186,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_07_connect_xprdC_xpdr1_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': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -195,7 +195,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_08_connect_roadmC_PP1_to_xpdrC_xprd1_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': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) @@ -204,7 +204,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_09_connect_xprdA_N2_to_roadmA_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': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -213,7 +213,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_10_connect_roadmA_PP2_to_xpdrA_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': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -222,7 +222,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_11_connect_xprdC_xpdr2_N1_to_roadmC_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': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -231,7 +231,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_12_connect_roadmC_PP2_to_xpdrC_xpdr2_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': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1', 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) @@ -251,8 +251,8 @@ class TransportPCEFulltesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) def test_14_add_omsAttributes_ROADMC_ROADMA(self): @@ -267,14 +267,14 @@ class TransportPCEFulltesting(unittest.TestCase): "fiber-type": "smf", "SRLG-length": 100000, "pmd": 0.5}]}} - response = test_utils_rfc8040.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", - data) + response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", + data) self.assertEqual(response.status_code, requests.codes.created) # test service-create for Eth service from xpdr to xpdr def test_15_create_eth_service2(self): self.cr_serv_input_data["service-name"] = "service2" - 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) @@ -283,7 +283,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_16_get_eth_service2(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service2") + response = test_utils.get_ordm_serv_list_attr_request("services", "service2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service2') @@ -292,7 +292,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_17_check_xc1_ROADMA(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -307,7 +307,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_18_check_xc1_ROADMC(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-C1", "roadm-connections", "SRG1-PP2-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -322,7 +322,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_19_check_topo_XPDRA(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: @@ -340,7 +340,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_20_check_topo_ROADMA_SRG1(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) @@ -358,7 +358,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_21_check_topo_ROADMA_DEG2(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) @@ -388,7 +388,7 @@ class TransportPCEFulltesting(unittest.TestCase): del self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"] del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"] - 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) @@ -397,7 +397,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_23_get_eth_service1(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service1") + response = test_utils.get_ordm_serv_list_attr_request("services", "service1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service1') @@ -406,7 +406,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_24_check_xc1_ROADMA(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-A1", "roadm-connections", "DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -420,7 +420,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_25_check_topo_XPDRA(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: @@ -436,7 +436,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_26_check_topo_ROADMA_SRG1(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) @@ -462,7 +462,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_27_check_topo_ROADMA_DEG2(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) @@ -488,7 +488,7 @@ class TransportPCEFulltesting(unittest.TestCase): # creation service test on a non-available resource def test_28_create_eth_service3(self): self.cr_serv_input_data["service-name"] = "service3" - 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) @@ -500,7 +500,7 @@ class TransportPCEFulltesting(unittest.TestCase): # add a test that check the openroadm-service-list still only contains 2 elements def test_29_delete_eth_service3(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service3" - 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) @@ -511,7 +511,7 @@ class TransportPCEFulltesting(unittest.TestCase): def test_30_delete_eth_service1(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1" - 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) @@ -521,7 +521,7 @@ class TransportPCEFulltesting(unittest.TestCase): def test_31_delete_eth_service2(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2" - 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) @@ -530,14 +530,14 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_32_check_no_xc_ROADMA(self): - response = test_utils_rfc8040.check_node_request("ROADM-A1") + response = test_utils.check_node_request("ROADM-A1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn('roadm-connections', dict.keys(response['org-openroadm-device'])) time.sleep(2) def test_33_check_topo_XPDRA(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: @@ -551,7 +551,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_34_check_topo_ROADMA_SRG1(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) @@ -577,7 +577,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_35_check_topo_ROADMA_DEG2(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) @@ -608,7 +608,7 @@ class TransportPCEFulltesting(unittest.TestCase): self.cr_serv_input_data["service-a-end"]["service-format"] = "OC" self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1" self.cr_serv_input_data["service-z-end"]["service-format"] = "OC" - 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) @@ -617,7 +617,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_37_get_oc_service1(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service1") + response = test_utils.get_ordm_serv_list_attr_request("services", "service1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service1') @@ -626,7 +626,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_38_check_xc1_ROADMA(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -641,7 +641,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_39_check_xc1_ROADMC(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-C1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -662,7 +662,7 @@ class TransportPCEFulltesting(unittest.TestCase): self.cr_serv_input_data["service-a-end"]["service-format"] = "OC" self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1" self.cr_serv_input_data["service-z-end"]["service-format"] = "OC" - 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) @@ -671,7 +671,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_41_get_oc_service2(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service2") + response = test_utils.get_ordm_serv_list_attr_request("services", "service2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['services'][0]['administrative-state'], 'inService') self.assertEqual(response['services'][0]['service-name'], 'service2') @@ -680,7 +680,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_42_check_xc2_ROADMA(self): - response = test_utils_rfc8040.check_node_attribute_request( + response = test_utils.check_node_attribute_request( "ROADM-A1", "roadm-connections", "SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760") self.assertEqual(response['status_code'], requests.codes.ok) # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -695,7 +695,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_43_check_topo_ROADMA(self): - response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) @@ -724,7 +724,7 @@ class TransportPCEFulltesting(unittest.TestCase): def test_44_delete_oc_service1(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1" - 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) @@ -734,7 +734,7 @@ class TransportPCEFulltesting(unittest.TestCase): def test_45_delete_oc_service2(self): self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2" - 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) @@ -743,7 +743,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(self.WAITING) def test_46_get_no_oc_services(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'], ( { @@ -762,7 +762,7 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_47_get_no_xc_ROADMA(self): - response = test_utils_rfc8040.check_node_request("ROADM-A1") + response = test_utils.check_node_request("ROADM-A1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertNotIn('roadm-connections', dict.keys(response['org-openroadm-device'])) time.sleep(1) @@ -785,10 +785,10 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_44_delete_oc_service1() def test_50_loop_create_eth_service(self): - response = test_utils_rfc8040.get_ordm_serv_list_attr_request("services", "service1") + response = test_utils.get_ordm_serv_list_attr_request("services", "service1") if response['status_code'] != 404: self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1" - 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) time.sleep(5) @@ -808,19 +808,19 @@ class TransportPCEFulltesting(unittest.TestCase): self.test_31_delete_eth_service2() def test_51_disconnect_XPDRA(self): - response = test_utils_rfc8040.unmount_device("XPDR-A1") + response = test_utils.unmount_device("XPDR-A1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_52_disconnect_XPDRC(self): - response = test_utils_rfc8040.unmount_device("XPDR-C1") + response = test_utils.unmount_device("XPDR-C1") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_53_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_54_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))