From: Balagangadhar (Bala) Bathula Date: Fri, 31 Mar 2023 11:53:04 +0000 (-0400) Subject: Add functional tests for regenerator type X-Git-Tag: 8.0.0~51^2 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=transportpce.git;a=commitdiff_plain;h=d9d6bea5b8ee420183152c3d8c7a8a355b1a904b Add functional tests for regenerator type - Verify the port-mapping data - Verify renderer for creating regenerator interfaces - Verify renderer for deleting regnerator interfaces JIRA: TRNSPRTPCE-740 Change-Id: I6051cc4b30d46f1e6b7baf8624ee88aea0a9d4f1 Signed-off-by: Balagangadhar (Bala) Bathula --- diff --git a/tests/transportpce_tests/7.1/test04_renderer_regen_mode.py b/tests/transportpce_tests/7.1/test04_renderer_regen_mode.py new file mode 100644 index 000000000..b84c942f1 --- /dev/null +++ b/tests/transportpce_tests/7.1/test04_renderer_regen_mode.py @@ -0,0 +1,314 @@ +#!/usr/bin/env python +############################################################################## +# Copyright (c) 2023 AT&T, Inc. and others. All rights reserved. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################## + +# pylint: disable=no-member +# pylint: disable=too-many-public-methods + +# pylint: disable=wrong-import-order +import sys +import time +import unittest + +import requests + +sys.path.append("transportpce_tests/common") +# pylint: disable=wrong-import-position +# pylint: disable=import-error +import test_utils # nopep8 + + +class TransportPCE400GPortMappingTesting(unittest.TestCase): + + processes = None + NETWORK1_CHECK_DICT = {"logical-connection-point": "XPDR4-NETWORK1", + "supporting-port": "L1", + "supported-interface-capability": [ + "org-openroadm-port-types:if-otsi-otsigroup" + ], + "port-direction": "bidirectional", + "port-qual": "xpdr-network", + "supporting-circuit-pack-name": "1/1/6-PLUG-NET", + "xpdr-type": "regen", + "port-admin-state": "InService", + "port-oper-state": "InService", + "regen-profiles": { + "regen-profile": [ + "OTUC4-REGEN", + "OTUC2-REGEN", + "OTUC3-REGEN" + ] + } + } + REGEN_CAPABILITIES = ["OTUC2-REGEN", + "OTUC3-REGEN", + "OTUC4-REGEN", + ] + NODE_VERSION = "7.1" + + @classmethod + def setUpClass(cls): + cls.processes = test_utils.start_tpce() + cls.processes = test_utils.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) + print("all processes killed") + + def setUp(self): + # pylint: disable=consider-using-f-string + print("execution of {}".format(self.id().split(".")[-1])) + time.sleep(2) + + def test_01_xpdr_device_connection(self): + response = test_utils.mount_device("XPDR-A2", + ("xpdra2", self.NODE_VERSION)) + self.assertEqual(response.status_code, requests.codes.created, + test_utils.CODE_SHOULD_BE_201) + # Check the xpdr-type as regen + + def test_02_check_xpdr_type(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK1") + self.assertEqual(response["status_code"], requests.codes.ok) + self.assertEqual( + "regen", response["mapping"][0]["xpdr-type"]) + self.assertEqual( + self.REGEN_CAPABILITIES, + sorted(response["mapping"][0]["regen-profiles"]["regen-profile"])) + + # Check the xpdr-type as regen + + def test_03_check_xpdr_type(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK2") + self.assertEqual(response["status_code"], requests.codes.ok) + self.assertEqual( + "regen", response["mapping"][0]["xpdr-type"]) + self.assertEqual( + self.REGEN_CAPABILITIES, + sorted(response["mapping"][0]["regen-profiles"]["regen-profile"])) + + def test_04_400g_regen_service_path_create(self): + response = test_utils.transportpce_api_rpc_request( + "transportpce-device-renderer", "service-path", + { + "service-name": "service_400g_regen", + "wave-number": "0", + "modulation-format": "dp-qam16", + "operation": "create", + "nodes": [ + { + "node-id": "XPDR-A2", + "src-tp": "XPDR4-NETWORK1", + "dest-tp": "XPDR4-NETWORK2" + } + ], + "center-freq": 195.0, + "nmc-width": 75, + "mc-width": 87.5, + "min-freq": 194.95625, + "max-freq": 195.04375, + "lower-spectral-slot-number": 582, + "higher-spectral-slot-number": 595, + }) + self.assertEqual(response["status_code"], requests.codes.ok) + self.assertIn("Interfaces created successfully for nodes: ", response["output"]["result"]) + + self.assertEqual( + {"node-id": "XPDR-A2", + "otu-interface-id": [ + "XPDR4-NETWORK1-OTUC4", + "XPDR4-NETWORK2-OTUC4" + ], + "odu-interface-id": [ + "XPDR4-NETWORK1-ODUC4", + "XPDR4-NETWORK2-ODUC4" + ], + "och-interface-id": [ + "XPDR4-NETWORK1-582:595", + "XPDR4-NETWORK1-OTSIGROUP-400G", + "XPDR4-NETWORK2-582:595", + "XPDR4-NETWORK2-OTSIGROUP-400G", + ]}, + {x: (sorted(response["output"]["node-interface"][0][x]) + if isinstance(response["output"]["node-interface"][0][x], list) + else response["output"]["node-interface"][0][x]) + for x in response["output"]["node-interface"][0].keys()}) + + def test_05_get_portmapping_network1(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK1") + self.assertEqual(response["status_code"], requests.codes.ok) + self.NETWORK1_CHECK_DICT["supporting-otucn"] = "XPDR4-NETWORK1-OTUC4" + self.NETWORK1_CHECK_DICT["lcp-hash-val"] = "AKkLPpWaa8x+" + self.NETWORK1_CHECK_DICT["connection-map-lcp"] = "XPDR4-NETWORK2" + self.assertIn( + self.NETWORK1_CHECK_DICT, + response["mapping"]) + + def test_06_get_portmapping_network1(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK2") + self.assertEqual(response["status_code"], requests.codes.ok) + self.NETWORK1_CHECK_DICT["logical-connection-point"] = "XPDR4-NETWORK2" + self.NETWORK1_CHECK_DICT["supporting-otucn"] = "XPDR4-NETWORK2-OTUC4" + self.NETWORK1_CHECK_DICT["lcp-hash-val"] = "AKkLPpWaa8x9" + self.NETWORK1_CHECK_DICT["connection-map-lcp"] = "XPDR4-NETWORK1" + self.NETWORK1_CHECK_DICT["supporting-circuit-pack-name"] = "1/1/5-PLUG-NET" + self.assertIn( + self.NETWORK1_CHECK_DICT, + response["mapping"]) + + def test_07_check_interface_oduc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK1-ODUC4") + self.assertEqual(response["status_code"], requests.codes.ok) + + input_dict_1 = {"name": "XPDR4-NETWORK1-ODUC4", + "administrative-state": "inService", + "supporting-circuit-pack-name": "1/1/6-PLUG_NET", + "supporting-interface-list": "XPDR4-NETWORK1-OTUC4", + "type": "org-openroadm-interfaces:otnOdu", + "supporting-port": "L1"} + + input_dict_2 = {"odu-function": "org-openroadm-otn-common-types:ODU-CTP", + "rate": "org-openroadm-otn-common-types:ODUCn", + "degm-intervals": 2, + "degthr-percentage": 100, + "monitoring-mode": "not-terminated" + } + self.assertDictEqual(dict(input_dict_1, **response["interface"][0]), + response["interface"][0]) + self.assertDictEqual(dict(input_dict_2, **response["interface"][0]["org-openroadm-otn-odu-interfaces:odu"]), + response["interface"][0]["org-openroadm-otn-odu-interfaces:odu"]) + + def test_08_check_interface_oduc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK2-ODUC4") + self.assertEqual(response["status_code"], requests.codes.ok) + + input_dict_1 = {"name": "XPDR4-NETWORK2-ODUC4", + "administrative-state": "inService", + "supporting-circuit-pack-name": "1/1/5-PLUG_NET", + "supporting-interface-list": "XPDR4-NETWORK2-ODUC4", + "type": "org-openroadm-interfaces:otnOdu", + "supporting-port": "L1"} + + input_dict_2 = {"odu-function": "org-openroadm-otn-common-types:ODU-CTP", + "rate": "org-openroadm-otn-common-types:ODUCn", + "degm-intervals": 2, + "degthr-percentage": 100, + "monitoring-mode": "not-terminated" + } + self.assertDictEqual(dict(input_dict_1, **response["interface"][0]), + response["interface"][0]) + self.assertDictEqual(dict(input_dict_2, **response["interface"][0]["org-openroadm-otn-odu-interfaces:odu"]), + response["interface"][0]["org-openroadm-otn-odu-interfaces:odu"]) + + # Delete the service path + + def test_09_service_path_delete_regen(self): + response = test_utils.transportpce_api_rpc_request( + "transportpce-device-renderer", "service-path", + { + "modulation-format": "dp-qam16", + "operation": "delete", + "service-name": "test1_regen", + "wave-number": "0", + "center-freq": 195.0, + "nmc-width": 75, + "mc-width": 87.5, + "min-freq": 194.95625, + "max-freq": 195.04375, + "lower-spectral-slot-number": 582, + "higher-spectral-slot-number": 595, + "nodes": [ + { + "node-id": "XPDR-A2", + "src-tp": "XPDR4-NETWORK1", + "dest-tp": "XPDR4-NETWORK2" + } + ] + }) + self.assertEqual(response["status_code"], requests.codes.ok) + self.assertIn("Request processed", response["output"]["result"]) + + def test_10_check_no_interface_oduc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK1-ODUC4") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_11_check_no_interface_oduc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK2-ODUC4") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_12_check_no_interface_otuc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK1-OTUC1") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_13_check_no_interface_otuc4(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK2-OTUC1") + self.assertEqual(response["status_code"], requests.codes.conflict) + + # Check if port-mapping data is updated, where the supporting-otucn is deleted + def test_14_check_no_otuc4(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK1") + self.assertRaises(KeyError, lambda: response["supporting-otucn"]) + + def test_15_check_no_otuc4(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR4-NETWORK1") + self.assertRaises(KeyError, lambda: response["supporting-otucn"]) + + def test_16_check_no_interface_otsig(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK1-OTSIGROUP-400G") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_17_check_no_interface_otsig(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK2-OTSIGROUP-400G") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_18_check_no_interface_otsi(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK1-582:595") + self.assertEqual(response["status_code"], requests.codes.conflict) + + def test_19_check_no_interface_otsi(self): + response = test_utils.check_node_attribute_request( + "XPDR-A2", "interface", "XPDR4-NETWORK2-582:595") + self.assertEqual(response["status_code"], requests.codes.conflict) + + # Disconnect the XPDR + def test_20_xpdr_device_disconnection(self): + response = test_utils.unmount_device("XPDR-A2") + self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) + + def test_21_xpdr_device_disconnected(self): + response = test_utils.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_22_xpdr_device_not_connected(self): + response = test_utils.get_portmapping_node_attr("XPDR-A2", "node-info", None) + 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__": + unittest.main(verbosity=2)