From 72e2991db41d42bd3bb70dcc99774cdb1fd26f68 Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Wed, 6 Jul 2022 16:58:23 +0200 Subject: [PATCH] Migrate OTN E2E functional tests to RFC8040 step 2 JIRA: TRNSPRTPCE-567 Signed-off-by: guillaume.lambert Change-Id: I73f0c083b6b1c9aebb9ca742e8c12bfb1a85263f --- .../2.2.1/test11_otn_end2end.py | 80 +++++++++---------- .../common/test_utils_rfc8040.py | 18 +++++ 2 files changed, 54 insertions(+), 44 deletions(-) diff --git a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py index e6649a1fc..2dfe2448e 100644 --- a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py +++ b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py @@ -351,25 +351,25 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response['status_code'], requests.codes.conflict) def test_19_check_openroadm_topo_spdra(self): - response = test_utils.get_ordm_topo_request("node/SPDR-SA1-XPDR1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - ele = res['node'][0]['ietf-network-topology:termination-point'][0] + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'SPDR-SA1-XPDR1', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) + ele = response['node']['ietf-network-topology:termination-point'][0] self.assertEqual('XPDR1-NETWORK1', ele['tp-id']) - self.assertEqual({'frequency': 196.1, - 'width': 40}, - ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) - time.sleep(3) + self.assertEqual( + 196.1, + float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency'])) + self.assertEqual( + 40, + float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width'])) def test_20_check_openroadm_topo_ROADMA_SRG(self): - response = test_utils.get_ordm_topo_request("node/ROADM-A1-SRG1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( - res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) + response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") - liste_tp = res['node'][0]['ietf-network-topology:termination-point'] + liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'SRG1-PP1-TXRX': freq_map = base64.b64decode( @@ -378,17 +378,15 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") if ele['tp-id'] == 'SRG1-PP2-TXRX': self.assertNotIn('avail-freq-maps', dict.keys(ele)) - time.sleep(3) def test_21_check_openroadm_topo_ROADMA_DEG(self): - response = test_utils.get_ordm_topo_request("node/ROADM-A1-DEG2") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( - res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) + response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") - liste_tp = res['node'][0]['ietf-network-topology:termination-point'] + liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'DEG2-CTP-TXRX': freq_map = base64.b64decode( @@ -400,7 +398,6 @@ class TransportPCEtesting(unittest.TestCase): ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map']) freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") - time.sleep(3) def test_22_check_otn_topo_otu4_links(self): response = test_utils.get_otn_topo_request() @@ -943,24 +940,21 @@ class TransportPCEtesting(unittest.TestCase): self.assertNotIn('ietf-network-topology:link', res['network'][0]) def test_59_check_openroadm_topo_spdra(self): - response = test_utils.get_ordm_topo_request("node/SPDR-SA1-XPDR1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - tp = res['node'][0]['ietf-network-topology:termination-point'][0] + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'SPDR-SA1-XPDR1', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) + tp = response['node']['ietf-network-topology:termination-point'][0] self.assertEqual('XPDR1-NETWORK1', tp['tp-id']) self.assertNotIn('wavelength', dict.keys( tp['org-openroadm-network-topology:xpdr-network-attributes'])) - time.sleep(3) def test_60_check_openroadm_topo_ROADMA_SRG(self): - response = test_utils.get_ordm_topo_request("node/ROADM-A1-SRG1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( - res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) + response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") - liste_tp = res['node'][0]['ietf-network-topology:termination-point'] + liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'SRG1-PP1-TXRX': freq_map = base64.b64decode( @@ -970,14 +964,13 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(3) def test_61_check_openroadm_topo_ROADMA_DEG(self): - response = test_utils.get_ordm_topo_request("node/ROADM-A1-DEG2") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) freq_map = base64.b64decode( - res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) + response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") - liste_tp = res['node'][0]['ietf-network-topology:termination-point'] + liste_tp = response['node']['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'DEG2-CTP-TXRX': freq_map = base64.b64decode( @@ -1421,10 +1414,9 @@ class TransportPCEtesting(unittest.TestCase): def test_92_disconnect_xponders_from_roadm(self): url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/" - response = test_utils.get_ordm_topo_request("") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - links = res['network'][0]['ietf-network-topology:link'] + response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) + links = response['network'][0]['ietf-network-topology:link'] for link in links: if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'): link_name = link["link-id"] @@ -1432,11 +1424,11 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) def test_93_check_openroadm_topology(self): - response = test_utils.get_ordm_topo_request("") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - links = res['network'][0]['ietf-network-topology:link'] - self.assertEqual(18, len(links), 'Topology should contain 18 links') + response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config') + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(18, + len(response['network'][0]['ietf-network-topology:link']), + 'Topology should contain 18 links') def test_94_disconnect_spdrA(self): response = test_utils_rfc8040.unmount_device("SPDR-SA1") diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index 785f2a6d8..ec2758682 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -494,6 +494,24 @@ def del_oms_attr_request(link: str,): return response +def get_ietf_network_node_request(network: str, node: str, content: str): + url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}', + 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'} + if RESTCONF_VERSION == 'rfc8040': + format_args = ('{}', network, node, content) + elif content == 'config': + format_args = ('{}', content, network, node) + else: + format_args = ('{}', 'operational', network, node) + response = get_request(url[RESTCONF_VERSION].format(*format_args)) + res = response.json() + return_key = {'rfc8040': 'ietf-network:node', + 'draft-bierman02': 'node'} + node = res[return_key[RESTCONF_VERSION]][0] + return {'status_code': response.status_code, + 'node': node} + + def del_ietf_network_node_request(network: str, node: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'} -- 2.36.6