Migrate flex grid func. tests to RFC8040

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

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 dc64616a5d02e405c24105afe3ac2562a47801de..c80259de9111c7f655907e00f48bbdabd09e6dc2 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test05_flex_grid.py
+++ b/tests/transportpce_tests/2.2.1/test05_flex_grid.py
@@ -19,7 +19,7 @@ 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
 
 
 class TransportPCEPortMappingTesting(unittest.TestCase):
@@ -29,14 +29,14 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls.processes = test_utils.start_tpce()
-        cls.processes = test_utils.start_sims([('roadmd', cls.NODE_VERSION)])
+        cls.processes = test_utils_rfc8040.start_tpce()
+        cls.processes = test_utils_rfc8040.start_sims([('roadmd', cls.NODE_VERSION)])
 
     @classmethod
     def tearDownClass(cls):
         # pylint: disable=not-an-iterable
         for process in cls.processes:
-            test_utils.shutdown_process(process)
+            test_utils_rfc8040.shutdown_process(process)
         print("all processes killed")
 
     def setUp(self):
@@ -45,120 +45,86 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         time.sleep(10)
 
     def test_01_rdm_device_connection(self):
-        response = test_utils.mount_device("ROADM-D1", ('roadmd', self.NODE_VERSION))
-        self.assertEqual(response.status_code, requests.codes.created,
-                         test_utils.CODE_SHOULD_BE_201)
+        response = test_utils_rfc8040.mount_device("ROADM-D1", ('roadmd', self.NODE_VERSION))
+        self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
 
     def test_02_rdm_device_connected(self):
-        response = test_utils.get_netconf_oper_request("ROADM-D1")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertEqual(
-            res['node'][0]['netconf-node-topology:connection-status'],
-            'connected')
+        response = test_utils_rfc8040.check_device_connection("ROADM-D1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(response['connection-status'], 'connected')
         time.sleep(10)
 
     def test_03_rdm_portmapping_info(self):
-        response = test_utils.portmapping_request("ROADM-D1/node-info")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.get_portmapping_node_info("ROADM-D1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(
-            {'node-info': {'node-type': 'rdm',
-                           'node-ip-address': '127.0.0.14',
-                           'node-clli': 'NodeD',
-                           'openroadm-version': '2.2.1',
-                           'node-vendor': 'vendorD',
-                           'node-model': 'model2',
-                           }},
-            res)
+            {'node-type': 'rdm',
+             'node-ip-address': '127.0.0.14',
+             'node-clli': 'NodeD',
+             'openroadm-version': '2.2.1',
+             'node-vendor': 'vendorD',
+             'node-model': 'model2'},
+            response['node-info'])
         time.sleep(3)
 
     def test_04_rdm_deg1_lcp(self):
