Migrate OTN topology func. tests to RFC8040

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

diff --git a/tests/transportpce_tests/2.2.1/test04_otn_topology.py b/tests/transportpce_tests/2.2.1/test04_otn_topology.py
index 7a72b34a4a67e3e286cdab7508efd49ab7e45f8b..e87ef6aab0858d44e3c7eb5ae5abceda47242d15 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test04_otn_topology.py
+++ b/tests/transportpce_tests/2.2.1/test04_otn_topology.py
@@ -21,7 +21,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 TransportPCEtesting(unittest.TestCase):
@@ -31,111 +31,94 @@ class TransportPCEtesting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls.processes = test_utils.start_tpce()
-        cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION)])
+        cls.processes = test_utils_rfc8040.start_tpce()
+        cls.processes = test_utils_rfc8040.start_sims([('spdra', 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):
         time.sleep(5)
 
     def test_01_connect_SPDR_SA1(self):
-        response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
-        self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
+        response = test_utils_rfc8040.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
+        self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
         time.sleep(10)
 
-        response = test_utils.get_netconf_oper_request("SPDR-SA1")
-        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("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(response['connection-status'], 'connected')
 
     def test_02_getClliNetwork(self):
-        response = test_utils.get_clli_net_request()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        logging.info(res)
-        self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeSA')
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
+        response = test_utils_rfc8040.get_ietf_network_request('clli-network', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        logging.info(response)
+        self.assertEqual(response['network'][0]['node'][0]['node-id'], 'NodeSA')
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
 
     def test_03_getOpenRoadmNetwork(self):
-        response = test_utils.get_ordm_net_request()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
-        self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['network-ref'], 'clli-network')
-        self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['node-ref'], 'NodeSA')
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-common-network:node-type'], 'XPONDER')
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'universal-switchponder')
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:vendor'], 'vendorA')
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-network', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(response['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
+        self.assertEqual(response['network'][0]['node'][0]['supporting-node'][0]['network-ref'], 'clli-network')
+        self.assertEqual(response['network'][0]['node'][0]['supporting-node'][0]['node-ref'], 'NodeSA')
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-common-network:node-type'], 'XPONDER')
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:model'], 'universal-switchponder')
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:vendor'], 'vendorA')
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
 
     def test_04_getLinks_OpenroadmTopology(self):
-        response = test_utils.get_ordm_topo_request("")
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        # Tests related to links
-        self.assertNotIn('ietf-network-topology:link', res['network'][0])
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertNotIn('ietf-network-topology:link', response['network'][0])
 
     def test_05_getNodes_OpenRoadmTopology(self):
         # pylint: disable=redundant-unittest-assert
-        response = test_utils.get_ordm_topo_request("")
-        res = response.json()
-        # Tests related to nodes
-        self.assertEqual(response.status_code, requests.codes.ok)
-        self.assertIn('node', res['network'][0])
-        if 'node' in res['network'][0]:
-            nbNode = len(res['network'][0]['node'])
-            self.assertEqual(nbNode, 3)
-            listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
-            for i in range(0, nbNode):
-                nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
-                nodeId = res['network'][0]['node'][i]['node-id']
-                if nodeId not in listNode:
-                    self.assertFalse(True)
-                    continue
-                self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
-                              res['network'][0]['node'][i]['supporting-node'])
-                self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
-                              res['network'][0]['node'][i]['supporting-node'])
-                self.assertEqual(nodeType, 'XPONDER')
-                nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
-                client = 0
-                network = 0
-                for j in range(0, nbTps):
-                    tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
-                                 ['org-openroadm-common-network:tp-type'])
-                    if tpType == 'XPONDER-CLIENT':
-                        client += 1
-                    elif tpType == 'XPONDER-NETWORK':
-                        network += 1
-                self.assertTrue(client == 0)
-                if nodeId in ('SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR3'):
-                    self.assertTrue(network == 1)
-                else:
-                    # elif nodeId == 'SPDR-SA1-XPDR2':
-                    self.assertTrue(network == 4)
-                listNode.remove(nodeId)
-            self.assertEqual(len(listNode), 0)
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn('node', response['network'][0])
+        self.assertEqual(len(response['network'][0]['node']), 3)
+        listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
+        for val in response['network'][0]['node']:
+            self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, val['supporting-node'])
+            self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'}, val['supporting-node'])
+            nodeType = val['org-openroadm-common-network:node-type']
+            nodeId = val['node-id']
+            if nodeId not in listNode:
+                self.assertFalse(True)
+                continue
+            self.assertEqual(nodeType, 'XPONDER')
+            client = 0
+            network = 0
+            for item in val['ietf-network-topology:termination-point']:
+                tpType = item['org-openroadm-common-network:tp-type']
+                if tpType == 'XPONDER-CLIENT':
+                    client += 1
+                elif tpType == 'XPONDER-NETWORK':
+                    network += 1
+            self.assertTrue(client == 0)
+            if nodeId in ('SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR3'):
+                self.assertTrue(network == 1)
+            else:
+                # elif nodeId == 'SPDR-SA1-XPDR2':
+                self.assertTrue(network == 4)
+            listNode.remove(nodeId)
+        self.assertEqual(len(listNode), 0)
 
     def test_06_getLinks_OtnTopology(self):
