Refactor func. tests RFC8040 yang mount utils

JIRA: TRNSPRTPCE-567
Signed-off-by: guillaume.lambert <guillaume.lambert@orange.com>
Change-Id: Ic22d3bf47141f23a015f56ac4c5dca8b5fa2bdc7

diff --git a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py
index 7b7bf35d80a9a34afe6be2f46564f3d83e2fee2b..9fd665d0269eaca2cd20e92b78e3ccdce2c1e227 100644 (file)
--- a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py
+++ b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py
@@ -107,7 +107,7 @@ class TransportPCERendererTesting(unittest.TestCase):
         self.assertEqual(response.status_code, requests.codes.ok)
 
     def test_06_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADMA01", "DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -123,7 +123,7 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 
     def test_07_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADMA01", "SRG1-PP7-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -139,7 +139,8 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 
     def test_08_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_roadm_connections_request("ROADMA01", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -152,7 +153,7 @@ class TransportPCERendererTesting(unittest.TestCase):
         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-713:720'}, response['roadm-connections'][0]['destination'])
 
     def test_09_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -175,7 +176,7 @@ class TransportPCERendererTesting(unittest.TestCase):
               'modulation-format': 'dp-qpsk'}])
 
     def test_10_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-OTU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -192,7 +193,7 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 
     def test_11_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-ODU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -215,7 +216,7 @@ class TransportPCERendererTesting(unittest.TestCase):
                              response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 
     def test_12_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-CLIENT1-ETHERNET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -235,7 +236,7 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 
     def test_13_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDRA01", "1%2F0%2F1-PLUG-NET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]["equipment-state"])