-        response = test_utils.portmapping_request("ROADM-D1/mc-capabilities/DEG1-TTP")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertEqual(
-            {
-                "mc-capabilities": [
-                    {
-                        "mc-node-name": "DEG1-TTP",
-                        "center-freq-granularity": 6.25,
-                        "slot-width-granularity": 12.5
-                    }
-                ]
-            }, res)
+        # pylint: disable=line-too-long
+        response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "DEG1-TTP")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(response['mc-capabilities'],
+                      [[{'mc-node-name': 'DEG1-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
+                       [{'mc-node-name': 'DEG1-TTP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
         time.sleep(3)
 
     def test_05_rdm_deg2_lcp(self):
-        response = test_utils.portmapping_request("ROADM-D1/mc-capabilities/DEG2-TTP")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertEqual(
-            {
-                "mc-capabilities": [
-                    {
-                        "mc-node-name": "DEG2-TTP",
-                        "center-freq-granularity": 6.25,
-                        "slot-width-granularity": 12.5
-                    }
-                ]
-            }, res)
+        # pylint: disable=line-too-long
+        response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "DEG2-TTP")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(response['mc-capabilities'],
+                      [[{'mc-node-name': 'DEG2-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
+                       [{'mc-node-name': 'DEG2-TTP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
         time.sleep(3)
 
     def test_06_rdm_srg1_lcp(self):
-        response = test_utils.portmapping_request("ROADM-D1/mc-capabilities/SRG1-PP")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertEqual(
-            {
-                "mc-capabilities": [
-                    {
-                        "mc-node-name": "SRG1-PP",
-                        "center-freq-granularity": 6.25,
-                        "slot-width-granularity": 12.5
-                    }
-                ]
-            }, res)
+        # pylint: disable=line-too-long
+        response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "SRG1-PP")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(response['mc-capabilities'],
+                      [[{'mc-node-name': 'SRG1-PP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
+                       [{'mc-node-name': 'SRG1-PP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
         time.sleep(3)
 
     # Renderer interface creations
     def test_07_device_renderer(self):
         data = {
-            "transportpce-device-renderer:input": {
-                "transportpce-device-renderer:modulation-format": "dp-qpsk",
-                "transportpce-device-renderer:operation": "create",
-                "transportpce-device-renderer:service-name": "testNMC-MC",
-                "transportpce-device-renderer:wave-number": "0",
-                "transportpce-device-renderer:center-freq": "196.05",
-                "transportpce-device-renderer:nmc-width": "80",
-                "transportpce-device-renderer:nodes": [
+            "input": {
+                "modulation-format": "dp-qpsk",
+                "operation": "create",
+                "service-name": "testNMC-MC",
+                "wave-number": "0",
+                "center-freq": "196.05",
+                "nmc-width": "80",
+                "nodes": [
                     {
-                        "transportpce-device-renderer:node-id": "ROADM-D1",
-                        "transportpce-device-renderer:src-tp": "SRG1-PP1-TXRX",
-                        "transportpce-device-renderer:dest-tp": "DEG1-TTP-TXRX"
+                        "node-id": "ROADM-D1",
+                        "src-tp": "SRG1-PP1-TXRX",
+                        "dest-tp": "DEG1-TTP-TXRX"
                     }
                 ],
-                "transportpce-device-renderer:min-freq": 196.00625,
-                "transportpce-device-renderer:max-freq": 196.09375,
-                "transportpce-device-renderer:lower-spectral-slot-number": 749,
-                "transportpce-device-renderer:higher-spectral-slot-number": 763
+                "min-freq": 196.00625,
+                "max-freq": 196.09375,
+                "lower-spectral-slot-number": 749,
+                "higher-spectral-slot-number": 763
             }
         }
-        url = test_utils.RESTCONF_BASE_URL + \
-            "/operations/transportpce-device-renderer:service-path"
-        response = test_utils.post_request(url, data)
+        response = test_utils_rfc8040.device_renderer_service_path_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertIn('Roadm-connection successfully created for nodes',
-                      res['output']['result'])
         time.sleep(10)
 
     # Get Degree MC interface and check
     def test_08_degree_mc_interface(self):
-        response = test_utils.check_netconf_node_request("ROADM-D1",
-                                                         "interface/DEG1-TTP-TXRX-mc-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "DEG1-TTP-TXRX-mc-749:763")
+        self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertDictEqual(
             dict({"name": "DEG1-TTP-TXRX-mc-749:763",
                   "supporting-interface": "OMS-DEG1-TTP-TXRX",
@@ -167,23 +133,18 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
                   "description": "TBD",
                   "supporting-port": "L1",
                   "type": "org-openroadm-interfaces:mediaChannelTrailTerminationPoint"},
-                 **res['interface'][0]),
-            res['interface'][0])
+                 **response['interface'][0]), response['interface'][0])
 
         # Check the mc-ttp max and min-freq
-        self.assertEqual({
-            "min-freq": 196.00625,
-            "max-freq": 196.09375
-        },
-            res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
+        self.assertIn(response['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'],
+                      [{'min-freq': '196.00625', 'max-freq': '196.09375'},
+                       {'min-freq': 196.00625, 'max-freq': 196.09375}])
         time.sleep(3)
 
     # get DEG-NMC interface and check
     def test_09_degree_nmc_interface(self):
-        response = test_utils.check_netconf_node_request("ROADM-D1",
-                                                         "interface/DEG1-TTP-TXRX-nmc-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "DEG1-TTP-TXRX-nmc-749:763")
+        self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertDictEqual(
             dict({"name": "DEG1-TTP-TXRX-nmc-749:763",
                   "supporting-interface": "DEG1-TTP-TXRX-mc-749:763",
@@ -192,23 +153,18 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
                   "description": "TBD",
                   "supporting-port": "L1",
                   "type": "org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint"},
-                 **res['interface'][0]),
-            res['interface'][0])
+                 **response['interface'][0]), response['interface'][0])
 
         # Check the mc-ttp max and min-freq
-        self.assertEqual({
-            "frequency": 196.05,
-            "width": 80
-        },
-            res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
+        self.assertIn(response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'],
+                      [{'frequency': '196.05000', 'width': '80'},
+                       {'frequency': 196.05, 'width': 80}])
         time.sleep(3)
 
     # get SRG-NMC interface
     def test_10_srg_nmc_interface(self):
-        response = test_utils.check_netconf_node_request("ROADM-D1",
-                                                         "interface/SRG1-PP1-TXRX-nmc-749:763")
-        res = response.json()
-        self.assertEqual(response.status_code, requests.codes.ok)
+        response = test_utils_rfc8040.check_interface_request("ROADM-D1", "SRG1-PP1-TXRX-nmc-749:763")
+        self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(
             dict({"name": "SRG1-PP1-TXRX-nmc-749:763",
                   "supporting-circuit-pack-name": "3/0",
@@ -216,63 +172,54 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
                   "description": "TBD",
                   "supporting-port": "C1",
                   "type": "org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint"},
-                 **res['interface'][0]), res['interface'][0])
-        self.assertEqual({"frequency": 196.05, "width": 80},
-                         res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
+                 **response['interface'][0]), response['interface'][0])
+        self.assertIn(response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'],
+                      [{'frequency': '196.05000', 'width': '80'},
+                       {'frequency': 196.05, 'width': 80}])
         time.sleep(3)
 
     # Create ROADM-connection
     def test_11_roadm_connection(self):
-        response = test_utils.check_netconf_node_request("ROADM-D1",
-                                                         "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
+        response = test_utils_rfc8040.check_roadm_connections_request("ROADM-D1", "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",
-                         res['roadm-connections'][0]['connection-name'])
+                         response['roadm-connections'][0]['connection-name'])
         self.assertEqual("SRG1-PP1-TXRX-nmc-749:763",
-                         res['roadm-connections'][0]['source']['src-if'])
+                         response['roadm-connections'][0]['source']['src-if'])
         self.assertEqual("DEG1-TTP-TXRX-nmc-749:763",
-                         res['roadm-connections'][0]['destination']['dst-if'])
+                         response['roadm-connections'][0]['destination']['dst-if'])
         time.sleep(3)
 
     # Delete ROADM connections and interfaces
 
     # delete ROADM connection
     def test_12_delete_roadm_connection(self):
-        response = test_utils.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
-                                             "node/ROADM-D1/yang-ext:mount/" +
-                                             "org-openroadm-device:org-openroadm-device/" +
-                                             "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
+        response = test_utils_rfc8040.del_roadm_connections_request("ROADM-D1", "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.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
-                                             "node/ROADM-D1/yang-ext:mount/" +
-                                             "org-openroadm-device:org-openroadm-device/" +
-                                             "interface/SRG1-PP1-TXRX-nmc-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
+        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "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.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
-                                             "node/ROADM-D1/yang-ext:mount/" +
-                                             "org-openroadm-device:org-openroadm-device/" +
-                                             "interface/DEG1-TTP-TXRX-nmc-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
+        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "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.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
-                                             "node/ROADM-D1/yang-ext:mount/" +
-                                             "org-openroadm-device:org-openroadm-device/" +
-                                             "interface/DEG1-TTP-TXRX-mc-749:763")
-        self.assertEqual(response.status_code, requests.codes.ok)
+        response = test_utils_rfc8040.del_interface_request("ROADM-D1", "DEG1-TTP-TXRX-mc-749:763")
+        self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
         time.sleep(3)
 
+    def test_16_disconnect_ROADM_D1(self):
+        response = test_utils_rfc8040.unmount_device("ROADM-D1")
+        self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index 5c88187b138c532abf4791ca0ed3c97df3eba296..a6ab0517a9e90c0a903bb18a2b295168f8184a76 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -317,6 +317,56 @@ def check_device_connection(node: str):
     return {'status_code': response.status_code,
             'connection-status': connection_status}
 
+
+def check_interface_request(node: str, interface: 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))
+    res = response.json()
+    return_key = {'rfc8040': 'org-openroadm-device:interface',
+                  'draft-bierman02': 'interface'}
+    if return_key[RESTCONF_VERSION] in res.keys():
+        interface = res[return_key[RESTCONF_VERSION]]
+    else:
+        interface = res['errors']['error'][0]
+    return {'status_code': response.status_code,
+            'interface': interface}
+
+
+def del_interface_request(node: str, interface: 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
+
 #
 # Portmapping operations
 #
@@ -495,3 +545,31 @@ def connect_rdm_to_xpdr_request(payload: dict):
         }
     }
     return post_request(url, data)
+
+
+#
+# TransportPCE device renderer service path operations
+#
+
+def prepend_dict_keys(input_dict: dict, prefix: str):
+    return_dict = {}
+    for key, value in input_dict.items():
+        newkey = prefix + key
+        if isinstance(value, dict):
+            return_dict[newkey] = prepend_dict_keys(value, prefix)
+            # TODO: perhaps some recursion depth limit or another solution has to be considered here
+            # even if recursion depth is given by the input_dict argument
+            # direct (self-)recursive functions may carry unwanted side-effects such as ressource consumptions
+        else:
+            return_dict[newkey] = value
+    return return_dict
+
+
+def device_renderer_service_path_request(payload: dict):
+    url = "{}/operations/transportpce-device-renderer:service-path"
+    payload_prefix = {'rfc8040': '', 'draft-bierman02': 'transportpce-device-renderer:'}
+    if RESTCONF_VERSION == 'draft-bierman02':
+        data = prepend_dict_keys(payload, payload_prefix[RESTCONF_VERSION])
+    else:
+        data = payload
+    return post_request(url, data)