From c015bf8edf32eaf6fe78caa8bfe20cb5acb02b31 Mon Sep 17 00:00:00 2001 From: "guillaume.lambert" Date: Mon, 25 Jul 2022 22:38:32 +0200 Subject: [PATCH] Migrate OTN E2E functional tests to RFC8040 step 5 JIRA: TRNSPRTPCE-567 Signed-off-by: guillaume.lambert Change-Id: I425a1d19c772e9ef6f5829c3b55604feae9534a0 --- .../2.2.1/test11_otn_end2end.py | 129 +++++++----------- .../common/test_utils_rfc8040.py | 40 ++++++ 2 files changed, 93 insertions(+), 76 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 f30d14dee..3c1e47674 100644 --- a/tests/transportpce_tests/2.2.1/test11_otn_end2end.py +++ b/tests/transportpce_tests/2.2.1/test11_otn_end2end.py @@ -249,19 +249,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_13_get_OCH_OTU4_service1(self): - response = test_utils.get_service_list_request( - "services/service1-OCH-OTU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service1-OCH-OTU4") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service1-OCH-OTU4') + response['services'][0]['service-name'], 'service1-OCH-OTU4') self.assertEqual( - res['services'][0]['connection-type'], 'infrastructure') + response['services'][0]['connection-type'], 'infrastructure') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') # Check correct configuration of devices def test_14_check_interface_och_spdra(self): @@ -442,19 +440,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_24_get_ODU4_service1(self): - response = test_utils.get_service_list_request( - "services/service1-ODU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service1-ODU4") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service1-ODU4') + response['services'][0]['service-name'], 'service1-ODU4') self.assertEqual( - res['services'][0]['connection-type'], 'infrastructure') + response['services'][0]['connection-type'], 'infrastructure') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') def test_25_check_interface_ODU4_spdra(self): response = test_utils_rfc8040.check_node_attribute_request( @@ -584,19 +580,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_30_get_10GE_service1(self): - response = test_utils.get_service_list_request( - "services/service1-10GE") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service1-10GE") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service1-10GE') + response['services'][0]['service-name'], 'service1-10GE') self.assertEqual( - res['services'][0]['connection-type'], 'service') + response['services'][0]['connection-type'], 'service') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') def test_31_check_interface_10GE_CLIENT_spdra(self): response = test_utils_rfc8040.check_node_attribute_request( @@ -812,11 +806,9 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_42_check_service_list(self): - response = test_utils.get_service_list_request("") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - self.assertEqual(len(res['service-list']['services']), 2) - time.sleep(2) + response = test_utils_rfc8040.get_ordm_serv_list_request() + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(len(response['service-list']['services']), 2) def test_43_check_no_ODU2e_connection_spdra(self): response = test_utils.check_netconf_node_request("SPDR-SA1", "") @@ -875,11 +867,9 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_50_check_service_list(self): - response = test_utils.get_service_list_request("") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - self.assertEqual(len(res['service-list']['services']), 1) - time.sleep(2) + response = test_utils_rfc8040.get_ordm_serv_list_request() + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(len(response['service-list']['services']), 1) def test_51_check_no_interface_ODU4_spdra(self): response = test_utils_rfc8040.check_node_attribute_request( @@ -911,14 +901,8 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_55_get_no_service(self): - response = test_utils.get_service_list_request("") - self.assertEqual(response.status_code, requests.codes.conflict) - res = response.json() - self.assertIn( - {"error-type": "application", "error-tag": "data-missing", - "error-message": "Request could not be completed because the relevant data model content does not exist"}, - res['errors']['error']) - time.sleep(1) + response = test_utils_rfc8040.get_ordm_serv_list_request() + self.assertEqual(response['status_code'], requests.codes.conflict) def test_56_check_no_interface_OTU4_spdra(self): response = test_utils_rfc8040.check_node_attribute_request( @@ -1033,19 +1017,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_67_get_OCH_OTU4_service2(self): - response = test_utils.get_service_list_request( - "services/service2-OCH-OTU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service2-OCH-OTU4") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service2-OCH-OTU4') + response['services'][0]['service-name'], 'service2-OCH-OTU4') self.assertEqual( - res['services'][0]['connection-type'], 'infrastructure') + response['services'][0]['connection-type'], 'infrastructure') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') def test_68_create_ODU4_service_2(self): # pylint: disable=line-too-long @@ -1071,19 +1053,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_69_get_ODU4_service2(self): - response = test_utils.get_service_list_request( - "services/service2-ODU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service2-ODU4") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service2-ODU4') + response['services'][0]['service-name'], 'service2-ODU4') self.assertEqual( - res['services'][0]['connection-type'], 'infrastructure') + response['services'][0]['connection-type'], 'infrastructure') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') def test_70_create_1GE_service(self): # pylint: disable=line-too-long @@ -1110,18 +1090,17 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_71_get_1GE_service1(self): - response = test_utils.get_service_list_request("services/service1-1GE") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.get_ordm_serv_list_attr_request( + "services", "service1-1GE") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertEqual( - res['services'][0]['administrative-state'], 'inService') + response['services'][0]['administrative-state'], 'inService') self.assertEqual( - res['services'][0]['service-name'], 'service1-1GE') + response['services'][0]['service-name'], 'service1-1GE') self.assertEqual( - res['services'][0]['connection-type'], 'service') + response['services'][0]['connection-type'], 'service') self.assertEqual( - res['services'][0]['lifecycle-state'], 'planned') - time.sleep(2) + response['services'][0]['lifecycle-state'], 'planned') def test_72_check_interface_1GE_CLIENT_spdra(self): response = test_utils_rfc8040.check_node_attribute_request( @@ -1330,11 +1309,9 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(self.WAITING) def test_83_check_service_list(self): - response = test_utils.get_service_list_request("") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - self.assertEqual(len(res['service-list']['services']), 2) - time.sleep(2) + response = test_utils_rfc8040.get_ordm_serv_list_request() + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(len(response['service-list']['services']), 2) def test_84_check_no_ODU0_connection_spdra(self): response = test_utils.check_netconf_node_request("SPDR-SA1", "") diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index ec2758682..d53ddbf06 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -524,6 +524,46 @@ def del_ietf_network_node_request(network: str, node: str, content: str): response = delete_request(url[RESTCONF_VERSION].format(*format_args)) return response + +# +# Service list operations +# + + +def get_ordm_serv_list_request(): + url = {'rfc8040': '{}/data/org-openroadm-service:service-list?content=nonconfig', + 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/'} + response = get_request(url[RESTCONF_VERSION]) + res = response.json() + return_key = {'rfc8040': 'org-openroadm-service:service-list', + 'draft-bierman02': 'service-list'} + if return_key[RESTCONF_VERSION] in res.keys(): + response_attribute = res[return_key[RESTCONF_VERSION]] + else: + response_attribute = res['errors']['error'][0] + return {'status_code': response.status_code, + 'service-list': response_attribute} + + +def get_ordm_serv_list_attr_request(attribute: str, value: str): + url = {'rfc8040': '{}/data/org-openroadm-service:service-list/{}={}?content=nonconfig', + 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/{}/{}'} + if RESTCONF_VERSION == 'rfc8040': + format_args = ('{}', attribute, value) + else: + format_args = ('{}', attribute, value) + response = get_request(url[RESTCONF_VERSION].format(*format_args)) + res = response.json() + return_key = {'rfc8040': 'org-openroadm-service:' + attribute, + 'draft-bierman02': attribute} + if return_key[RESTCONF_VERSION] in res.keys(): + response_attribute = res[return_key[RESTCONF_VERSION]] + else: + response_attribute = res['errors']['error'][0] + return {'status_code': response.status_code, + attribute: response_attribute} + + # # TransportPCE internal API RPCs # -- 2.36.6