From e72024a00f020ebfe4ade77016f4add2cb238e3f Mon Sep 17 00:00:00 2001 From: Gilles Thouenon Date: Thu, 25 Aug 2022 11:13:21 +0200 Subject: [PATCH] Refactor test_utils_rfc8040.py portmapping methods JIRA: TRNSPRTPCE-567 Co-authored-by: Guillaume Lambert Signed-off-by: Gilles Thouenon Signed-off-by: Guillaume Lambert Change-Id: Ibe593c9ecc5ea2f863b16ed87629b26f4f5ab93c --- .../1.2.1/test01_portmapping.py | 26 +++---- .../1.2.1/test02_topo_portmapping.py | 4 +- .../test04_renderer_service_path_nominal.py | 4 +- .../2.2.1/test01_portmapping.py | 64 ++++++++--------- .../2.2.1/test02_topo_portmapping.py | 6 +- .../2.2.1/test05_flex_grid.py | 8 +-- .../test06_renderer_service_path_nominal.py | 4 +- .../2.2.1/test07_otn_renderer.py | 8 +-- .../7.1/test01_portmapping.py | 22 +++--- .../7.1/test02_otn_renderer.py | 26 +++---- .../7.1/test03_renderer_or_modes.py | 20 +++--- .../common/test_utils_rfc8040.py | 71 +++++-------------- 12 files changed, 112 insertions(+), 151 deletions(-) diff --git a/tests/transportpce_tests/1.2.1/test01_portmapping.py b/tests/transportpce_tests/1.2.1/test01_portmapping.py index 1d7a5c24b..5553da737 100644 --- a/tests/transportpce_tests/1.2.1/test01_portmapping.py +++ b/tests/transportpce_tests/1.2.1/test01_portmapping.py @@ -56,7 +56,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_03_rdm_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("ROADMA01") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'rdm', @@ -69,7 +69,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_04_rdm_portmapping_DEG1_TTP_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADMA01", "DEG1-TTP-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", "mapping", "DEG1-TTP-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0', @@ -78,7 +78,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_05_rdm_portmapping_SRG1_PP7_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADMA01", "SRG1-PP7-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", "mapping", "SRG1-PP7-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0', @@ -87,7 +87,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_06_rdm_portmapping_SRG3_PP1_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADMA01", "SRG3-PP1-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", "mapping", "SRG3-PP1-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0', @@ -106,7 +106,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_09_xpdr_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDRA01") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'xpdr', @@ -119,7 +119,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_10_xpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', @@ -130,7 +130,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_11_xpdr_portmapping_NETWORK2(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-NETWORK2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-NETWORK2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET', @@ -141,7 +141,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_12_xpdr_portmapping_CLIENT1(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-CLIENT1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C1', @@ -153,7 +153,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_13_xpdr_portmapping_CLIENT2(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-CLIENT2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-CLIENT2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C2', @@ -165,7 +165,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_14_xpdr_portmapping_CLIENT3(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-CLIENT3") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-CLIENT3") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C3', 'xponder-type': 'tpdr', @@ -177,7 +177,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_15_xpdr_portmapping_CLIENT4(self): - response = test_utils_rfc8040.portmapping_request("XPDRA01", "XPDR1-CLIENT4") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "mapping", "XPDR1-CLIENT4") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C4', 'xponder-type': 'tpdr', @@ -200,7 +200,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_18_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDRA01") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", "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') @@ -220,7 +220,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_21_rdm_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("ROADMA01") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", "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') diff --git a/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py b/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py index 653ae8455..1f902c3e7 100644 --- a/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py +++ b/tests/transportpce_tests/1.2.1/test02_topo_portmapping.py @@ -57,12 +57,12 @@ class TransportPCEtesting(unittest.TestCase): for node in resTopo['network'][0]['node']: nodeId = node['node-id'] nodeMapId = nodeId.split("-")[0] - response = test_utils_rfc8040.get_portmapping_node_info(nodeMapId) + response = test_utils_rfc8040.get_portmapping_node_attr(nodeMapId, "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) for tp in node['ietf-network-topology:termination-point']: tpId = tp['tp-id'] if (not "CP" in tpId) and (not "CTP" in tpId): - response2 = test_utils_rfc8040.portmapping_request(nodeMapId, tpId) + response2 = test_utils_rfc8040.get_portmapping_node_attr(nodeMapId, "mapping", tpId) self.assertEqual(response2['status_code'], requests.codes.ok) # Disconnect the ROADMA diff --git a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py index 40f6bb961..b95ed2539 100644 --- a/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py @@ -54,7 +54,7 @@ class TransportPCERendererTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) def test_03_rdm_portmapping(self): - response = test_utils_rfc8040.get_portmapping("ROADMA01") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADMA01", None, None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0', @@ -68,7 +68,7 @@ class TransportPCERendererTesting(unittest.TestCase): response['nodes'][0]['mapping']) def test_04_xpdr_portmapping(self): - response = test_utils_rfc8040.get_portmapping("XPDRA01") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDRA01", None, None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', diff --git a/tests/transportpce_tests/2.2.1/test01_portmapping.py b/tests/transportpce_tests/2.2.1/test01_portmapping.py index a77584ddd..645db1c41 100644 --- a/tests/transportpce_tests/2.2.1/test01_portmapping.py +++ b/tests/transportpce_tests/2.2.1/test01_portmapping.py @@ -56,7 +56,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): self.assertEqual(response['connection-status'], 'connected') def test_03_rdm_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'rdm', @@ -68,7 +68,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['node-info']) def test_04_rdm_portmapping_DEG1_TTP_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADM-A1", "DEG1-TTP-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "mapping", "DEG1-TTP-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0', @@ -77,7 +77,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self): - response = test_utils_rfc8040.portmapping_request("ROADM-A1", "DEG2-TTP-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "mapping", "DEG2-TTP-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0', @@ -88,7 +88,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_06_rdm_portmapping_SRG1_PP3_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADM-A1", "SRG1-PP3-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "mapping", "SRG1-PP3-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0', @@ -97,7 +97,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_07_rdm_portmapping_SRG3_PP1_TXRX(self): - response = test_utils_rfc8040.portmapping_request("ROADM-A1", "SRG3-PP1-TXRX") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "mapping", "SRG3-PP1-TXRX") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0', @@ -115,7 +115,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): self.assertEqual(response['connection-status'], 'connected') def test_10_xpdr_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'xpdr', @@ -127,7 +127,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['node-info']) def test_11_xpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A1", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'], @@ -139,7 +139,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_12_xpdr_portmapping_XPDR2_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A1", "XPDR1-NETWORK2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-NETWORK2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'], @@ -151,7 +151,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_13_xpdr_portmapping_XPDR1_CLIENT1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A1", "XPDR1-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-CLIENT1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'], @@ -164,7 +164,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_14_xpdr_portmapping_XPDR1_CLIENT2(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A1", "XPDR1-CLIENT2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-CLIENT2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'], @@ -186,7 +186,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): self.assertEqual(response['connection-status'], 'connected') def test_17_spdr_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("SPDR-SA1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'xpdr', @@ -198,26 +198,26 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['node-info']) def test_18_spdr_switching_pool_1(self): - response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual("blocking", - response['switching_pool'][0]['switching-pool-type']) + response['switching-pool-lcp'][0]['switching-pool-type']) self.assertEqual(4, - len(response['switching_pool'][0]['non-blocking-list'])) + len(response['switching-pool-lcp'][0]['non-blocking-list'])) self.assertIn( {'nbl-number': 11, 'lcp-list': ['XPDR1-CLIENT1', 'XPDR1-NETWORK1'], 'interconnect-bandwidth-unit': 1000000000, 'interconnect-bandwidth': 0}, - response['switching_pool'][0]['non-blocking-list']) + response['switching-pool-lcp'][0]['non-blocking-list']) def test_19_spdr_switching_pool_2(self): - response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "2") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual("non-blocking", - response['switching_pool'][0]['switching-pool-type']) + response['switching-pool-lcp'][0]['switching-pool-type']) self.assertEqual(1, - len(response['switching_pool'][0]['non-blocking-list'])) + len(response['switching-pool-lcp'][0]['non-blocking-list'])) self.assertIn( {'nbl-number': 2, 'lcp-list': ['XPDR2-NETWORK2', 'XPDR2-NETWORK3', 'XPDR2-CLIENT1', @@ -225,29 +225,29 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'XPDR2-CLIENT4', 'XPDR2-NETWORK4'], 'interconnect-bandwidth-unit': 1000000000, 'interconnect-bandwidth': 0}, - response['switching_pool'][0]['non-blocking-list']) + response['switching-pool-lcp'][0]['non-blocking-list']) def test_20_spdr_switching_pool_3(self): - response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "3") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "3") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual("blocking", - response['switching_pool'][0]['switching-pool-type']) + response['switching-pool-lcp'][0]['switching-pool-type']) self.assertEqual(4, - len(response['switching_pool'][0]['non-blocking-list'])) + len(response['switching-pool-lcp'][0]['non-blocking-list'])) self.assertIn( {'nbl-number': 83, 'interconnect-bandwidth': 0, 'interconnect-bandwidth-unit': 1000000000, 'lcp-list': ['XPDR3-NETWORK1', 'XPDR3-CLIENT3']}, - response['switching_pool'][0]['non-blocking-list']) + response['switching-pool-lcp'][0]['non-blocking-list']) def test_21_spdr_portmapping_mappings(self): - response = test_utils_rfc8040.get_portmapping("SPDR-SA1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", None, None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual(18, len(response['nodes'][0]['mapping'])) def test_22_spdr_portmapping_XPDR1_CLIENT1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-CLIENT1") self.assertEqual(response['status_code'], requests.codes.ok) expected_subset_response = { "logical-connection-point": "XPDR1-CLIENT1", @@ -266,7 +266,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list) def test_23_spdr_portmapping_XPDR1_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {"logical-connection-point": "XPDR1-NETWORK1", @@ -282,7 +282,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_24_spdr_portmapping_XPDR2_CLIENT2(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-CLIENT2") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR2-CLIENT2") self.assertEqual(response['status_code'], requests.codes.ok) expected_subset_response = { 'logical-connection-point': 'XPDR2-CLIENT2', @@ -301,7 +301,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list) def test_25_spdr_portmapping_XPDR2_NETWORK2(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-NETWORK2") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR2-NETWORK2") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {"logical-connection-point": "XPDR2-NETWORK2", @@ -317,7 +317,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_26_spdr_portmapping_XPDR3_CLIENT3(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-CLIENT3") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR3-CLIENT3") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {"logical-connection-point": "XPDR3-CLIENT3", @@ -333,7 +333,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): response['mapping']) def test_27_spdr_portmapping_XPDR3_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR3-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {"logical-connection-point": "XPDR3-NETWORK1", @@ -373,7 +373,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_32_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", "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') @@ -393,7 +393,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_35_rdm_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", "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') diff --git a/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py b/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py index 586b882f1..ec1688580 100644 --- a/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py +++ b/tests/transportpce_tests/2.2.1/test02_topo_portmapping.py @@ -60,15 +60,15 @@ class TransportPCEtesting(unittest.TestCase): print("nodeId={}".format(nodeId)) nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1] print("nodeMapId={}".format(nodeMapId)) - response = test_utils_rfc8040.get_portmapping_node_info(nodeMapId) + response = test_utils_rfc8040.get_portmapping_node_attr(nodeMapId, "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) - responseMapList = test_utils_rfc8040.get_portmapping(nodeMapId) + responseMapList = test_utils_rfc8040.get_portmapping_node_attr(nodeMapId, None, None) nbMappings = len(responseMapList['nodes'][0]['mapping']) - nbMapCumul nbMapCurrent = 0 for tp in node['ietf-network-topology:termination-point']: tpId = tp['tp-id'] if (not "CP" in tpId) and (not "CTP" in tpId): - responseMap = test_utils_rfc8040.portmapping_request(nodeMapId, tpId) + responseMap = test_utils_rfc8040.get_portmapping_node_attr(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/2.2.1/test05_flex_grid.py b/tests/transportpce_tests/2.2.1/test05_flex_grid.py index 2d5674bc5..0d3c13b0f 100644 --- a/tests/transportpce_tests/2.2.1/test05_flex_grid.py +++ b/tests/transportpce_tests/2.2.1/test05_flex_grid.py @@ -55,7 +55,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_03_rdm_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("ROADM-D1") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-D1", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'rdm', @@ -69,7 +69,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_04_rdm_deg1_lcp(self): # pylint: disable=line-too-long - response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "DEG1-TTP") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "DEG1-TTP") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn(response['mc-capabilities'], [[{'mc-node-name': 'DEG1-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}], @@ -78,7 +78,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_05_rdm_deg2_lcp(self): # pylint: disable=line-too-long - response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "DEG2-TTP") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "DEG2-TTP") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn(response['mc-capabilities'], [[{'mc-node-name': 'DEG2-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}], @@ -87,7 +87,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_06_rdm_srg1_lcp(self): # pylint: disable=line-too-long - response = test_utils_rfc8040.portmapping_mc_capa_request("ROADM-D1", "SRG1-PP") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "SRG1-PP") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn(response['mc-capabilities'], [[{'mc-node-name': 'SRG1-PP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}], diff --git a/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py b/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py index dc7680acd..d5f6889d2 100644 --- a/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py +++ b/tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py @@ -49,7 +49,7 @@ class TransportPCERendererTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) def test_03_rdm_portmapping(self): - response = test_utils_rfc8040.get_portmapping("ROADM-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("ROADM-A1", None, None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0', @@ -63,7 +63,7 @@ class TransportPCERendererTesting(unittest.TestCase): response['nodes'][0]['mapping']) def test_04_xpdr_portmapping(self): - response = test_utils_rfc8040.get_portmapping("XPDR-A1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A1", None, None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'], diff --git a/tests/transportpce_tests/2.2.1/test07_otn_renderer.py b/tests/transportpce_tests/2.2.1/test07_otn_renderer.py index 096fddb8e..228b91e23 100644 --- a/tests/transportpce_tests/2.2.1/test07_otn_renderer.py +++ b/tests/transportpce_tests/2.2.1/test07_otn_renderer.py @@ -65,7 +65,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response['connection-status'], 'connected') def test_02_get_portmapping_CLIENT4(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-CLIENT4") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-CLIENT4") self.assertEqual(response['status_code'], requests.codes.ok) res_mapping = response['mapping'][0] self.assertEqual('CP1-SFP4-P1', res_mapping['supporting-port']) @@ -81,7 +81,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertIn('org-openroadm-port-types:if-10GE', res_mapping['supported-interface-capability']) def test_03_get_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( self.NETWORK1_CHECK_DICT, @@ -112,7 +112,7 @@ class TransportPCEtesting(unittest.TestCase): 'och-interface-id': ['XPDR1-NETWORK1-761:768']}, response['output']['node-interface']) def test_05_get_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK1_CHECK_DICT["supporting-otu4"] = "XPDR1-NETWORK1-OTU" self.assertIn( @@ -177,7 +177,7 @@ class TransportPCEtesting(unittest.TestCase): 'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, response['output']['node-interface']) def test_09_get_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK1_CHECK_DICT["supporting-odu4"] = "XPDR1-NETWORK1-ODU4" self.NETWORK1_CHECK_DICT["supporting-otu4"] = "XPDR1-NETWORK1-OTU" diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py index a7bf2b949..1f3f7791f 100644 --- a/tests/transportpce_tests/7.1/test01_portmapping.py +++ b/tests/transportpce_tests/7.1/test01_portmapping.py @@ -58,7 +58,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check node info in the port-mappings def test_03_xpdr_portmapping_info(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "node-info", None) self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( {'node-type': 'xpdr', @@ -71,7 +71,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check the if-capabilities and the other details for network def test_04_tpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': @@ -90,7 +90,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['mapping']) def test_05_tpdr_portmapping_CLIENT1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-CLIENT1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'], @@ -110,7 +110,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check the port-mapping for the switch-client and switch-network port-quals def test_06_mpdr_portmapping_NETWORK1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn( {'supported-interface-capability': @@ -128,7 +128,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['mapping']) def test_07_mpdr_portmapping_CLIENT1(self): - res = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-CLIENT1") + res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "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']) @@ -150,24 +150,24 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Added test to check mc-capability-profile for a transponder def test_08_check_mccapprofile(self): - res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "XPDR-mcprofile") + res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mc-capabilities", "XPDR-mcprofile") self.assertEqual(res['status_code'], requests.codes.ok) self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile') 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_mpdr_switching_pool(self): - response = test_utils_rfc8040.portmapping_switching_pool_request("XPDR-A2", "1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "switching-pool-lcp", "1") self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual("blocking", - response['switching_pool'][0]['switching-pool-type']) + response['switching-pool-lcp'][0]['switching-pool-type']) self.assertEqual(2, - len(response['switching_pool'][0]['non-blocking-list'])) + len(response['switching-pool-lcp'][0]['non-blocking-list'])) self.assertIn( {'nbl-number': 2, 'interconnect-bandwidth': 0, 'lcp-list': ['XPDR2-NETWORK1', 'XPDR2-CLIENT2']}, - response['switching_pool'][0]['non-blocking-list']) + response['switching-pool-lcp'][0]['non-blocking-list']) def test_10_xpdr_device_disconnection(self): response = test_utils_rfc8040.unmount_device("XPDR-A2") @@ -182,7 +182,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_12_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + response = test_utils_rfc8040.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') diff --git a/tests/transportpce_tests/7.1/test02_otn_renderer.py b/tests/transportpce_tests/7.1/test02_otn_renderer.py index c4a0ecdb4..4bceb0993 100644 --- a/tests/transportpce_tests/7.1/test02_otn_renderer.py +++ b/tests/transportpce_tests/7.1/test02_otn_renderer.py @@ -91,7 +91,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response['output']['node-interface']) def test_03_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-otucn"] = "XPDR2-NETWORK1-OTUC2" self.assertIn( @@ -182,7 +182,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'odu-interface-id': ['XPDR2-NETWORK1-ODUC2']}, response['output']['node-interface']) def test_08_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-oducn"] = "XPDR2-NETWORK1-ODUC2" self.assertIn( @@ -392,7 +392,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check if port-mapping data is updated, where the supporting-oducn is deleted def test_21a_check_no_oduc2(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-oducn"]) # 1f) Delete OTUC2 device interfaces @@ -431,7 +431,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_25a_check_no_otuc2(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) # 2a) create a OTUC3 device renderer @@ -463,7 +463,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(sorted(response['output']['node-interface'][0]['och-interface-id']), expected_sorted_list) def test_27_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-otucn"] = "XPDR2-NETWORK1-OTUC3" self.assertIn( @@ -555,7 +555,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'odu-interface-id': ['XPDR2-NETWORK1-ODUC3']}, response['output']['node-interface']) def test_32_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-oducn"] = "XPDR2-NETWORK1-ODUC3" self.assertIn( @@ -617,7 +617,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_35a_check_no_oduc3(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-oducn"]) # 2f) Delete OTUC3 device interfaces @@ -655,7 +655,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_39a_check_no_otuc3(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) # 3a) create a OTUC4 device renderer @@ -687,7 +687,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(sorted(response['output']['node-interface'][0]['och-interface-id']), expected_sorted_list) def test_41_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-otucn"] = "XPDR2-NETWORK1-OTUC4" self.assertIn( @@ -779,7 +779,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'odu-interface-id': ['XPDR2-NETWORK1-ODUC4']}, response['output']['node-interface']) def test_46_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response['status_code'], requests.codes.ok) self.NETWORK2_CHECK_DICT["supporting-oducn"] = "XPDR2-NETWORK1-ODUC4" self.assertIn( @@ -841,7 +841,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_49a_check_no_oduc4(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-oducn"]) # 3f) Delete OTUC4 device interfaces @@ -879,7 +879,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_53a_check_no_otuc4(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) # Disconnect the XPDR @@ -896,7 +896,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): 'Request could not be completed because the relevant data model content does not exist') def test_56_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + response = test_utils_rfc8040.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') diff --git a/tests/transportpce_tests/7.1/test03_renderer_or_modes.py b/tests/transportpce_tests/7.1/test03_renderer_or_modes.py index 3d8ee82a0..ecb41ef93 100644 --- a/tests/transportpce_tests/7.1/test03_renderer_or_modes.py +++ b/tests/transportpce_tests/7.1/test03_renderer_or_modes.py @@ -75,14 +75,14 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check the correct capabilities for the client ports (if-100GE, if-100GE-ODU4, # if-OCH-OTU4-ODU4) def test_02_check_client_capabilities(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR3-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR3-CLIENT1") self.assertEqual(response["status_code"], requests.codes.ok) self.assertEqual( self.CLIENT_CAPABILITIES, sorted(response["mapping"][0]["supported-interface-capability"])) def test_03_check_client_capabilities(self): - response = test_utils_rfc8040.portmapping_request("XPDR-C2", "XPDR3-CLIENT1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-C2", "mapping", "XPDR3-CLIENT1") self.assertEqual(response["status_code"], requests.codes.ok) self.assertEqual( self.CLIENT_CAPABILITIES, @@ -162,7 +162,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): for x in response["output"]["node-interface"][1].keys()}) def test_05_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR3-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR3-NETWORK1") self.assertEqual(response["status_code"], requests.codes.ok) self.NETWORK1_CHECK_DICT["supporting-otucn"] = "XPDR3-NETWORK1-OTUC1" self.NETWORK1_CHECK_DICT["lcp-hash-val"] = "FDvaQIf2Z08=" @@ -172,7 +172,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): response["mapping"]) def test_06_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-C2", "XPDR3-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-C2", "mapping", "XPDR3-NETWORK1") self.assertEqual(response["status_code"], requests.codes.ok) self.NETWORK1_CHECK_DICT["supporting-otucn"] = "XPDR3-NETWORK1-OTUC1" self.NETWORK1_CHECK_DICT["lcp-hash-val"] = "AJpkaVmZKJk5" @@ -519,11 +519,11 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): # Check if port-mapping data is updated, where the supporting-otucn is deleted def test_26_check_no_otuc1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR3-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR3-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) def test_27_check_no_otuc1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-C2", "XPDR3-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-C2", "mapping", "XPDR3-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) def test_28_check_no_interface_otsig(self): @@ -605,7 +605,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.NETWORK1_CHECK_DICT["lcp-hash-val"] = "LY9PxYJqUbw=" def test_33_get_portmapping_network1(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertEqual(response["status_code"], requests.codes.ok) self.NETWORK1_CHECK_DICT["supporting-otucn"] = "XPDR2-NETWORK1-OTUC2" del self.NETWORK1_CHECK_DICT["connection-map-lcp"] @@ -725,7 +725,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): self.assertEqual(response["status_code"], requests.codes.conflict) def test_41_check_no_otuc2(self): - response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1") self.assertRaises(KeyError, lambda: response["supporting-otucn"]) # Disconnect the XPDR @@ -742,7 +742,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): "Request could not be completed because the relevant data model content does not exist") def test_44_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2") + response = test_utils_rfc8040.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") @@ -762,7 +762,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase): "Request could not be completed because the relevant data model content does not exist") def test_47_xpdr_device_not_connected(self): - response = test_utils_rfc8040.get_portmapping_node_info("XPDR-C2") + response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-C2", "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") diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index 17ef51d23..18d2a7d3e 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -375,67 +375,28 @@ def del_node_attribute_request(node: str, attribute: str, attribute_value: str): # -def get_portmapping(node: str): +def get_portmapping_node_attr(node: str, attr: str, value: str): + # pylint: disable=consider-using-f-string url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}', 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}'} - response = get_request(url[RESTCONF_VERSION].format('{}', node)) - res = response.json() - return_key = {'rfc8040': 'transportpce-portmapping:nodes', - 'draft-bierman02': 'nodes'} - nodes = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'nodes': nodes} - - -def get_portmapping_node_info(node: str): - url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/node-info', - 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/node-info'} - response = get_request(url[RESTCONF_VERSION].format('{}', node)) + target_url = url[RESTCONF_VERSION].format('{}', node) + if attr is not None: + target_url = (target_url + '/{}').format('{}', attr) + if value is not None: + suffix = {'rfc8040': '={}', 'draft-bierman02': '/{}'} + target_url = (target_url + suffix[RESTCONF_VERSION]).format('{}', value) + else: + attr = 'nodes' + response = get_request(target_url) res = response.json() - return_key = {'rfc8040': 'transportpce-portmapping:node-info', - 'draft-bierman02': 'node-info'} + return_key = {'rfc8040': 'transportpce-portmapping:' + attr, + 'draft-bierman02': attr} if return_key[RESTCONF_VERSION] in res.keys(): - node_info = res[return_key[RESTCONF_VERSION]] + return_output = res[return_key[RESTCONF_VERSION]] else: - node_info = res['errors']['error'][0] - return {'status_code': response.status_code, - 'node-info': node_info} - - -def portmapping_request(node: str, mapping: str): - url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/mapping={}', - 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/mapping/{}'} - response = get_request(url[RESTCONF_VERSION].format('{}', node, mapping)) - res = response.json() - return_key = {'rfc8040': 'transportpce-portmapping:mapping', - 'draft-bierman02': 'mapping'} - mapping = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'mapping': mapping} - - -def portmapping_switching_pool_request(node: str, switching_pool: str): - url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/switching-pool-lcp={}', - 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/switching-pool-lcp/{}'} - response = get_request(url[RESTCONF_VERSION].format('{}', node, switching_pool)) - res = response.json() - return_key = {'rfc8040': 'transportpce-portmapping:switching-pool-lcp', - 'draft-bierman02': 'switching-pool-lcp'} - switching_pool = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'switching_pool': switching_pool} - - -def portmapping_mc_capa_request(node: str, mc_capa: str): - url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/mc-capabilities={}', - 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/mc-capabilities/{}'} - response = get_request(url[RESTCONF_VERSION].format('{}', node, mc_capa)) - res = response.json() - return_key = {'rfc8040': 'transportpce-portmapping:mc-capabilities', - 'draft-bierman02': 'mc-capabilities'} - capabilities = res[return_key[RESTCONF_VERSION]] + return_output = res['errors']['error'][0] return {'status_code': response.status_code, - 'mc-capabilities': capabilities} + attr: return_output} # # Topology operations -- 2.36.6