Switch functional tests to RFC8040

[transportpce.git] / tests / transportpce_tests / 7.1 / test01_portmapping.py

diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py
index b297e77cff4970a24e0fdd1f29b8a858ffedf0df..c448bac478903ba2aa7c22b2fe62a9ef99d0dd2a 100644 (file)
--- 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')