Refactor test_utils_rfc8040.py portmapping methods

[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 a7bf2b9490e898c02227e17223366f6123c4bf49..1f3f7791f3b1e548b9438ed997096f1b0dba3988 100644 (file)
--- a/tests/transportpce_tests/7.1/test01_portmapping.py
+++ b/tests/transportpce_tests/7.1/test01_portmapping.py
@@ -58,7 +58,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # Check node info in the port-mappings
     def test_03_xpdr_portmapping_info(self):
-        response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
+        response = test_utils_rfc8040.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.portmapping_request("XPDR-A2", "XPDR1-NETWORK1")
+        response = test_utils_rfc8040.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.portmapping_request("XPDR-A2", "XPDR1-CLIENT1")
+        response = test_utils_rfc8040.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.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        response = test_utils_rfc8040.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.portmapping_request("XPDR-A2", "XPDR2-CLIENT1")
+        res = test_utils_rfc8040.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'])
@@ -150,24 +150,24 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # Added test to check mc-capability-profile for a transponder
     def test_08_check_mccapprofile(self):
-        res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "XPDR-mcprofile")
+        res = test_utils_rfc8040.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(str(res['mc-capabilities'][0]['center-freq-granularity']), '3.125')
         self.assertEqual(str(res['mc-capabilities'][0]['slot-width-granularity']), '6.25')
 
     def test_09_mpdr_switching_pool(self):
-        response = test_utils_rfc8040.portmapping_switching_pool_request("XPDR-A2", "1")
+        response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "switching-pool-lcp", "1")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual("blocking",
-                         response['switching_pool'][0]['switching-pool-type'])
+                         response['switching-pool-lcp'][0]['switching-pool-type'])
         self.assertEqual(2,
-                         len(response['switching_pool'][0]['non-blocking-list']))
+                         len(response['switching-pool-lcp'][0]['non-blocking-list']))
         self.assertIn(
             {'nbl-number': 2,
              'interconnect-bandwidth': 0,
              'lcp-list': ['XPDR2-NETWORK1', 'XPDR2-CLIENT2']},
-            response['switching_pool'][0]['non-blocking-list'])
+            response['switching-pool-lcp'][0]['non-blocking-list'])
 
     def test_10_xpdr_device_disconnection(self):
         response = test_utils_rfc8040.unmount_device("XPDR-A2")
@@ -182,7 +182,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_info("XPDR-A2")
+        response = test_utils_rfc8040.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')