From: guillaume.lambert Date: Sat, 11 Jul 2020 09:01:13 +0000 (+0200) Subject: add a portmapping method in func tests X-Git-Tag: 2.0.0~40 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F28%2F91228%2F1;p=transportpce.git add a portmapping method in func tests Signed-off-by: guillaume.lambert Change-Id: I56e0208cbac615defc8ff6db7b0dc8fd9859d6f0 --- diff --git a/tests/transportpce_tests/1.2.1/test_portmapping.py b/tests/transportpce_tests/1.2.1/test_portmapping.py index a1daf17f2..9bfeb4840 100644 --- a/tests/transportpce_tests/1.2.1/test_portmapping.py +++ b/tests/transportpce_tests/1.2.1/test_portmapping.py @@ -42,9 +42,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): # 'connected') # def test_02_restconfAPI(self): -# url = ("{}/config/transportpce-portmapping:network/nodes/controller-config") -# response = requests.request( -# "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) +# response = test_utils.portmapping_request("controller-config") # self.assertEqual(response.status_code, requests.codes.not_found) # res = response.json() # self.assertIn( @@ -66,10 +64,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_03_rdm_portmapping_info(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/node-info" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -82,10 +77,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_04_rdm_portmapping_DEG1_TTP_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/DEG1-TTP-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/DEG1-TTP-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -94,10 +86,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_05_rdm_portmapping_SRG1_PP7_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/SRG1-PP7-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -106,10 +95,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_06_rdm_portmapping_SRG3_PP1_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/SRG3-PP1-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -131,10 +117,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_09_xpdr_portmapping_info(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/node-info" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -147,10 +130,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_10_xpdr_portmapping_NETWORK1(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-NETWORK1" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -161,10 +141,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_11_xpdr_portmapping_NETWORK2(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-NETWORK2" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -175,10 +152,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_12_xpdr_portmapping_CLIENT1(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT1" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -190,10 +164,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_13_xpdr_portmapping_CLIENT2(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT2" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -205,10 +176,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_14_xpdr_portmapping_CLIENT3(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT3" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -219,11 +187,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'}, res['mapping']) - def test_15_xpdr_portmapping_CLIENT4(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT4" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -247,8 +211,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_18_xpdr_device_not_connected(self): - url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01") - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01") self.assertEqual(response.status_code, requests.codes.not_found) res = response.json() self.assertIn( @@ -270,8 +233,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_21_rdm_device_not_connected(self): - url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01") - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01") self.assertEqual(response.status_code, requests.codes.not_found) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py b/tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py index 7dd4bf66a..d1f7ed766 100644 --- a/tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.py @@ -44,10 +44,7 @@ class TransportPCERendererTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_03_rdm_portmapping(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -60,10 +57,7 @@ class TransportPCERendererTesting(unittest.TestCase): res['nodes'][0]['mapping']) def test_04_xpdr_portmapping(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/1.2.1/test_topoPortMapping.py b/tests/transportpce_tests/1.2.1/test_topoPortMapping.py index 38fb5e652..1c3ef934c 100644 --- a/tests/transportpce_tests/1.2.1/test_topoPortMapping.py +++ b/tests/transportpce_tests/1.2.1/test_topoPortMapping.py @@ -48,8 +48,7 @@ class TransportPCEtesting(unittest.TestCase): for i in range(0, nbNode): nodeId = resTopo['network'][0]['node'][i]['node-id'] nodeMapId = nodeId.split("-")[0] - urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId - responseMapList = test_utils.get_request(urlMapList) + responseMapList = test_utils.portmapping_request(nodeMapId) resMapList = responseMapList.json() nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point']) @@ -57,8 +56,7 @@ class TransportPCEtesting(unittest.TestCase): for j in range(0, nbTp): tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id'] if((not "CP" in tpId) and (not "CTP" in tpId)): - urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId - responseMap = test_utils.get_request(urlMap) + responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId) # Disconnect the ROADMA def test_03_disconnect_rdm(self): diff --git a/tests/transportpce_tests/2.2.1/test_otn_renderer.py b/tests/transportpce_tests/2.2.1/test_otn_renderer.py index 0fb79f45e..f07c78e0a 100644 --- a/tests/transportpce_tests/2.2.1/test_otn_renderer.py +++ b/tests/transportpce_tests/2.2.1/test_otn_renderer.py @@ -47,8 +47,7 @@ class TransportPCEtesting(unittest.TestCase): 'connected') def test_02_get_portmapping_CLIENT1(self): - url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-CLIENT1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -65,8 +64,7 @@ class TransportPCEtesting(unittest.TestCase): res['mapping']) def test_03_get_portmapping_NETWORK1(self): - url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -104,8 +102,7 @@ class TransportPCEtesting(unittest.TestCase): 'och-interface-id': ['XPDR1-NETWORK1-1']}, res["output"]['node-interface']) def test_05_get_portmapping_NETWORK1(self): - url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -186,8 +183,7 @@ class TransportPCEtesting(unittest.TestCase): 'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, res["output"]['node-interface']) def test_09_get_portmapping_NETWORK1(self): - url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/2.2.1/test_portmapping.py b/tests/transportpce_tests/2.2.1/test_portmapping.py index 942a0ee5a..d84aa8ffa 100644 --- a/tests/transportpce_tests/2.2.1/test_portmapping.py +++ b/tests/transportpce_tests/2.2.1/test_portmapping.py @@ -48,8 +48,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_03_rdm_portmapping_info(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/node-info" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -62,8 +61,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_04_rdm_portmapping_DEG1_TTP_TXRX(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG1-TTP-TXRX" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1/mapping/DEG1-TTP-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -72,8 +70,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG2-TTP-TXRX" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1/mapping/DEG2-TTP-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -84,8 +81,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_06_rdm_portmapping_SRG1_PP3_TXRX(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG1-PP3-TXRX" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1/mapping/SRG1-PP3-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -94,8 +90,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_07_rdm_portmapping_SRG3_PP1_TXRX(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG3-PP1-TXRX" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1/mapping/SRG3-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -117,8 +112,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_10_xpdr_portmapping_info(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/node-info" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -131,8 +125,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_11_xpdr_portmapping_NETWORK1(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -144,8 +137,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_12_xpdr_portmapping_NETWORK2(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK2" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -157,8 +149,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_13_xpdr_portmapping_CLIENT1(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -171,8 +162,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['mapping']) def test_14_xpdr_portmapping_CLIENT2(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT2" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -198,8 +188,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_17_xpdr_device_not_connected(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1") self.assertEqual(response.status_code, requests.codes.not_found) res = response.json() self.assertIn( @@ -221,8 +210,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_20_rdm_device_not_connected(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1") self.assertEqual(response.status_code, requests.codes.not_found) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py b/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py index bb5f5bb27..7332ebe38 100644 --- a/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py @@ -40,8 +40,7 @@ class TransportPCERendererTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_03_rdm_portmapping(self): - url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADM-A1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -54,8 +53,7 @@ class TransportPCERendererTesting(unittest.TestCase): res['nodes'][0]['mapping']) def test_04_xpdr_portmapping(self): - url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1" - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDR-A1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( diff --git a/tests/transportpce_tests/2.2.1/test_topoPortMapping.py b/tests/transportpce_tests/2.2.1/test_topoPortMapping.py index fda216c2f..d60aac9b4 100644 --- a/tests/transportpce_tests/2.2.1/test_topoPortMapping.py +++ b/tests/transportpce_tests/2.2.1/test_topoPortMapping.py @@ -51,8 +51,7 @@ class TransportPCEtesting(unittest.TestCase): print("nodeId={}".format(nodeId)) nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1] print("nodeMapId={}".format(nodeMapId)) - urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId - responseMapList = test_utils.get_request(urlMapList) + responseMapList = test_utils.portmapping_request(nodeMapId) resMapList = responseMapList.json() nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul @@ -61,8 +60,7 @@ class TransportPCEtesting(unittest.TestCase): for j in range(0, nbTp): tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id'] if((not "CP" in tpId) and (not "CTP" in tpId)): - urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId - responseMap = test_utils.get_request(urlMap) + responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId) self.assertEqual(responseMap.status_code, requests.codes.ok) if(responseMap.status_code == requests.codes.ok): nbMapCurrent += 1 diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py index 5a063b4d0..2abe3ee76 100644 --- a/tests/transportpce_tests/common/test_utils.py +++ b/tests/transportpce_tests/common/test_utils.py @@ -39,6 +39,7 @@ URL_CONFIG_ORDM_TOPO = "{}/config/ietf-network:networks/network/openroadm-topolo URL_CONFIG_OTN_TOPO = "{}/config/ietf-network:networks/network/otn-topology/" URL_CONFIG_CLLI_NET = "{}/config/ietf-network:networks/network/clli-network/" URL_CONFIG_ORDM_NET = "{}/config/ietf-network:networks/network/openroadm-network/" +URL_PORTMAPPING = "{}/config/transportpce-portmapping:network/nodes/" TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'} TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'} @@ -308,6 +309,11 @@ def del_node_request(node: str): return delete_request(url) +def portmapping_request(suffix: str): + url = URL_PORTMAPPING + suffix + return get_request(url) + + def shutdown_process(process): if process is not None: for child in psutil.Process(process.pid).children():