X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F7.1%2Ftest01_portmapping.py;h=b5317f550fcdd73d4c968a2dc0e8b4959c2abe2c;hb=80c1e9a96ad3e992fb6504463ce1eaacd1e5a7e9;hp=34632ee2e1980cf637ed7d9224e9f2e9a253ded9;hpb=89d54863e044b85f1dababf3ed92cf50a7436552;p=transportpce.git diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py index 34632ee2e..b5317f550 100644 --- a/tests/transportpce_tests/7.1/test01_portmapping.py +++ b/tests/transportpce_tests/7.1/test01_portmapping.py @@ -14,9 +14,12 @@ import unittest import time import requests +# pylint: disable=wrong-import-order import sys sys.path.append('transportpce_tests/common') -import test_utils +# pylint: disable=wrong-import-position +# pylint: disable=import-error +import test_utils_rfc8040 # nopep8 class TransportPCE400GPortMappingTesting(unittest.TestCase): @@ -26,56 +29,52 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): @classmethod def setUpClass(cls): - cls.processes = test_utils.start_tpce() - cls.processes = test_utils.start_sims([('xpdra2', cls.NODE_VERSION)]) + cls.processes = test_utils_rfc8040.start_tpce() + cls.processes = test_utils_rfc8040.start_sims([('xpdra2', 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): + # pylint: disable=consider-using-f-string print("execution of {}".format(self.id().split(".")[-1])) time.sleep(10) def test_01_xpdr_device_connection(self): - response = test_utils.mount_device("XPDR-A2", - ('xpdra2', self.NODE_VERSION)) + response = test_utils_rfc8040.mount_device("XPDR-A2", + ('xpdra2', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, - test_utils.CODE_SHOULD_BE_201) + test_utils_rfc8040.CODE_SHOULD_BE_201) # Check if the node appears in the ietf-network topology def test_02_xpdr_device_connected(self): - response = test_utils.get_netconf_oper_request("XPDR-A2") - 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("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(response['connection-status'], 'connected') time.sleep(10) # Check node info in the port-mappings def test_03_xpdr_portmapping_info(self): - response = test_utils.portmapping_request("XPDR-A2/node-info") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - {u'node-info': {u'node-type': u'xpdr', - u'node-ip-address': u'1.2.3.4', - u'node-clli': u'NodeA', - u'openroadm-version': u'7.1', - u'node-vendor': u'vendorA', - u'node-model': u'model'}}, - res) + {'node-type': 'xpdr', + 'node-ip-address': '1.2.3.4', + 'node-clli': 'NodeA', + 'openroadm-version': '7.1', + 'node-vendor': 'vendorA', + 'node-model': 'model'}, + response['node-info']) time.sleep(3) # Check the if-capabilities and the other details for network def test_04_tpdr_portmapping_NETWORK1(self): - response = test_utils.portmapping_request("XPDR-A2/mapping/XPDR1-NETWORK1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-NETWORK1") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-otsi-otsigroup'], @@ -90,27 +89,31 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'port-oper-state': 'InService', 'xponder-type': 'tpdr' }, - res['mapping']) + response['mapping']) def test_05_tpdr_portmapping_CLIENT1(self): - response = test_utils.portmapping_request("XPDR-A2/mapping/XPDR1-CLIENT1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-CLIENT1") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'], 'supporting-port': 'C1', 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT', - 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional', - 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client', + 'logical-connection-point': 'XPDR1-CLIENT1', + 'port-direction': 'bidirectional', + 'connection-map-lcp': 'XPDR1-NETWORK1', + 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AODABTVSOHH0', - 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, - res['mapping']) + 'port-admin-state': 'InService', + 'port-oper-state': 'InService', + 'xponder-type': 'tpdr' + }, + response['mapping'] + ) # Check the port-mapping for the switch-client and switch-network port-quals def test_06_mpdr_portmapping_NETWORK1(self): - response = test_utils.portmapping_request("XPDR-A2/mapping/XPDR2-NETWORK1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-otsi-otsigroup'], @@ -124,12 +127,11 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'port-oper-state': 'InService', 'xponder-type': 'mpdr' }, - res['mapping']) + response['mapping']) def test_07_mpdr_portmapping_CLIENT1(self): - response = test_utils.portmapping_request("XPDR-A2/mapping/XPDR2-CLIENT1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + res = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-CLIENT1") + self.assertEqual(res['status_code'], requests.codes.ok) self.assertIn('org-openroadm-port-types:if-100GE-ODU4', res['mapping'][0]['supported-interface-capability']) self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4', @@ -142,6 +144,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual('AK+Cna4EclRH', res['mapping'][0]['lcp-hash-val']) self.assertEqual('InService', res['mapping'][0]['port-admin-state']) self.assertEqual('InService', res['mapping'][0]['port-oper-state']) + self.assertEqual('mpdr', res['mapping'][0]['xponder-type']) self.assertEqual({ "min-trib-slot": "1.1", "max-trib-slot": "1.20" @@ -149,36 +152,33 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Added test to check mc-capability-profile for a transponder def test_08_check_mccapprofile(self): - response = test_utils.portmapping_request("XPDR-A2/mc-capabilities/XPDR-mcprofile") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "XPDR-mcprofile") + self.assertEqual(res['status_code'], requests.codes.ok) self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile') - self.assertEqual(res['mc-capabilities'][0]['center-freq-granularity'], 3.125) - self.assertEqual(res['mc-capabilities'][0]['slot-width-granularity'], 6.25) + self.assertEqual(res['mc-capabilities'][0]['center-freq-granularity'], '3.125') + self.assertEqual(res['mc-capabilities'][0]['slot-width-granularity'], '6.25') def test_09_xpdr_device_disconnection(self): - response = test_utils.unmount_device("XPDR-A2") - self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) + response = test_utils_rfc8040.unmount_device("XPDR-A2") + self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_10_xpdr_device_disconnected(self): - response = test_utils.get_netconf_oper_request("XPDR-A2") - self.assertEqual(response.status_code, requests.codes.conflict) - res = response.json() + response = test_utils_rfc8040.check_device_connection("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.conflict) self.assertIn( - {"error-type": "application", "error-tag": "data-missing", - "error-message": "Request could not be completed because the " - "relevant data model content does not exist"}, - res['errors']['error']) + {"error-tag": "data-missing", + "error-message": "Request could not be completed because the relevant data model content does not exist", + "error-type": "protocol"}, + response['connection-status']) def test_11_xpdr_device_not_connected(self): - response = test_utils.portmapping_request("XPDR-A2") - self.assertEqual(response.status_code, requests.codes.conflict) - res = response.json() + response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.conflict) self.assertIn( - {"error-type": "application", "error-tag": "data-missing", - "error-message": "Request could not be completed because the " - "relevant data model content does not exist"}, - res['errors']['error']) + {"error-tag": "data-missing", + "error-message": "Request could not be completed because the relevant data model content does not exist", + "error-type": "protocol"}, + response['node-info']) if __name__ == '__main__':