Migrate OTN E2E functional tests to RFC8040 step 2

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

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 e6649a1fc2f266e5247510ea224ecb16b52e80c3..2dfe2448ebdf349c26f08ca7c8d02a5d3fdc1cab 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py
+++ b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py
@@ -351,25 +351,25 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_19_check_openroadm_topo_spdra(self):
-        response = test_utils.get_ordm_topo_request("node/SPDR-SA1-XPDR1")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        ele = res['node'][0]['ietf-network-topology:termination-point'][0]
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'SPDR-SA1-XPDR1', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        ele = response['node']['ietf-network-topology:termination-point'][0]
         self.assertEqual('XPDR1-NETWORK1', ele['tp-id'])
-        self.assertEqual({'frequency': 196.1,
-                          'width': 40},
-                         ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
-        time.sleep(3)
+        self.assertEqual(
+            196.1,
+            float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency']))
+        self.assertEqual(
+            40,
+            float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width']))
 
     def test_20_check_openroadm_topo_ROADMA_SRG(self):
-        response = test_utils.get_ordm_topo_request("node/ROADM-A1-SRG1")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
         freq_map = base64.b64decode(
-            res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
+            response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
         freq_map_array = [int(x) for x in freq_map]
         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
-        liste_tp = res['node'][0]['ietf-network-topology:termination-point']
+        liste_tp = response['node']['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'SRG1-PP1-TXRX':
                 freq_map = base64.b64decode(
@@ -378,17 +378,15 @@ class TransportPCEtesting(unittest.TestCase):
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
             if ele['tp-id'] == 'SRG1-PP2-TXRX':
                 self.assertNotIn('avail-freq-maps', dict.keys(ele))
-        time.sleep(3)
 
     def test_21_check_openroadm_topo_ROADMA_DEG(self):
-        response = test_utils.get_ordm_topo_request("node/ROADM-A1-DEG2")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
         freq_map = base64.b64decode(
-            res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
+            response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
         freq_map_array = [int(x) for x in freq_map]
         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
-        liste_tp = res['node'][0]['ietf-network-topology:termination-point']
+        liste_tp = response['node']['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'DEG2-CTP-TXRX':
                 freq_map = base64.b64decode(
@@ -400,7 +398,6 @@ class TransportPCEtesting(unittest.TestCase):
                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
-        time.sleep(3)
 
     def test_22_check_otn_topo_otu4_links(self):
         response = test_utils.get_otn_topo_request()
@@ -943,24 +940,21 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertNotIn('ietf-network-topology:link', res['network'][0])
 
     def test_59_check_openroadm_topo_spdra(self):
-        response = test_utils.get_ordm_topo_request("node/SPDR-SA1-XPDR1")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        tp = res['node'][0]['ietf-network-topology:termination-point'][0]
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'SPDR-SA1-XPDR1', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        tp = response['node']['ietf-network-topology:termination-point'][0]
         self.assertEqual('XPDR1-NETWORK1', tp['tp-id'])
         self.assertNotIn('wavelength', dict.keys(
             tp['org-openroadm-network-topology:xpdr-network-attributes']))
-        time.sleep(3)
 
     def test_60_check_openroadm_topo_ROADMA_SRG(self):
-        response = test_utils.get_ordm_topo_request("node/ROADM-A1-SRG1")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
         freq_map = base64.b64decode(
-            res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
+            response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
         freq_map_array = [int(x) for x in freq_map]
         self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
-        liste_tp = res['node'][0]['ietf-network-topology:termination-point']
+        liste_tp = response['node']['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'SRG1-PP1-TXRX':
                 freq_map = base64.b64decode(
@@ -970,14 +964,13 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(3)
 
     def test_61_check_openroadm_topo_ROADMA_DEG(self):
-        response = test_utils.get_ordm_topo_request("node/ROADM-A1-DEG2")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
         freq_map = base64.b64decode(
-            res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
+            response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
         freq_map_array = [int(x) for x in freq_map]
         self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
-        liste_tp = res['node'][0]['ietf-network-topology:termination-point']
+        liste_tp = response['node']['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'DEG2-CTP-TXRX':
                 freq_map = base64.b64decode(
@@ -1421,10 +1414,9 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_92_disconnect_xponders_from_roadm(self):
         url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
-        response = test_utils.get_ordm_topo_request("")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        links = res['network'][0]['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"]
@@ -1432,11 +1424,11 @@ class TransportPCEtesting(unittest.TestCase):
                 self.assertEqual(response.status_code, requests.codes.ok)
 
     def test_93_check_openroadm_topology(self):
-        response = test_utils.get_ordm_topo_request("")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        links = res['network'][0]['ietf-network-topology:link']
-        self.assertEqual(18, len(links), 'Topology should contain 18 links')
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(18,
+                         len(response['network'][0]['ietf-network-topology:link']),
+                         'Topology should contain 18 links')
 
     def test_94_disconnect_spdrA(self):
         response = test_utils_rfc8040.unmount_device("SPDR-SA1")

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index 785f2a6d8f5c329c305921d8ba07f54838c35813..ec2758682c175f5dfcc340447d13e4f1ac7164cd 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -494,6 +494,24 @@ def del_oms_attr_request(link: str,):
     return response
 
 
+def get_ietf_network_node_request(network: str, node: str, content: str):
+    url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
+           'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
+    if RESTCONF_VERSION == 'rfc8040':
+        format_args = ('{}', network, node, content)
+    elif content == 'config':
+        format_args = ('{}', content, network, node)
+    else:
+        format_args = ('{}', 'operational', network, node)
+    response = get_request(url[RESTCONF_VERSION].format(*format_args))
+    res = response.json()
+    return_key = {'rfc8040': 'ietf-network:node',
+                  'draft-bierman02': 'node'}
+    node = res[return_key[RESTCONF_VERSION]][0]
+    return {'status_code': response.status_code,
+            'node': node}
+
+
 def del_ietf_network_node_request(network: str, node: str, content: str):
     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}