X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F7.1%2Ftest01_portmapping.py;fp=tests%2Ftransportpce_tests%2F7.1%2Ftest01_portmapping.py;h=c448bac478903ba2aa7c22b2fe62a9ef99d0dd2a;hb=654a9fb915061ac8eb5ec52cacbc3c76a8455aa8;hp=b297e77cff4970a24e0fdd1f29b8a858ffedf0df;hpb=6233872e577c6abc704af61bc26362202017c103;p=transportpce.git diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py index b297e77cf..c448bac47 100644 --- a/tests/transportpce_tests/7.1/test01_portmapping.py +++ b/tests/transportpce_tests/7.1/test01_portmapping.py @@ -19,7 +19,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 TransportPCE400GPortMappingTesting(unittest.TestCase): @@ -29,14 +29,14 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): @classmethod def setUpClass(cls): - cls.processes = test_utils_rfc8040.start_tpce() - cls.processes = test_utils_rfc8040.start_sims([('xpdra2', cls.NODE_VERSION)]) + cls.processes = test_utils.start_tpce() + cls.processes = test_utils.start_sims([('xpdra2', 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): @@ -45,20 +45,20 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): time.sleep(1) def test_01_xpdr_device_connection(self): - response = test_utils_rfc8040.mount_device("XPDR-A2", - ('xpdra2', self.NODE_VERSION)) + response = test_utils.mount_device("XPDR-A2", + ('xpdra2', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, - test_utils_rfc8040.CODE_SHOULD_BE_201) + test_utils.CODE_SHOULD_BE_201) # Check if the node appears in the ietf-network topology def test_02_xpdr_device_connected(self): - response = test_utils_rfc8040.check_device_connection("XPDR-A2") + response = test_utils.check_device_connection("XPDR-A2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(response['connection-status'], 'connected') # Check node info in the port-mappings def test_03_xpdr_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "node-info", None) + response = test_utils.get_portmapping_node_attr("XPDR-A2", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'xpdr', @@ -71,7 +71,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check the if-capabilities and the other details for network def test_04_tpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-NETWORK1") + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': @@ -90,7 +90,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['mapping']) def test_05_tpdr_portmapping_CLIENT1(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-CLIENT1") + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-CLIENT1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'], @@ -110,7 +110,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check the port-mapping for the switch-client and switch-network port-quals def test_06_mpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': @@ -128,7 +128,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['mapping']) def test_07_mpdr_portmapping_CLIENT1(self): - res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-CLIENT1") + res = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-CLIENT1") self.assertEqual(res['status_code'], requests.codes.ok) self.assertIn('org-openroadm-port-types:if-100GE-ODU4', res['mapping'][0]['supported-interface-capability']) @@ -148,14 +148,14 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Added test to check mc-capability-profile for a transponder def test_08_check_mccapprofile(self): - res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mc-capabilities", "XPDR-mcprofile") + res = test_utils.get_portmapping_node_attr("XPDR-A2", "mc-capabilities", "XPDR-mcprofile") self.assertEqual(res['status_code'], requests.codes.ok) self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile') self.assertEqual(float(res['mc-capabilities'][0]['center-freq-granularity']), 3.125) self.assertEqual(float(res['mc-capabilities'][0]['slot-width-granularity']), 6.25) def test_09_mpdr_switching_pool(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "switching-pool-lcp", "1") + response = test_utils.get_portmapping_node_attr("XPDR-A2", "switching-pool-lcp", "1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual("blocking", response['switching-pool-lcp'][0]['switching-pool-type']) @@ -168,11 +168,11 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['switching-pool-lcp'][0]['non-blocking-list']) def test_10_xpdr_device_disconnection(self): - response = test_utils_rfc8040.unmount_device("XPDR-A2") + response = test_utils.unmount_device("XPDR-A2") self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_11_xpdr_device_disconnected(self): - response = test_utils_rfc8040.check_device_connection("XPDR-A2") + response = test_utils.check_device_connection("XPDR-A2") self.assertEqual(response['status_code'], requests.codes.conflict) self.assertIn(response['connection-status']['error-type'], ('protocol', 'application')) self.assertEqual(response['connection-status']['error-tag'], 'data-missing') @@ -180,7 +180,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_12_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "node-info", None) + response = test_utils.get_portmapping_node_attr("XPDR-A2", "node-info", None) self.assertEqual(response['status_code'], requests.codes.conflict) self.assertIn(response['node-info']['error-type'], ('protocol', 'application')) self.assertEqual(response['node-info']['error-tag'], 'data-missing')