X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=bbed914aaa9fd0346ba07fcc8ac9d2732c1b0bdc;hb=c834356f10cb8bc04c8b0a9eb2a438b2e9c50632;hp=4092c064975f6ec1d49a14d43b132d654347a190;hpb=6543a9d6021738e0fabaddd22f4f71f0c0008b0c;p=transportpce.git diff --git a/tests/transportpce_tests/1.2.1/test_pce.py b/tests/transportpce_tests/1.2.1/test_pce.py index 4092c0649..bbed914aa 100644 --- a/tests/transportpce_tests/1.2.1/test_pce.py +++ b/tests/transportpce_tests/1.2.1/test_pce.py @@ -8,13 +8,16 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +# pylint: disable=no-member +# pylint: disable=too-many-public-methods + import unittest -import json import os import sys import time import requests -from common import test_utils +sys.path.append('transportpce_tests/common/') +import test_utils class TransportPCEtesting(unittest.TestCase): @@ -22,10 +25,12 @@ class TransportPCEtesting(unittest.TestCase): simple_topo_bi_dir_data = None simple_topo_uni_dir_data = None complex_topo_uni_dir_data = None + port_mapping_data = None processes = None @classmethod def setUpClass(cls): + # pylint: disable=bare-except try: sample_files_parsed = False TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), @@ -43,6 +48,10 @@ class TransportPCEtesting(unittest.TestCase): "..", "..", "sample_configs", "NW-for-test-5-4.xml") with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex: cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read() + PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), + "..", "..", "sample_configs", "pce_portmapping_121.json") + with open(PORT_MAPPING_FILE, 'r') as port_mapping: + cls.port_mapping_data = port_mapping.read() sample_files_parsed = True except PermissionError as err: print("Permission Error when trying to read sample files\n", err) @@ -61,6 +70,7 @@ class TransportPCEtesting(unittest.TestCase): @classmethod def tearDownClass(cls): + # pylint: disable=not-an-iterable for process in cls.processes: test_utils.shutdown_process(process) print("all processes killed") @@ -68,6 +78,12 @@ class TransportPCEtesting(unittest.TestCase): def setUp(self): # instruction executed before each test method time.sleep(1) + # Load port mapping + def test_00_load_port_mapping(self): + response = test_utils.rawpost_request(test_utils.URL_FULL_PORTMAPPING, self.port_mapping_data) + self.assertEqual(response.status_code, requests.codes.no_content) + time.sleep(2) + # Load simple bidirectional topology def test_01_load_simple_topology_bi(self): response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data) @@ -95,29 +111,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_04_path_computation_xpdr_bi(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "XPDRA01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "nodeA" - }, - "service-z-end": { - "node-id": "XPDRC01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "nodeC" - } - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "XPDRA01", "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeA"}, + {"node-id": "XPDRC01", "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeC"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -126,29 +124,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_05_path_computation_rdm_bi(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "ROADMA01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "NodeA" - }, - "service-z-end": { - "node-id": "ROADMC01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "NodeC" - } - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "ROADMA01", "service-rate": "100", + "service-format": "Ethernet", "clli": "NodeA"}, + {"node-id": "ROADMC01", "service-rate": "100", + "service-format": "Ethernet", "clli": "NodeC"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -164,7 +144,7 @@ class TransportPCEtesting(unittest.TestCase): # Test deleted topology def test_07_test_topology_simple_bi_deleted(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") - self.assertEqual(response.status_code, 404) + self.assertEqual(response.status_code, requests.codes.conflict) time.sleep(1) # Load simple bidirectional topology @@ -195,29 +175,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_11_path_computation_xpdr_uni(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "XPONDER-1-2", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "ORANGE1" - }, - "service-z-end": { - "node-id": "XPONDER-3-2", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "ORANGE3" - } - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "XPONDER-1-2", "service-rate": "100", + "service-format": "Ethernet", "clli": "ORANGE1"}, + {"node-id": "XPONDER-3-2", "service-rate": "100", + "service-format": "Ethernet", "clli": "ORANGE3"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -226,29 +188,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_12_path_computation_rdm_uni(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "clli": "cll21", - "node-id": "OpenROADM-2-1" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "clli": "ncli22", - "node-id": "OpenROADM-2-2" - }, - "pce-metric": "hop-count" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request1", "service1", + {"service-rate": "100", "service-format": "Ethernet", + "clli": "cll21", "node-id": "OpenROADM-2-1"}, + {"service-rate": "100", "service-format": "Ethernet", + "clli": "ncli22", "node-id": "OpenROADM-2-2"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -261,9 +205,9 @@ class TransportPCEtesting(unittest.TestCase): for i in range(0, 15): atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i] - if (atoz['id'] == '14'): + if atoz['id'] == '14': self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX') - if (ztoa['id'] == '0'): + if ztoa['id'] == '0': self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX') time.sleep(5) @@ -276,7 +220,7 @@ class TransportPCEtesting(unittest.TestCase): # Test deleted topology def test_14_test_topology_simple_deleted(self): response = test_utils.get_ordm_topo_request("node/XPONDER-1-2") - self.assertEqual(response.status_code, 404) + self.assertEqual(response.status_code, requests.codes.conflict) time.sleep(1) # Load complex topology @@ -297,14 +241,8 @@ class TransportPCEtesting(unittest.TestCase): # Test failed path computation def test_17_fail_path_computation(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-handler-header": { - "request-id": "request-1" - } - } - } - response = test_utils.post_request(url, data) + response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST, + {"input": {"service-handler-header": {"request-id": "request-1"}}}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service Name is not set', @@ -313,94 +251,54 @@ class TransportPCEtesting(unittest.TestCase): # Test1 success path computation def test_18_success1_path_computation(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request1" - }, - "service-a-end": { - "service-format": "Ethernet", - "service-rate": "100", - "clli": "ORANGE2", - "node-id": "XPONDER-2-2", - "tx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - }, - "rx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - } - }, - "service-z-end": { - "service-format": "Ethernet", - "service-rate": "100", - "clli": "ORANGE1", - "node-id": "XPONDER-1-2", - "tx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - }, - "rx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - } - }, - "hard-constraints": { - "customer-code": [ - "Some customer-code" - ], - "co-routing": { - "existing-service": [ - "Some existing-service" - ] - } - }, - "soft-constraints": { - "customer-code": [ - "Some customer-code" - ], - "co-routing": { - "existing-service": [ - "Some existing-service" - ] - } - }, - "pce-metric": "hop-count", - "locally-protected-links": "true" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request1", "service1", + {"service-format": "Ethernet", "service-rate": "100", + "clli": "ORANGE2", "node-id": "XPONDER-2-2", + "tx-direction": {"port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + }}, + "rx-direction": {"port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + }}}, + {"service-format": "Ethernet", "service-rate": "100", + "clli": "ORANGE1", "node-id": "XPONDER-1-2", + "tx-direction": {"port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + }}, + "rx-direction": {"port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + }}}, + {"customer-code": ["Some customer-code"], + "co-routing": {"existing-service": ["Some existing-service"]} + }, + {"customer-code": ["Some customer-code"], + "co-routing": {"existing-service": ["Some existing-service"]} + }, + "hop-count", {"locally-protected-links": "true"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -409,29 +307,11 @@ class TransportPCEtesting(unittest.TestCase): # Test2 success path computation with path description def test_19_success2_path_computation(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service 1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request 1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-1-2", - "clli": "ORANGE1" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-3-2", - "clli": "ORANGE3" - }, - "pce-metric": "hop-count" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-3-2", "clli": "ORANGE3"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -444,34 +324,12 @@ class TransportPCEtesting(unittest.TestCase): # Test3 success path computation with hard-constraints exclude def test_20_success3_path_computation(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service 1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request 1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-1-2", - "clli": "ORANGE1" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-3-2", - "clli": "ORANGE3" - }, - "hard-constraints": { - "exclude_": { - "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"] - } - }, - "pce-metric": "hop-count" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-3-2", "clli": "ORANGE3"}, + {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -484,29 +342,11 @@ class TransportPCEtesting(unittest.TestCase): # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2 def test_21_path_computation_before_oms_attribute_deletion(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service 1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request 1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-2-2", - "clli": "ORANGE2" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-1-2", - "clli": "ORANGE1" - }, - "pce-metric": "hop-count" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-2-2", "clli": "ORANGE2"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -514,11 +354,12 @@ class TransportPCEtesting(unittest.TestCase): nbElmPath = len(res['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ']) self.assertEqual(31, nbElmPath) - link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"} + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"} find = False for i in range(0, nbElmPath): - resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource'] - if(resource_i == link): + resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] + ['resource']) + if resource_i == link: find = True self.assertEqual(find, True) time.sleep(5) @@ -531,29 +372,11 @@ class TransportPCEtesting(unittest.TestCase): # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2 def test_23_path_computation_after_oms_attribute_deletion(self): - url = "{}/operations/transportpce-pce:path-computation-request" - data = {"input": { - "service-name": "service 1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request 1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-2-2", - "clli": "ORANGE2" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-1-2", - "clli": "ORANGE1" - }, - "pce-metric": "hop-count" - } - } - response = test_utils.post_request(url, data) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-2-2", "clli": "ORANGE2"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -561,11 +384,12 @@ class TransportPCEtesting(unittest.TestCase): nbElmPath = len(res['output']['response-parameters']['path-description'] ['aToZ-direction']['aToZ']) self.assertEqual(47, nbElmPath) - link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"} + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"} find = False for i in range(0, nbElmPath): - resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource'] - if (resource_i == link): + resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] + ['resource']) + if resource_i == link: find = True self.assertNotEqual(find, True) time.sleep(5) @@ -579,9 +403,15 @@ class TransportPCEtesting(unittest.TestCase): # Test deleted complex topology def test_25_test_topology_complex_deleted(self): response = test_utils.get_ordm_topo_request("node/XPONDER-3-2") - self.assertEqual(response.status_code, 404) + self.assertEqual(response.status_code, requests.codes.conflict) time.sleep(1) + # Delete portmapping + def test_26_delete_port_mapping(self): + response = test_utils.delete_request(test_utils.URL_FULL_PORTMAPPING) + self.assertEqual(response.status_code, requests.codes.ok) + time.sleep(2) + if __name__ == "__main__": unittest.main(verbosity=2)