Migrate OTN E2E functional tests to RFC8040 step 6

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

diff --git a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py
index 3c1e4767430c80f15b91e45ee6a4b2aa7bbcef56..07a493a5ed83fbf617c253508fd974114f9859d6 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py
+++ b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py
@@ -22,7 +22,6 @@ import sys
 sys.path.append('transportpce_tests/common/')
 # pylint: disable=wrong-import-position
 # pylint: disable=import-error
-import test_utils  # nopep8
 import test_utils_rfc8040  # nopep8
 
 
@@ -811,11 +810,9 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(len(response['service-list']['services']), 2)
 
     def test_43_check_no_ODU2e_connection_spdra(self):
-        response = test_utils.check_netconf_node_request("SPDR-SA1", "")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertNotIn(['odu-connection'][0], res['org-openroadm-device'])
-        time.sleep(1)
+        response = test_utils_rfc8040.check_node_request("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertNotIn(['odu-connection'][0], response['org-openroadm-device'])
 
     def test_44_check_no_interface_ODU2E_NETWORK_spdra(self):
         response = test_utils_rfc8040.check_node_attribute_request(
@@ -1314,11 +1311,9 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(len(response['service-list']['services']), 2)
 
     def test_84_check_no_ODU0_connection_spdra(self):
-        response = test_utils.check_netconf_node_request("SPDR-SA1", "")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertNotIn(['odu-connection'][0], res['org-openroadm-device'])
-        time.sleep(1)
+        response = test_utils_rfc8040.check_node_request("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertNotIn(['odu-connection'][0], response['org-openroadm-device'])
 
     def test_85_check_no_interface_ODU0_NETWORK_spdra(self):
         response = test_utils_rfc8040.check_node_attribute_request(
@@ -1378,15 +1373,14 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(self.WAITING)
 
     def test_92_disconnect_xponders_from_roadm(self):
-        url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
         response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
         self.assertEqual(response['status_code'], requests.codes.ok)
         links = response['network'][0]['ietf-network-topology:link']
         for link in links:
             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
-                link_name = link["link-id"]
-                response = test_utils.delete_request(url+link_name)
-                self.assertEqual(response.status_code, requests.codes.ok)
+                response = test_utils_rfc8040.del_ietf_network_link_request(
+                    'openroadm-topology', link['link-id'], 'config')
+                self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
     def test_93_check_openroadm_topology(self):
         response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index d53ddbf061f4c0e607ece5ec07178cb8bf40df25..e83a284a221260238a92c3cca7afdae5a2efe07f 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -318,6 +318,22 @@ def check_device_connection(node: str):
             'connection-status': connection_status}
 
 
+def check_node_request(node: 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?content=config',  # 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))
+    res = response.json()
+    return_key = {'rfc8040': 'org-openroadm-device:org-openroadm-device',
+                  'draft-bierman02': 'org-openroadm-device'}
+    if return_key[RESTCONF_VERSION] in res.keys():
+        response_attribute = res[return_key[RESTCONF_VERSION]]
+    else:
+        response_attribute = res['errors']['error'][0]
+    return {'status_code': response.status_code,
+            'org-openroadm-device': response_attribute}
+
+
 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/{}={}?content=nonconfig',  # nopep8