@@ -264,35 +265,36 @@ class TransportPCERendererTesting(unittest.TestCase):
                        {'transportpce-device-renderer:output': {'result': 'Request processed', 'success': True}}))
 
     def test_15_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADMA01", "DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_16_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADMA01", "SRG1-PP7-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_17_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_roadm_connections_request("ROADMA01", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_18_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_19_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-OTU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_20_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-NETWORK1-ODU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_21_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDRA01", "XPDR1-CLIENT1-ETHERNET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_22_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDRA01", "1%2F0%2F1-PLUG-NET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 

diff --git a/tests/transportpce_tests/2.2.1/test05_flex_grid.py b/tests/transportpce_tests/2.2.1/test05_flex_grid.py
index c80259de9111c7f655907e00f48bbdabd09e6dc2..ef3faf8dd640db215008357282a54bf2cd5abcf8 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test05_flex_grid.py
+++ b/tests/transportpce_tests/2.2.1/test05_flex_grid.py
@@ -123,7 +123,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     # Get Degree MC interface and check
     def test_08_degree_mc_interface(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "DEG1-TTP-TXRX-mc-749:763")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-mc-749:763")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertDictEqual(
             dict({"name": "DEG1-TTP-TXRX-mc-749:763",
@@ -143,7 +143,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     # get DEG-NMC interface and check
     def test_09_degree_nmc_interface(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "DEG1-TTP-TXRX-nmc-749:763")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-nmc-749:763")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertDictEqual(
             dict({"name": "DEG1-TTP-TXRX-nmc-749:763",
@@ -163,7 +163,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     # get SRG-NMC interface
     def test_10_srg_nmc_interface(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "SRG1-PP1-TXRX-nmc-749:763")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-D1", "interface", "SRG1-PP1-TXRX-nmc-749:763")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(
             dict({"name": "SRG1-PP1-TXRX-nmc-749:763",
@@ -180,7 +180,8 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     # Create ROADM-connection
     def test_11_roadm_connection(self):
-        response = test_utils_rfc8040.check_roadm_connections_request("ROADM-D1", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADM-D1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual("SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763",
                          response['roadm-connections'][0]['connection-name'])
@@ -194,25 +195,26 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     # delete ROADM connection
     def test_12_delete_roadm_connection(self):
-        response = test_utils_rfc8040.del_roadm_connections_request("ROADM-D1", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
+        response = test_utils_rfc8040.del_node_attribute_request(
+            "ROADM-D1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
         time.sleep(3)
 
     # Delete NMC SRG interface
     def test_13_delete_srg_interface(self):
-        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "SRG1-PP1-TXRX-nmc-749:763")
+        response = test_utils_rfc8040.del_node_attribute_request("ROADM-D1", "interface", "SRG1-PP1-TXRX-nmc-749:763")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
         time.sleep(3)
 
     # Delete NMC Degree interface
     def test_14_delete_degree_interface(self):
-        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "DEG1-TTP-TXRX-nmc-749:763")
+        response = test_utils_rfc8040.del_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-nmc-749:763")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
         time.sleep(3)
 
     # Delete MC Degree interface
     def test_15_delete_degree_interface(self):
-        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "DEG1-TTP-TXRX-mc-749:763")
+        response = test_utils_rfc8040.del_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-mc-749:763")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
         time.sleep(3)
 

diff --git a/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py b/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py
index 33632fe3faf55a43d09abe394f768034a16f82bf..ef6645103d68c717f2aa2515295ebfe5581c3dc3 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py
+++ b/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py
@@ -104,7 +104,7 @@ class TransportPCERendererTesting(unittest.TestCase):
         self.assertEqual(response.status_code, requests.codes.ok)
 
     def test_06_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "DEG1-TTP-TXRX-nmc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -121,7 +121,7 @@ class TransportPCERendererTesting(unittest.TestCase):
                        {'frequency': 195.8, 'width': 40}])
 
     def test_07_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "DEG1-TTP-TXRX-mc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -138,7 +138,7 @@ class TransportPCERendererTesting(unittest.TestCase):
                        {'min-freq': 195.775, 'max-freq': 195.825}])
 
     def test_08_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "SRG1-PP3-TXRX-nmc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -156,11 +156,12 @@ class TransportPCERendererTesting(unittest.TestCase):
 
     # -mc supporting interfaces must not be created for SRG, only degrees
     def test_09_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "SRG1-PP3-TXRX-mc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_10_service_path_create_rdm_check(self):
-        response = test_utils_rfc8040.check_roadm_connections_request("ROADM-A1", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -172,7 +173,7 @@ class TransportPCERendererTesting(unittest.TestCase):
         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-713:720'}, response['roadm-connections'][0]['destination'])
 
     def test_11_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -192,7 +193,7 @@ class TransportPCERendererTesting(unittest.TestCase):
               'modulation-format': 'dp-qpsk', 'frequency': 195.8}])
 
     def test_12_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-OTU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-OTU")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -210,7 +211,7 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 
     def test_13_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-ODU4")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -230,7 +231,7 @@ class TransportPCERendererTesting(unittest.TestCase):
                              response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 
     def test_14_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-CLIENT1-ETHERNET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-CLIENT1-ETHERNET")
         self.assertEqual(response['status_code'], requests.codes.ok)
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
         self.assertDictEqual(
@@ -248,13 +249,13 @@ class TransportPCERendererTesting(unittest.TestCase):
             response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 
     def test_15_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDR-A1", "1%2F0%2F1-PLUG-NET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 
     def test_16_service_path_create_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDR-A1", "1%2F0%2F1-PLUG-CLIENT")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
@@ -283,52 +284,54 @@ class TransportPCERendererTesting(unittest.TestCase):
                        {'transportpce-device-renderer:output': {'result': 'Request processed', 'success': True}}))
 
     def test_18_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "DEG1-TTP-TXRX-mc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_19_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "DEG1-TTP-TXRX-nmc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_20_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "SRG1-PP3-TXRX-mc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_21_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "SRG1-PP3-TXRX-nmc-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_22_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_interface_request("ROADM-A1", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADM-A1", "interface", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_23_service_path_delete_rdm_check(self):
-        response = test_utils_rfc8040.check_roadm_connections_request("ROADM-A1", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request(
+            "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_24_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-713:720")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_25_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-OTU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-OTU")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_26_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-NETWORK1-ODU")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-ODU")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_27_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_interface_request("XPDR-A1", "XPDR1-CLIENT1-ETHERNET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-CLIENT1-ETHERNET")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_28_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDR-A1", "1%2F0%2F1-PLUG-NET")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 
     def test_29_service_path_delete_xpdr_check(self):
