From 35ad809f1c6767abb5102927d42c655a4b809d26 Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Wed, 19 Jan 2022 16:41:18 +0100 Subject: [PATCH] Migrate OTN topology func. tests to RFC8040 JIRA: TRNSPRTPCE-567 Signed-off-by: guillaume.lambert Change-Id: Icb22e78df1f6728dfead4f23277517e7a41eb596 --- .../2.2.1/test04_otn_topology.py | 174 ++++++++---------- 1 file changed, 76 insertions(+), 98 deletions(-) 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 7a72b34a4..e87ef6aab 100644 --- 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__": -- 2.36.6