-        response = test_utils.get_otn_topo_request()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertNotIn('ietf-network-topology:link', res['network'][0])
+        response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertNotIn('ietf-network-topology:link', response['network'][0])
 
     def test_07_getNodes_OtnTopology(self):
         # pylint: disable=redundant-unittest-assert
-        response = test_utils.get_otn_topo_request()
-        res = response.json()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        self.assertEqual(len(res['network'][0]['node']), 3)
+        response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(len(response['network'][0]['node']), 3)
         listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
         CHECK_LIST = {
             'SPDR-SA1-XPDR1': {
@@ -161,7 +144,7 @@ class TransportPCEtesting(unittest.TestCase):
                 'tp-unchecklist': []
             }
         }
-        for val in res['network'][0]['node']:
+        for val in response['network'][0]['node']:
             nodeId = val['node-id']
             if nodeId in CHECK_LIST:
                 self.assertEqual(val['org-openroadm-common-network:node-type'], CHECK_LIST[nodeId]['node-type'])
@@ -220,32 +203,27 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(len(listNode), 0)
 
     def test_08_disconnect_SPDR_SA1(self):
-        response = test_utils.unmount_device("SPDR-SA1")
-        self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+        response = test_utils_rfc8040.unmount_device("SPDR-SA1")
+        self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
     def test_09_getClliNetwork(self):
-        response = test_utils.get_clli_net_request()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        nbNode = len(res['network'][0]['node'])
-        self.assertEqual(nbNode, 1)
-        self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
+        response = test_utils_rfc8040.get_ietf_network_request('clli-network', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(len(response['network'][0]['node']), 1)
+        self.assertEqual(response['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
 
     def test_10_getOpenRoadmNetwork(self):
-        response = test_utils.get_ordm_net_request()
-        self.assertEqual(response.status_code, requests.codes.ok)
-        res = response.json()
-        self.assertNotIn('node', res['network'][0])
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-network', 'config')
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertNotIn('node', response['network'][0])
 
     def test_11_getNodes_OpenRoadmTopology(self):
-        response = test_utils.get_ordm_topo_request("")
-        res = response.json()
-        self.assertNotIn('node', res['network'][0])
+        response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
+        self.assertNotIn('node', response['network'][0])
 
     def test_12_getNodes_OtnTopology(self):
-        response = test_utils.get_otn_topo_request()
-        res = response.json()
-        self.assertNotIn('node', res['network'][0])
+        response = test_utils_rfc8040.get_ietf_network_request('otn-topology', 'config')
+        self.assertNotIn('node', response['network'][0])
 
 
 if __name__ == "__main__":