X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F7.1%2Ftest01_portmapping.py;h=6aed5f30971968dc6c4f8eeed699ad397b9211cf;hb=819d009eb8f97c15bdf074e132612f17f2e4fbc1;hp=34632ee2e1980cf637ed7d9224e9f2e9a253ded9;hpb=83f3af2d91fb44be21d2d6a17b3539784068211c;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..6aed5f309 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,31 @@ 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(str(res['mc-capabilities'][0]['center-freq-granularity']), '3.125') + self.assertEqual(str(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() - 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']) + response = test_utils_rfc8040.check_device_connection("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.conflict) + self.assertIn(response['connection-status']['error-type'], ('protocol', 'application')) + self.assertEqual(response['connection-status']['error-tag'], 'data-missing') + self.assertEqual(response['connection-status']['error-message'], + 'Request could not be completed because the relevant data model content does not exist') 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() - 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']) + response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + self.assertEqual(response['status_code'], requests.codes.conflict) + self.assertIn(response['node-info']['error-type'], ('protocol', 'application')) + self.assertEqual(response['node-info']['error-tag'], 'data-missing') + self.assertEqual(response['node-info']['error-message'], + 'Request could not be completed because the relevant data model content does not exist') if __name__ == '__main__':