-        response = test_utils_rfc8040.check_circuit_packs_request("XPDR-A1", "1%2F0%2F1-PLUG-CLIENT")
+        response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index b2cfc4c4c75599fb626f9699dd289ca9a4ce7feb..bf4fb6d66ff44ec5194a2fee5943cc7d215e0432 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -318,79 +318,27 @@ def check_device_connection(node: str):
             'connection-status': connection_status}
 
 
-def check_interface_request(node: str, interface: str):
+def check_node_attribute_request(node: str, attribute: str, attribute_value: str):
     # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface={}',  # nopep8
-           'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/{}'}  # nopep8
-    response = get_request(url[RESTCONF_VERSION].format('{}', node, interface))
+    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}',  # nopep8
+           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'}  # nopep8
+    response = get_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value))
     res = response.json()
-    return_key = {'rfc8040': 'org-openroadm-device:interface',
-                  'draft-bierman02': 'interface'}
+    return_key = {'rfc8040': 'org-openroadm-device:' + attribute,
+                  'draft-bierman02': attribute}
     if return_key[RESTCONF_VERSION] in res.keys():
-        interface = res[return_key[RESTCONF_VERSION]]
+        response_attribute = res[return_key[RESTCONF_VERSION]]
     else:
-        interface = res['errors']['error'][0]
+        response_attribute = res['errors']['error'][0]
     return {'status_code': response.status_code,
-            'interface': interface}
+            attribute: response_attribute}
 
 
-def del_interface_request(node: str, interface: str):
+def del_node_attribute_request(node: str, attribute: str, attribute_value: str):
     # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface={}',  # nopep8
-           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/{}'}  # nopep8
-    # draft-bierman02: note this is config here and not operational as previously in check_interface_request
-    response = delete_request(url[RESTCONF_VERSION].format('{}', node, interface))
-    return response
-
-
-def check_roadm_connections_request(node: str, connections: str):
-    # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections={}',  # nopep8
-           'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections/{}'}  # nopep8
-    response = get_request(url[RESTCONF_VERSION].format('{}', node, connections))
-    res = response.json()
-    return_key = {'rfc8040': 'org-openroadm-device:roadm-connections',
-                  'draft-bierman02': 'roadm-connections'}
-    if return_key[RESTCONF_VERSION] in res.keys():
-        roadm_connections = res[return_key[RESTCONF_VERSION]]
-    else:
-        roadm_connections = res['errors']['error'][0]
-    return {'status_code': response.status_code,
-            'roadm-connections': roadm_connections}
-
-
-def del_roadm_connections_request(node: str, connections: str):
-    # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections={}',  # nopep8
-           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections/{}'}  # nopep8
-    # draft-bierman02: note this is config here and not operational as previously in check_interface_request
-    response = delete_request(url[RESTCONF_VERSION].format('{}', node, connections))
-    return response
-
-
-def check_circuit_packs_request(node: str, circuit_packs: str):
-    # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs={}',  # nopep8
-           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs/{}'}  # nopep8
-    # draft-bierman02: note this is config here and not operational as previously in check_interface_request
-    # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
-    response = get_request(url[RESTCONF_VERSION].format('{}', node, circuit_packs))
-    res = response.json()
-    return_key = {'rfc8040': 'org-openroadm-device:circuit-packs',
-                  'draft-bierman02': 'circuit-packs'}
-    if return_key[RESTCONF_VERSION] in res.keys():
-        cp = res[return_key[RESTCONF_VERSION]]
-    else:
-        cp = res['errors']['error'][0]
-    return {'status_code': response.status_code,
-            'circuit-packs': cp}
-
-
-def del_circuit_packs_request(node: str, circuit_packs: str):
-    # pylint: disable=line-too-long
-    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs={}',  # nopep8
-           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs/{}'}  # nopep8
-    response = delete_request(url[RESTCONF_VERSION].format('{}', node, circuit_packs))
+    url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}',  # nopep8
+           'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'}  # nopep8
+    response = delete_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value))
     return response
 
 #