From f21439dd643bb19d0d83e6f7e37addb4c3b39c6e Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Fri, 18 Mar 2022 18:37:35 +0100 Subject: [PATCH] Refactor func tests transportpce api rpc calls 2/2 JIRA: TRNSPRTPCE-567 Signed-off-by: guillaume.lambert Change-Id: I74bbe8b19118025f505d3c86430bc9185faebe64 --- .../1.2.1/test03_topology.py | 18 ++--- tests/transportpce_tests/1.2.1/test05_olm.py | 72 +++++++++++-------- .../2.2.1/test03_topology.py | 18 ++--- tests/transportpce_tests/2.2.1/test09_olm.py | 72 +++++++++++-------- .../common/test_utils_rfc8040.py | 30 -------- 5 files changed, 108 insertions(+), 102 deletions(-) diff --git a/tests/transportpce_tests/1.2.1/test03_topology.py b/tests/transportpce_tests/1.2.1/test03_topology.py index f9c4d1f27..1600fe200 100644 --- a/tests/transportpce_tests/1.2.1/test03_topology.py +++ b/tests/transportpce_tests/1.2.1/test03_topology.py @@ -253,17 +253,19 @@ class TransportPCETopologyTesting(unittest.TestCase): # Connect the tail XPDRA to ROADMA and vice versa def test_10_connect_tail_xpdr_rdm(self): # Connect the tail: XPDRA to ROADMA - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_11_connect_tail_rdm_xpdr(self): # Connect the tail: ROADMA to XPDRA - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_12_getLinks_OpenRoadmTopology(self): response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config') diff --git a/tests/transportpce_tests/1.2.1/test05_olm.py b/tests/transportpce_tests/1.2.1/test05_olm.py index ca5306bd3..115b1f7f9 100644 --- a/tests/transportpce_tests/1.2.1/test05_olm.py +++ b/tests/transportpce_tests/1.2.1/test05_olm.py @@ -65,36 +65,50 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) def test_05_connect_xprdA_to_roadmA(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_06_connect_roadmA_to_xpdrA(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_07_connect_xprdC_to_roadmC(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_08_connect_roadmC_to_xpdrC(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_09_create_OTS_ROADMA(self): - response = test_utils_rfc8040.device_renderer_create_ots_oms_request('ROADMA01', 'DEG1-TTP-TXRX') - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-device-renderer', 'create-ots-oms', + { + 'node-id': 'ROADMA01', + 'logical-connection-point': 'DEG1-TTP-TXRX' + }) + self.assertEqual(response['status_code'], requests.codes.ok) def test_10_create_OTS_ROADMC(self): - response = test_utils_rfc8040.device_renderer_create_ots_oms_request('ROADMC01', 'DEG2-TTP-TXRX') - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-device-renderer', 'create-ots-oms', + { + 'node-id': 'ROADMC01', + 'logical-connection-point': 'DEG2-TTP-TXRX' + }) + self.assertEqual(response['status_code'], requests.codes.ok) def test_11_get_PM_ROADMA(self): response = test_utils_rfc8040.transportpce_api_rpc_request( @@ -475,16 +489,18 @@ class TransportOlmTesting(unittest.TestCase): #"""to test case where SRG where the xpdr is connected to has no optical range data""" def test_31_connect_xprdA_to_roadmA(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_32_connect_roadmA_to_xpdrA(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2', - 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2', + 'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_33_servicePath_create_AToZ(self): response = test_utils_rfc8040.transportpce_api_rpc_request( diff --git a/tests/transportpce_tests/2.2.1/test03_topology.py b/tests/transportpce_tests/2.2.1/test03_topology.py index 6a263a6c7..88ae14f69 100644 --- a/tests/transportpce_tests/2.2.1/test03_topology.py +++ b/tests/transportpce_tests/2.2.1/test03_topology.py @@ -255,16 +255,18 @@ class TransportPCEtesting(unittest.TestCase): # Connect the tail XPDRA to ROADMA and vice versa def test_10_connect_tail_xpdr_rdm(self): # Connect the tail: XPDRA to ROADMA - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_11_connect_tail_rdm_xpdr(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_12_getLinks_OpenRoadmTopology(self): # pylint: disable=redundant-unittest-assert diff --git a/tests/transportpce_tests/2.2.1/test09_olm.py b/tests/transportpce_tests/2.2.1/test09_olm.py index 85467aed1..b8e3c9d93 100644 --- a/tests/transportpce_tests/2.2.1/test09_olm.py +++ b/tests/transportpce_tests/2.2.1/test09_olm.py @@ -65,36 +65,50 @@ class TransportOlmTesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201) def test_05_connect_xprdA_to_roadmA(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_06_connect_roadmA_to_xpdrA(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_07_connect_xprdC_to_roadmC(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_08_connect_roadmC_to_xpdrC(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', - 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1', + 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_09_create_OTS_ROADMA(self): - response = test_utils_rfc8040.device_renderer_create_ots_oms_request('ROADM-A1', 'DEG1-TTP-TXRX') - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-device-renderer', 'create-ots-oms', + { + 'node-id': 'ROADM-A1', + 'logical-connection-point': 'DEG1-TTP-TXRX' + }) + self.assertEqual(response['status_code'], requests.codes.ok) def test_10_create_OTS_ROADMC(self): - response = test_utils_rfc8040.device_renderer_create_ots_oms_request('ROADM-C1', 'DEG2-TTP-TXRX') - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-device-renderer', 'create-ots-oms', + { + 'node-id': 'ROADM-C1', + 'logical-connection-point': 'DEG2-TTP-TXRX' + }) + self.assertEqual(response['status_code'], requests.codes.ok) def test_11_get_PM_ROADMA(self): response = test_utils_rfc8040.transportpce_api_rpc_request( @@ -460,16 +474,18 @@ class TransportOlmTesting(unittest.TestCase): #"""to test case where SRG where the xpdr is connected to has no optical range data""" def test_31_connect_xprdA_to_roadmA(self): - response = test_utils_rfc8040.connect_xpdr_to_rdm_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-xpdr-rdm-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_32_connect_roadmA_to_xpdrA(self): - response = test_utils_rfc8040.connect_rdm_to_xpdr_request( - {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', - 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}) - self.assertEqual(response.status_code, requests.codes.ok) + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-networkutils', 'init-rdm-xpdr-links', + {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2', + 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}}) + self.assertEqual(response['status_code'], requests.codes.ok) def test_33_servicePath_create_AToZ(self): response = test_utils_rfc8040.transportpce_api_rpc_request( diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index d83c1d9df..785f2a6d8 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -525,36 +525,6 @@ def prepend_dict_keys(input_dict: dict, prefix: str): return return_dict -def connect_xpdr_to_rdm_request(payload: dict): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:') - else: - data = {'input': {'links-input': payload}} - return post_request(url, data) - - -def connect_rdm_to_xpdr_request(payload: dict): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:') - else: - data = {'input': {'links-input': payload}} - return post_request(url, data) - - -def device_renderer_create_ots_oms_request(nodeid: str, lcp: str): - url = "{}/operations/transportpce-device-renderer:create-ots-oms" - payload = { - 'node-id': nodeid, - 'logical-connection-point': lcp} - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:') - else: - data = {'input': payload} - return post_request(url, data) - - def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict): # pylint: disable=consider-using-f-string url = "{}/operations/{}:{}".format('{}', api_module, rpc) -- 2.36.6