add a method to check nodes configs in func tests

[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_otn_renderer.py

diff --git a/tests/transportpce_tests/2.2.1/test_otn_renderer.py b/tests/transportpce_tests/2.2.1/test_otn_renderer.py
index 8a3b86e4110d07e20d3140d065c62b04b0c1eb83..6e41792439f96c157e8645236dbfcd0f6f907118 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test_otn_renderer.py
+++ b/tests/transportpce_tests/2.2.1/test_otn_renderer.py
@@ -23,7 +23,7 @@ class TransportPCEtesting(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.processes = test_utils.start_tpce()
-        cls.processes = test_utils.start_sims(['spdrav2'])
+        cls.processes = test_utils.start_sims(['spdra'])
 
     @classmethod
     def tearDownClass(cls):
@@ -35,15 +35,12 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(5)
 
     def test_01_connect_SPDR_SA1(self):
-        response = test_utils.mount_device("SPDR-SA1", 'spdrav2')
+        response = test_utils.mount_device("SPDR-SA1", 'spdra')
         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
         time.sleep(10)
 
-        url = ("{}/operational/network-topology:"
-               "network-topology/topology/topology-netconf/node/SPDR-SA1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/SPDR-SA1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -51,11 +48,8 @@ class TransportPCEtesting(unittest.TestCase):
             'connected')
 
     def test_02_get_portmapping_CLIENT1(self):
-        url = ("{}/config/transportpce-portmapping:network/"
-               "nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn(
@@ -72,11 +66,8 @@ class TransportPCEtesting(unittest.TestCase):
             res['mapping'])
 
     def test_03_get_portmapping_NETWORK1(self):
-        url = ("{}/config/transportpce-portmapping:network/"
-               "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn(
@@ -93,7 +84,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['mapping'])
 
     def test_04_service_path_create_OCH_OTU4(self):
-        url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:service-path"
         data = {"renderer:input": {
             "service-name": "service_ODU4",
             "wave-number": "1",
@@ -102,9 +93,7 @@ class TransportPCEtesting(unittest.TestCase):
             "nodes": [
                 {"node-id": "SPDR-SA1",
                  "dest-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -116,11 +105,8 @@ class TransportPCEtesting(unittest.TestCase):
              'och-interface-id': ['XPDR1-NETWORK1-1']}, res["output"]['node-interface'])
 
     def test_05_get_portmapping_NETWORK1(self):
-        url = ("{}/config/transportpce-portmapping:network/"
-               "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn(
@@ -137,12 +123,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['mapping'])
 
     def test_06_check_interface_och(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-1")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
 
@@ -160,12 +141,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 
     def test_07_check_interface_OTU(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-OTU"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         input_dict_1 = {'name': 'XPDR1-NETWORK1-OTU',
@@ -191,7 +167,7 @@ class TransportPCEtesting(unittest.TestCase):
                              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 
     def test_08_otn_service_path_create_ODU4(self):
-        url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:otn-service-path"
         data = {"renderer:input": {
             "service-name": "service_ODU4",
             "operation": "create",
@@ -200,9 +176,7 @@ class TransportPCEtesting(unittest.TestCase):
             "nodes": [
                 {"node-id": "SPDR-SA1",
                  "network-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -213,11 +187,8 @@ class TransportPCEtesting(unittest.TestCase):
              'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, res["output"]['node-interface'])
 
     def test_09_get_portmapping_NETWORK1(self):
-        url = ("{}/config/transportpce-portmapping:network/"
-               "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn(
@@ -236,12 +207,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['mapping'])
 
     def test_10_check_interface_ODU4(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-ODU4"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService',
@@ -267,7 +233,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 
     def test_11_otn_service_path_create_10GE(self):
-        url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:otn-service-path"
         data = {"renderer:input": {
             "service-name": "service1",
             "operation": "create",
@@ -280,9 +246,7 @@ class TransportPCEtesting(unittest.TestCase):
                 {"node-id": "SPDR-SA1",
                  "client-tp": "XPDR1-CLIENT1",
                  "network-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -295,12 +259,7 @@ class TransportPCEtesting(unittest.TestCase):
              'eth-interface-id': ['XPDR1-CLIENT1-ETHERNET10G']}, res["output"]['node-interface'])
 
     def test_12_check_interface_10GE_CLIENT(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-CLIENT1-ETHERNET10G"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G',
@@ -316,12 +275,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 
     def test_13_check_interface_ODU2E_CLIENT(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-CLIENT1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
 
@@ -346,12 +300,7 @@ class TransportPCEtesting(unittest.TestCase):
             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 
     def test_14_check_interface_ODU2E_NETWORK(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService',
@@ -382,12 +331,9 @@ class TransportPCEtesting(unittest.TestCase):
                           'parent-odu-allocation']['trib-slots'])
 
     def test_15_check_ODU2E_connection(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request(
+            "SPDR-SA1",
+            "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         input_dict_1 = {
@@ -404,7 +350,7 @@ class TransportPCEtesting(unittest.TestCase):
                              res['odu-connection'][0]['source'])
 
     def test_16_otn_service_path_delete_10GE(self):
-        url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:otn-service-path"
         data = {"renderer:input": {
             "service-name": "service1",
             "operation": "delete",
@@ -417,9 +363,7 @@ class TransportPCEtesting(unittest.TestCase):
                 {"node-id": "SPDR-SA1",
                  "client-tp": "XPDR1-CLIENT1",
                  "network-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -427,43 +371,25 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertTrue(res["output"]["success"])
 
     def test_17_check_no_ODU2E_connection(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request(
+            "SPDR-SA1",
+            "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_18_check_no_interface_ODU2E_NETWORK(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_19_check_no_interface_ODU2E_CLIENT(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-CLIENT1-ODU2e-service1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_20_check_no_interface_10GE_CLIENT(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-CLIENT1-ETHERNET10G"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_21_otn_service_path_delete_ODU4(self):
-        url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:otn-service-path"
         data = {"renderer:input": {
             "service-name": "service_ODU4",
             "operation": "delete",
@@ -472,9 +398,7 @@ class TransportPCEtesting(unittest.TestCase):
             "nodes": [
                 {"node-id": "SPDR-SA1",
                  "network-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -482,16 +406,11 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertTrue(res["output"]["success"])
 
     def test_22_check_no_interface_ODU4(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-ODU4"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_23_service_path_delete_OCH_OTU4(self):
-        url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/transportpce-device-renderer:service-path"
         data = {"renderer:input": {
             "service-name": "service_OTU4",
             "wave-number": "1",
@@ -500,9 +419,7 @@ class TransportPCEtesting(unittest.TestCase):
             "nodes": [
                 {"node-id": "SPDR-SA1",
                  "dest-tp": "XPDR1-NETWORK1"}]}}
-        response = requests.request(
-            "POST", url, data=json.dumps(data),
-            headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         time.sleep(3)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -510,21 +427,11 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertTrue(res["output"]["success"])
 
     def test_24_check_no_interface_OTU4(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-OTU"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_25_check_no_interface_OCH(self):
-        url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
-               "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
-               "interface/XPDR1-NETWORK1-1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-1")
         self.assertEqual(response.status_code, requests.codes.not_found)
 
     def test_26_disconnect_SPDR_SA1(self):