X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fpce%2Ftest02_pce_400G.py;h=dbd8e684a24b7e39d98f63c4fa00f0663812a8cd;hb=HEAD;hp=2979105ec965c476168b094187db0158139a3a8c;hpb=1012f1723c3815f5c704fa9a908c3cfdce58ec31;p=transportpce.git diff --git a/tests/transportpce_tests/pce/test02_pce_400G.py b/tests/transportpce_tests/pce/test02_pce_400G.py index 2979105ec..dbd8e684a 100644 --- a/tests/transportpce_tests/pce/test02_pce_400G.py +++ b/tests/transportpce_tests/pce/test02_pce_400G.py @@ -8,19 +8,43 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +# pylint: disable=invalid-name # pylint: disable=no-member # pylint: disable=too-many-public-methods import unittest import os +# pylint: disable=wrong-import-order import sys import time import requests sys.path.append('transportpce_tests/common/') -import test_utils +# pylint: disable=wrong-import-position +# pylint: disable=import-error +import test_utils # nopep8 class TransportPCE400Gtesting(unittest.TestCase): + path_computation_input_data = { + "service-name": "service-1", + "resource-reserve": "true", + "service-handler-header": { + "request-id": "request1" + }, + "service-a-end": { + "service-rate": "400", + "clli": "nodeA", + "service-format": "Ethernet", + "node-id": "XPDR-A2" + }, + "service-z-end": { + "service-rate": "400", + "clli": "nodeC", + "service-format": "Ethernet", + "node-id": "XPDR-C2" + }, + "pce-routing-metric": "hop-count" + } simple_topo_bi_dir_data = None port_mapping_data = None @@ -28,31 +52,37 @@ class TransportPCE400Gtesting(unittest.TestCase): @classmethod def setUpClass(cls): + # pylint: disable=bare-except + sample_files_parsed = False try: - sample_files_parsed = False TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-topo400G.json") - with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir: + "..", "..", "sample_configs", + "honeynode-topo400G.json") + with open(TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as topo_bi_dir: cls.topo_bi_dir_data = topo_bi_dir.read() OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400G.json") - with open(OTN_TOPO_BI_DIR_FILE, 'r') as otn_topo_bi_dir: + "..", "..", "sample_configs", + "honeynode-otntopo400G.json") + with open(OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otn_topo_bi_dir: cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read() OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json") - with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r') as otuc4_otn_topo_bi_dir: + "..", "..", "sample_configs", + "honeynode-otntopo400GwithOTUC4.json") + with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otuc4_otn_topo_bi_dir: cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read() ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json") - with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r') as oduc4_otn_topo_bi_dir: + "..", "..", "sample_configs", + "honeynode-otntopo400GwithODUC4.json") + with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as oduc4_otn_topo_bi_dir: cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read() PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), - "..", "..", "sample_configs", "pce_portmapping_71.json") - with open(PORT_MAPPING_FILE, 'r') as port_mapping: + "..", "..", "sample_configs", + "pce_portmapping_71.json") + with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping: cls.port_mapping_data = port_mapping.read() sample_files_parsed = True except PermissionError as err: @@ -71,181 +101,188 @@ class TransportPCE400Gtesting(unittest.TestCase): @classmethod def tearDownClass(cls): + # clean datastores + test_utils.del_portmapping() + test_utils.del_ietf_network('openroadm-topology') + test_utils.del_ietf_network('otn-topology') # pylint: disable=not-an-iterable for process in cls.processes: test_utils.shutdown_process(process) print("all processes killed") def setUp(self): # instruction executed before each test method + # pylint: disable=consider-using-f-string print("execution of {}".format(self.id().split(".")[-1])) time.sleep(1) # Load port mapping def test_01_load_port_mapping(self): - response = test_utils.put_jsonrequest(test_utils.URL_FULL_PORTMAPPING, self.port_mapping_data) - self.assertIn(response.status_code, (requests.codes.ok, requests.codes.created)) - time.sleep(2) + response = test_utils.post_portmapping(self.port_mapping_data) + self.assertIn(response['status_code'], (requests.codes.created, requests.codes.no_content)) + time.sleep(1) # Load openroadm topology def test_02_load_openroadm_topology_bi(self): - response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.topo_bi_dir_data) - self.assertEqual(response.status_code, requests.codes.ok) - time.sleep(2) + response = test_utils.put_ietf_network('openroadm-topology', self.topo_bi_dir_data) + self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content)) + time.sleep(1) # Path Computation success def test_03_path_computation_400G_xpdr_bi(self): - response = test_utils.path_computation_request("request-1", "service-1", - {"node-id": "XPDR-A2", "service-rate": "400", - "service-format": "Ethernet", "clli": "nodeA"}, - {"node-id": "XPDR-C2", "service-rate": "400", - "service-format": "Ethernet", "clli": "nodeC"}) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils.transportpce_api_rpc_request('transportpce-pce', + 'path-computation-request', + self.path_computation_input_data) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + response['output']['configuration-response-common']['response-message']) - self.assertEqual(1, res['output']['response-parameters']['path-description'] + self.assertEqual(1, response['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ-wavelength-number']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['aToZ-direction']['rate']) - self.assertEqual(196.0375, res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-min-frequency']) - self.assertEqual(196.12500, res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-max-frequency']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-min-frequency'])) + self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-max-frequency'])) + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['aToZ-direction']['modulation-format']) - self.assertEqual(1, res['output']['response-parameters']['path-description'] + self.assertEqual(1, response['output']['response-parameters']['path-description'] ['zToA-direction']['zToA-wavelength-number']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['zToA-direction']['rate']) - self.assertEqual(196.0375, res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-min-frequency']) - self.assertEqual(196.12500, res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-max-frequency']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-min-frequency'])) + self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-max-frequency'])) + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['zToA-direction']['modulation-format']) - time.sleep(5) + time.sleep(2) # Load otn topology def test_04_load_otn_topology_bi(self): - response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.otn_topo_bi_dir_data) - self.assertEqual(response.status_code, requests.codes.ok) - time.sleep(2) + response = test_utils.put_ietf_network('otn-topology', self.otn_topo_bi_dir_data) + self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content)) + time.sleep(1) # Path Computation success def test_05_path_computation_OTUC4_xpdr_bi(self): - response = test_utils.path_computation_request("request-1", "service-OTUC4", - {"service-rate": "400", "clli": "NodeA", - "service-format": "OTU", "node-id": "XPDR-A2", - "rx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}} - }, - {"service-rate": "400", "clli": "NodeC", - "service-format": "OTU", "node-id": "XPDR-C2", - "rx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}} - }) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + self.path_computation_input_data["service-name"] = "service-OTUC4" + self.path_computation_input_data["service-a-end"]["service-format"] = "OTU" + self.path_computation_input_data["service-a-end"]["tx-direction"] =\ + {"port": {"port-device-name": "XPDR-A2-XPDR2"}} + self.path_computation_input_data["service-z-end"]["service-format"] = "OTU" + self.path_computation_input_data["service-z-end"]["tx-direction"] =\ + {"port": {"port-device-name": "XPDR-C2-XPDR2"}} + response = test_utils.transportpce_api_rpc_request('transportpce-pce', + 'path-computation-request', + self.path_computation_input_data) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + response['output']['configuration-response-common']['response-message']) - self.assertEqual(1, res['output']['response-parameters']['path-description'] + self.assertEqual(1, response['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ-wavelength-number']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['aToZ-direction']['rate']) - self.assertEqual(196.0375, res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-min-frequency']) - self.assertEqual(196.12500, res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-max-frequency']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-min-frequency'])) + self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-max-frequency'])) + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['aToZ-direction']['modulation-format']) - self.assertEqual(1, res['output']['response-parameters']['path-description'] + self.assertEqual(1, response['output']['response-parameters']['path-description'] ['zToA-direction']['zToA-wavelength-number']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['zToA-direction']['rate']) - self.assertEqual(196.0375, res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-min-frequency']) - self.assertEqual(196.12500, res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-max-frequency']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-min-frequency'])) + self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-max-frequency'])) + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['zToA-direction']['modulation-format']) - time.sleep(5) + time.sleep(2) # Load otn topology with OTUC4 links def test_06_load_otuc4_otn_topology_bi(self): - response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.otuc4_otn_topo_bi_dir_data) - self.assertEqual(response.status_code, requests.codes.ok) - time.sleep(2) + response = test_utils.put_ietf_network('otn-topology', self.otuc4_otn_topo_bi_dir_data) + self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content)) + time.sleep(1) # Path Computation success def test_07_path_computation_ODUC4_xpdr_bi(self): - response = test_utils.path_computation_request("request-1", "service-ODUC4", - {"service-rate": "400", "clli": "NodeA", "service-format": "ODU", - "node-id": "XPDR-A2", - "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}} - }, - {"service-rate": "400", "clli": "NodeC", "service-format": "ODU", - "node-id": "XPDR-C2", - "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}} - }) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + self.path_computation_input_data["service-name"] = "service-ODUC4" + self.path_computation_input_data["service-a-end"]["service-format"] = "ODU" + self.path_computation_input_data["service-a-end"]["tx-direction"] = \ + {"port": {"port-device-name": "XPDR-A2-XPDR2"}} + self.path_computation_input_data["service-z-end"]["service-format"] = "ODU" + self.path_computation_input_data["service-z-end"]["tx-direction"] = \ + {"port": {"port-device-name": "XPDR-C2-XPDR2"}} + response = test_utils.transportpce_api_rpc_request('transportpce-pce', + 'path-computation-request', + self.path_computation_input_data) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + response['output']['configuration-response-common']['response-message']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['aToZ-direction']['rate']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['aToZ-direction']['modulation-format']) - self.assertEqual(400, res['output']['response-parameters']['path-description'] + self.assertEqual(400, response['output']['response-parameters']['path-description'] ['zToA-direction']['rate']) - self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description'] + self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description'] ['zToA-direction']['modulation-format']) - time.sleep(5) + time.sleep(2) # Load otn topology with OTUC4 links def test_08_load_oduc4_otn_topology_bi(self): - response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.oduc4_otn_topo_bi_dir_data) - self.assertEqual(response.status_code, requests.codes.ok) - time.sleep(2) + response = test_utils.put_ietf_network('otn-topology', self.oduc4_otn_topo_bi_dir_data) + self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content)) + time.sleep(1) # Path Computation success def test_09_path_computation_100G_xpdr_bi(self): - response = test_utils.path_computation_request("request-1", "service-100GE", - {"service-rate": "100", "clli": "NodeA", "service-format": "Ethernet", - "node-id": "XPDR-A2", - "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2", - "port-name": "XPDR2-CLIENT1"}}}, - {"service-rate": "100", "clli": "NodeC", "service-format": "Ethernet", - "node-id": "XPDR-C2", - "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2", - "port-name": "XPDR2-CLIENT1"}}}) - - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + self.path_computation_input_data["service-name"] = "service-100GE" + self.path_computation_input_data["service-a-end"]["service-rate"] = "100" + self.path_computation_input_data["service-a-end"]["service-format"] = "Ethernet" + self.path_computation_input_data["service-a-end"]["tx-direction"] = \ + {"port": {"port-device-name": "XPDR-A2-XPDR2", + "port-name": "XPDR2-CLIENT1"}} + self.path_computation_input_data["service-z-end"]["service-rate"] = "100" + self.path_computation_input_data["service-z-end"]["service-format"] = "Ethernet" + self.path_computation_input_data["service-z-end"]["tx-direction"] = \ + {"port": {"port-device-name": "XPDR-C2-XPDR2", + "port-name": "XPDR2-CLIENT1"}} + self.path_computation_input_data["service-z-end"]["service-format"] = "ODU" + response = test_utils.transportpce_api_rpc_request('transportpce-pce', + 'path-computation-request', + self.path_computation_input_data) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + response['output']['configuration-response-common']['response-message']) - self.assertEqual(100, res['output']['response-parameters']['path-description'] + self.assertEqual(100, response['output']['response-parameters']['path-description'] ['aToZ-direction']['rate']) - self.assertEqual('1.1', res['output']['response-parameters']['path-description'] + self.assertEqual('1.1', response['output']['response-parameters']['path-description'] ['aToZ-direction']['min-trib-slot']) - self.assertEqual('1.20', res['output']['response-parameters']['path-description'] + self.assertEqual('1.20', response['output']['response-parameters']['path-description'] ['aToZ-direction']['max-trib-slot']) - self.assertEqual('dp-qpsk', res['output']['response-parameters']['path-description'] + self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description'] ['aToZ-direction']['modulation-format']) - self.assertEqual(100, res['output']['response-parameters']['path-description'] + self.assertEqual(100, response['output']['response-parameters']['path-description'] ['zToA-direction']['rate']) - self.assertEqual('1.1', res['output']['response-parameters']['path-description'] + self.assertEqual('1.1', response['output']['response-parameters']['path-description'] ['zToA-direction']['min-trib-slot']) - self.assertEqual('1.20', res['output']['response-parameters']['path-description'] + self.assertEqual('1.20', response['output']['response-parameters']['path-description'] ['zToA-direction']['max-trib-slot']) - self.assertEqual('dp-qpsk', res['output']['response-parameters']['path-description'] + self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description'] ['zToA-direction']['modulation-format']) - time.sleep(5) + time.sleep(2) + if __name__ == "__main__": unittest.main(verbosity=2)