X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest08_otn_sh_renderer.py;h=74796810a2c2a816a15813d59196e88f58f7a54a;hb=8ec42577fc9bb12bed1a7689cd4a17487e3c1ff6;hp=cf358f2207c7913ee2b20cb842e99750c90e50c4;hpb=de894bc83aa112264418e7eb6a2f2dd674451745;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test08_otn_sh_renderer.py b/tests/transportpce_tests/2.2.1/test08_otn_sh_renderer.py index cf358f220..74796810a 100644 --- a/tests/transportpce_tests/2.2.1/test08_otn_sh_renderer.py +++ b/tests/transportpce_tests/2.2.1/test08_otn_sh_renderer.py @@ -15,9 +15,12 @@ import unittest import time import requests +# pylint: disable=wrong-import-order import sys sys.path.append('transportpce_tests/common/') -import test_utils # nopep8 +# pylint: disable=wrong-import-position +# pylint: disable=import-error +import test_utils_rfc8040 # nopep8 class TransportPCEtesting(unittest.TestCase): @@ -27,176 +30,164 @@ class TransportPCEtesting(unittest.TestCase): @classmethod def setUpClass(cls): - cls.processes = test_utils.start_tpce() - cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION), - ('spdrc', cls.NODE_VERSION)]) + cls.processes = test_utils_rfc8040.start_tpce() + cls.processes = test_utils_rfc8040.start_sims([('spdra', cls.NODE_VERSION), + ('spdrc', cls.NODE_VERSION)]) @classmethod def tearDownClass(cls): # pylint: disable=not-an-iterable for process in cls.processes: - test_utils.shutdown_process(process) + test_utils_rfc8040.shutdown_process(process) print("all processes killed") def setUp(self): time.sleep(5) def test_01_connect_SPDR_SA1(self): - response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) + response = test_utils_rfc8040.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, - test_utils.CODE_SHOULD_BE_201) + test_utils_rfc8040.CODE_SHOULD_BE_201) time.sleep(10) - response = test_utils.get_netconf_oper_request("SPDR-SA1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - self.assertEqual( - res['node'][0]['netconf-node-topology:connection-status'], - 'connected') + response = test_utils_rfc8040.check_device_connection("SPDR-SA1") + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(response['connection-status'], 'connected') def test_02_connect_SPDR_SC1(self): - response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) + response = test_utils_rfc8040.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, - test_utils.CODE_SHOULD_BE_201) + test_utils_rfc8040.CODE_SHOULD_BE_201) time.sleep(10) - response = test_utils.get_netconf_oper_request("SPDR-SC1") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() - self.assertEqual( - res['node'][0]['netconf-node-topology:connection-status'], - 'connected') + response = test_utils_rfc8040.check_device_connection("SPDR-SC1") + self.assertEqual(response['status_code'], requests.codes.ok) + self.assertEqual(response['connection-status'], 'connected') def test_03_service_create_OTU4(self): - url = "{}/operations/transportpce-renderer:service-implementation-request" - data = { - "transportpce-renderer:input": { - "transportpce-renderer:service-name": "SPDRA-SPDRC-OTU4-ODU4", - "transportpce-renderer:connection-type": "infrastructure", - "transportpce-renderer:service-handler-header": { - "transportpce-renderer:request-id": "abcd12-efgh34" + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-renderer', 'service-implementation-request', + { + 'service-name': 'SPDRA-SPDRC-OTU4-ODU4', + 'connection-type': 'infrastructure', + 'service-handler-header': { + 'request-id': 'abcd12-efgh34' }, - "transportpce-renderer:service-a-end": { - "transportpce-renderer:service-format": "OTU", - "transportpce-renderer:otu-service-rate": "org-openroadm-otn-common-types:OTU4", - "transportpce-renderer:clli": "nodeSA", - "transportpce-renderer:node-id": "SPDR-SA1" + 'service-a-end': { + 'service-format': 'OTU', + 'otu-service-rate': 'org-openroadm-otn-common-types:OTU4', + 'clli': 'nodeSA', + 'node-id': 'SPDR-SA1' }, - "transportpce-renderer:service-z-end": { - "transportpce-renderer:service-format": "OTU", - "transportpce-renderer:otu-service-rate": "org-openroadm-otn-common-types:OTU4", - "transportpce-renderer:clli": "nodeSC", - "transportpce-renderer:node-id": "SPDR-SC1" + 'service-z-end': { + 'service-format': 'OTU', + 'otu-service-rate': 'org-openroadm-otn-common-types:OTU4', + 'clli': 'nodeSC', + 'node-id': 'SPDR-SC1' }, - "transportpce-renderer:path-description": { - "aToZ-direction": { - "rate": 100, - "transportpce-renderer:modulation-format": "dp-qpsk", - "aToZ-wavelength-number": 1, - "aToZ": [ + 'path-description': { + 'aToZ-direction': { + 'rate': 100, + 'modulation-format': 'dp-qpsk', + 'aToZ-wavelength-number': 1, + 'aToZ': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': '' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': '' } } ], - "transportpce-renderer:aToZ-min-frequency": 196.075, - "transportpce-renderer:aToZ-max-frequency": 196.125 + 'aToZ-min-frequency': 196.075, + 'aToZ-max-frequency': 196.125 }, - "transportpce-renderer:zToA-direction": { - "transportpce-renderer:zToA-wavelength-number": "1", - "transportpce-renderer:rate": "100", - "transportpce-renderer:modulation-format": "dp-qpsk", - "zToA": [ + 'zToA-direction': { + 'zToA-wavelength-number': '1', + 'rate': '100', + 'modulation-format': 'dp-qpsk', + 'zToA': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': '' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': '' } } ], - "transportpce-renderer:zToA-min-frequency": 196.075, - "transportpce-renderer:zToA-max-frequency": 196.125 + 'zToA-min-frequency': 196.075, + 'zToA-max-frequency': 196.125 } } - } - } - response = test_utils.post_request(url, data) - time.sleep(3) - print(response.json()) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + }) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Operation Successful', - res["output"]["configuration-response-common"]["response-message"]) + response['output']['configuration-response-common']['response-message']) # Test OCH-OTU interfaces on SPDR-A1 def test_04_check_interface_och(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-761:768") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SA1", "interface", "XPDR1-NETWORK1-761:768") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertDictEqual(dict({'name': 'XPDR1-NETWORK1-761:768', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'type': 'org-openroadm-interfaces:opticalChannel', 'supporting-port': 'CP1-CFP0-P1' - }, **res['interface'][0]), - res['interface'][0]) - - self.assertDictEqual( - {u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G', - u'transmit-power': -5, u'modulation-format': 'dp-qpsk'}, - res['interface'][0]['org-openroadm-optical-channel-interfaces:och']) + }, **response['interface'][0]), + response['interface'][0]) + self.assertIn( + response['interface'][0]['org-openroadm-optical-channel-interfaces:och'], + [{'frequency': '196.1000', 'rate': 'org-openroadm-common-types:R100G', + 'transmit-power': '-5', 'modulation-format': 'dp-qpsk'}, + {'frequency': 196.1, 'rate': 'org-openroadm-common-types:R100G', + 'transmit-power': -5, 'modulation-format': 'dp-qpsk'}]) def test_05_check_interface_OTU(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SA1", "interface", "XPDR1-NETWORK1-OTU") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -204,42 +195,39 @@ class TransportPCEtesting(unittest.TestCase): 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'CP1-CFP0-P1' } - input_dict_2 = {'tx-dapi': 'fuYZwEO660g=', - 'expected-sapi': 'fuYZwEO660g=', - 'tx-sapi': 'Swfw02qXGyI=', - 'expected-dapi': 'Swfw02qXGyI=', + input_dict_2 = {'tx-dapi': 'AMf1n5hK6Xkk', + 'expected-sapi': 'AMf1n5hK6Xkk', + 'tx-sapi': 'H/OelLynehI=', + 'expected-dapi': 'H/OelLynehI=', 'rate': 'org-openroadm-otn-common-types:OTU4', 'fec': 'scfec' } - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) - - self.assertDictEqual(input_dict_2, - res['interface'][0] - ['org-openroadm-otn-otu-interfaces:otu']) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) + self.assertDictEqual(dict(input_dict_2, **response['interface'][0]['org-openroadm-otn-otu-interfaces:otu']), + response['interface'][0]['org-openroadm-otn-otu-interfaces:otu']) # Test OCH-OTU interfaces on SPDR-C1 def test_06_check_interface_och(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-NETWORK1-761:768") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SC1", "interface", "XPDR1-NETWORK1-761:768") + self.assertEqual(response['status_code'], requests.codes.ok) self.assertDictEqual(dict({'name': 'XPDR1-NETWORK1-1', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'type': 'org-openroadm-interfaces:opticalChannel', 'supporting-port': 'CP1-CFP0-P1' - }, **res['interface'][0]), - res['interface'][0]) - - self.assertDictEqual( - {u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G', - u'transmit-power': -5, u'modulation-format': 'dp-qpsk'}, - res['interface'][0]['org-openroadm-optical-channel-interfaces:och']) + }, **response['interface'][0]), + response['interface'][0]) + self.assertIn( + response['interface'][0]['org-openroadm-optical-channel-interfaces:och'], + [{'frequency': '196.1000', 'rate': 'org-openroadm-common-types:R100G', + 'transmit-power': '-5', 'modulation-format': 'dp-qpsk'}, + {'frequency': 196.1, 'rate': 'org-openroadm-common-types:R100G', + 'transmit-power': -5, 'modulation-format': 'dp-qpsk'}]) def test_07_check_interface_OTU(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-NETWORK1-OTU") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SC1", "interface", "XPDR1-NETWORK1-OTU") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -247,131 +235,121 @@ class TransportPCEtesting(unittest.TestCase): 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'CP1-CFP0-P1' } - input_dict_2 = {'tx-dapi': 'Swfw02qXGyI=', - 'expected-sapi': 'Swfw02qXGyI=', - 'tx-sapi': 'fuYZwEO660g=', - 'expected-dapi': 'fuYZwEO660g=', + input_dict_2 = {'tx-dapi': 'H/OelLynehI=', + 'expected-sapi': 'H/OelLynehI=', + 'tx-sapi': 'AMf1n5hK6Xkk', + 'expected-dapi': 'AMf1n5hK6Xkk', 'rate': 'org-openroadm-otn-common-types:OTU4', 'fec': 'scfec' } - - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) - - self.assertDictEqual(input_dict_2, - res['interface'][0] - ['org-openroadm-otn-otu-interfaces:otu']) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) + self.assertDictEqual(dict(input_dict_2, **response['interface'][0]['org-openroadm-otn-otu-interfaces:otu']), + response['interface'][0]['org-openroadm-otn-otu-interfaces:otu']) # Test creation of ODU4 service def test_08_service_create_ODU4(self): - url = "{}/operations/transportpce-renderer:service-implementation-request" - - data = { - "transportpce-renderer:input": { - "transportpce-renderer:service-name": - "SPDRA-SPDRC-OTU4-ODU4", - "transportpce-renderer:connection-type": "infrastructure", - "transportpce-renderer:service-handler-header": { - "transportpce-renderer:request-id": "abcd12-efgh34" + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-renderer', 'service-implementation-request', + { + 'service-name': + 'SPDRA-SPDRC-OTU4-ODU4', + 'connection-type': 'infrastructure', + 'service-handler-header': { + 'request-id': 'abcd12-efgh34' }, - "transportpce-renderer:service-a-end": { - "transportpce-renderer:service-format": "ODU", - "transportpce-renderer:odu-service-rate": - "org-openroadm-otn-common-types:ODU4", - "transportpce-renderer:clli": "nodeSA", - "transportpce-renderer:node-id": "SPDR-SA1" + 'service-a-end': { + 'service-format': 'ODU', + 'odu-service-rate': + 'org-openroadm-otn-common-types:ODU4', + 'clli': 'nodeSA', + 'node-id': 'SPDR-SA1' }, - "transportpce-renderer:service-z-end": { - "transportpce-renderer:service-format": "ODU", - "transportpce-renderer:odu-service-rate": - "org-openroadm-otn-common-types:ODU4", - "transportpce-renderer:clli": "nodeSC", - "transportpce-renderer:node-id": "SPDR-SC1" + 'service-z-end': { + 'service-format': 'ODU', + 'odu-service-rate': + 'org-openroadm-otn-common-types:ODU4', + 'clli': 'nodeSC', + 'node-id': 'SPDR-SC1' }, - "transportpce-renderer:path-description": { - "aToZ-direction": { - "rate": 100, - "aToZ": [ + 'path-description': { + 'aToZ-direction': { + 'rate': 100, + 'aToZ': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': '' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': '' } } ] }, - "transportpce-renderer:zToA-direction": { - "transportpce-renderer:rate": "100", - "zToA": [ + 'zToA-direction': { + 'rate': '100', + 'zToA': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': '' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': '' } } ] } } - } - } - response = test_utils.post_request(url, data) - time.sleep(3) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + }) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Operation Successful', - res["output"]["configuration-response-common"] - ["response-message"]) + response['output']['configuration-response-common']['response-message']) # Test ODU4 interfaces on SPDR-A1 and SPDR-C1 def test_09_check_interface_ODU4(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SA1", "interface", "XPDR1-NETWORK1-ODU4") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -381,25 +359,24 @@ class TransportPCEtesting(unittest.TestCase): # SAPI/DAPI are added in the Otu4 renderer input_dict_2 = {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP', 'rate': 'org-openroadm-otn-common-types:ODU4', - 'expected-dapi': 'Swfw02qXGyI=', - 'expected-sapi': 'fuYZwEO660g=', - 'tx-dapi': 'fuYZwEO660g=', - 'tx-sapi': 'Swfw02qXGyI='} - - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) - self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'], + 'expected-dapi': 'H/OelLynehI=', + 'expected-sapi': 'AMf1n5hK6Xkk', + 'tx-dapi': 'AMf1n5hK6Xkk', + 'tx-sapi': 'H/OelLynehI='} + + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) + self.assertDictEqual(dict(response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'], **input_dict_2), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'] + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'] ) self.assertDictEqual( - {u'payload-type': u'21', u'exp-payload-type': u'21'}, - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + {'payload-type': '21', 'exp-payload-type': '21'}, + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_10_check_interface_ODU4(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-NETWORK1-ODU4") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SC1", "interface", "XPDR1-NETWORK1-ODU4") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -409,150 +386,140 @@ class TransportPCEtesting(unittest.TestCase): # SAPI/DAPI are added in the Otu4 renderer input_dict_2 = {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP', 'rate': 'org-openroadm-otn-common-types:ODU4', - 'tx-sapi': 'fuYZwEO660g=', - 'tx-dapi': 'Swfw02qXGyI=', - 'expected-sapi': 'Swfw02qXGyI=', - 'expected-dapi': 'fuYZwEO660g=' + 'tx-sapi': 'AMf1n5hK6Xkk', + 'tx-dapi': 'H/OelLynehI=', + 'expected-sapi': 'H/OelLynehI=', + 'expected-dapi': 'AMf1n5hK6Xkk' } - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) - self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'], + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) + self.assertDictEqual(dict(response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'], **input_dict_2), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'] + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'] ) self.assertDictEqual( - {u'payload-type': u'21', u'exp-payload-type': u'21'}, - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + {'payload-type': '21', 'exp-payload-type': '21'}, + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) # Test creation of 10G service def test_11_service_create_10GE(self): - url = "{}/operations/transportpce-renderer:service-implementation-request" - - data = { - "transportpce-renderer:input": { - "transportpce-renderer:service-name": "SPDRA-SPDRC-10G", - "transportpce-renderer:connection-type": "service", - "transportpce-renderer:service-handler-header": { - "transportpce-renderer:request-id": "abcd12-efgh34" + response = test_utils_rfc8040.transportpce_api_rpc_request( + 'transportpce-renderer', 'service-implementation-request', + { + 'service-name': 'SPDRA-SPDRC-10G', + 'connection-type': 'service', + 'service-handler-header': { + 'request-id': 'abcd12-efgh34' }, - "transportpce-renderer:service-a-end": { - "transportpce-renderer:service-format": "Ethernet", - "transportpce-renderer:service-rate": "10", - "transportpce-renderer:clli": "nodeSA", - "transportpce-renderer:node-id": "SPDR-SA1" + 'service-a-end': { + 'service-format': 'Ethernet', + 'service-rate': '10', + 'clli': 'nodeSA', + 'node-id': 'SPDR-SA1' }, - "transportpce-renderer:service-z-end": { - "transportpce-renderer:service-format": "Ethernet", - "transportpce-renderer:service-rate": "10", - "transportpce-renderer:clli": "nodeSC", - "transportpce-renderer:node-id": "SPDR-SC1" + 'service-z-end': { + 'service-format': 'Ethernet', + 'service-rate': '10', + 'clli': 'nodeSC', + 'node-id': 'SPDR-SC1' }, - "transportpce-renderer:path-description": { - "aToZ-direction": { - "rate": 10, - "min-trib-slot": "1.1", - "max-trib-slot": "1.8", - "aToZ": [ + 'path-description': { + 'aToZ-direction': { + 'rate': 10, + 'min-trib-slot': '1.1', + 'max-trib-slot': '1.8', + 'aToZ': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-CLIENT1" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-CLIENT1' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-CLIENT1" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-CLIENT1' } } ] }, - "transportpce-renderer:zToA-direction": { - "rate": "10", - "min-trib-slot": "1.1", - "max-trib-slot": "1.8", - "zToA": [ + 'zToA-direction': { + 'rate': '10', + 'min-trib-slot': '1.1', + 'max-trib-slot': '1.8', + 'zToA': [ { - "id": "0", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-CLIENT1" + 'id': '0', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-CLIENT1' } }, { - "id": "1", - "resource": { - "tp-node-id": "SPDR-SC1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '1', + 'resource': { + 'tp-node-id': 'SPDR-SC1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "2", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-NETWORK1" + 'id': '2', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-NETWORK1' } }, { - "id": "3", - "resource": { - "tp-node-id": "SPDR-SA1-XPDR1", - "tp-id": "XPDR1-CLIENT1" + 'id': '3', + 'resource': { + 'tp-node-id': 'SPDR-SA1-XPDR1', + 'tp-id': 'XPDR1-CLIENT1' } } ] } } - } - } - - response = test_utils.post_request(url, data) - time.sleep(3) - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + }) + self.assertEqual(response['status_code'], requests.codes.ok) self.assertIn('Operation Successful', - res["output"]["configuration-response-common"] - ["response-message"]) + response['output']['configuration-response-common']['response-message']) # Test the interfaces on SPDR-A1 def test_12_check_interface_10GE_CLIENT(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SA1", "interface", "XPDR1-CLIENT1-ETHERNET10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', 'type': 'org-openroadm-interfaces:ethernetCsmacd', 'supporting-port': 'CP1-SFP4-P1' } - self.assertDictEqual(dict(input_dict, **res['interface'][0]), - res['interface'][0]) - self.assertDictEqual( - {u'speed': 10000}, - res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) + self.assertDictEqual(dict(input_dict, **response['interface'][0]), + response['interface'][0]) + self.assertEqual(response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']['speed'], 10000) def test_13_check_interface_ODU2E_CLIENT(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request( + "SPDR-SA1", "interface", "XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', @@ -564,38 +531,36 @@ class TransportPCEtesting(unittest.TestCase): 'rate': 'org-openroadm-otn-common-types:ODU2e', 'monitoring-mode': 'terminated'} - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) self.assertDictEqual(dict(input_dict_2, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'03', u'exp-payload-type': u'03'}, - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + {'payload-type': '03', 'exp-payload-type': '03'}, + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_14_check_ODU2E_connection(self): - response = test_utils.check_netconf_node_request( + response = test_utils_rfc8040.check_node_attribute_request( "SPDR-SA1", - "odu-connection/XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + "odu-connection", "XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { 'connection-name': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G', 'direction': 'bidirectional' } - - self.assertDictEqual(dict(input_dict_1, **res['odu-connection'][0]), - res['odu-connection'][0]) - self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G'}, - res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G'}, - res['odu-connection'][0]['source']) + self.assertDictEqual(dict(input_dict_1, **response['odu-connection'][0]), + response['odu-connection'][0]) + self.assertDictEqual({'dst-if': 'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G'}, + response['odu-connection'][0]['destination']) + self.assertDictEqual({'src-if': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G'}, + response['odu-connection'][0]['source']) def test_15_check_interface_ODU2E_NETWORK(self): - response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request( + "SPDR-SA1", "interface", "XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -606,29 +571,27 @@ class TransportPCEtesting(unittest.TestCase): 'odu-function': 'org-openroadm-otn-common-types:ODU-CTP', 'rate': 'org-openroadm-otn-common-types:ODU2e', 'monitoring-mode': 'monitored'} - input_dict_3 = {'trib-port-number': 1} - - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) self.assertDictEqual(dict(input_dict_2, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual(dict(input_dict_3, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']) self.assertIn(1, - res['interface'][0][ + response['interface'][0][ 'org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']['trib-slots']) # Test the interfaces on SPDR-C1 def test_16_check_interface_ODU2E_NETWORK(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request( + "SPDR-SC1", "interface", "XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -639,44 +602,39 @@ class TransportPCEtesting(unittest.TestCase): 'odu-function': 'org-openroadm-otn-common-types:ODU-CTP', 'rate': 'org-openroadm-otn-common-types:ODU2e', 'monitoring-mode': 'monitored'} - input_dict_3 = {'trib-port-number': 1} - - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) self.assertDictEqual(dict(input_dict_2, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual(dict(input_dict_3, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']) self.assertIn(1, - res['interface'][0][ + response['interface'][0][ 'org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']['trib-slots']) def test_17_check_interface_10GE_CLIENT(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-CLIENT1-ETHERNET10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request("SPDR-SC1", "interface", "XPDR1-CLIENT1-ETHERNET10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', 'type': 'org-openroadm-interfaces:ethernetCsmacd', 'supporting-port': 'CP1-SFP4-P1' } - self.assertDictEqual(dict(input_dict, **res['interface'][0]), - res['interface'][0]) - self.assertDictEqual( - {u'speed': 10000}, - res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) + self.assertDictEqual(dict(input_dict, **response['interface'][0]), + response['interface'][0]) + self.assertEqual(response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']['speed'], 10000) def test_18_check_interface_ODU2E_CLIENT(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request( + "SPDR-SC1", "interface", "XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', @@ -687,39 +645,36 @@ class TransportPCEtesting(unittest.TestCase): 'odu-function': 'org-openroadm-otn-common-types:ODU-TTP-CTP', 'rate': 'org-openroadm-otn-common-types:ODU2e', 'monitoring-mode': 'terminated'} - - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) self.assertDictEqual(dict(input_dict_2, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'03', u'exp-payload-type': u'03'}, - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + {'payload-type': '03', 'exp-payload-type': '03'}, + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_19_check_ODU2E_connection(self): - response = test_utils.check_netconf_node_request( + response = test_utils_rfc8040.check_node_attribute_request( "SPDR-SC1", - "odu-connection/XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + "odu-connection", "XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = { 'connection-name': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G-x-XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G', 'direction': 'bidirectional' } - - self.assertDictEqual(dict(input_dict_1, **res['odu-connection'][0]), - res['odu-connection'][0]) - self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G'}, - res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G'}, - res['odu-connection'][0]['source']) + self.assertDictEqual(dict(input_dict_1, **response['odu-connection'][0]), + response['odu-connection'][0]) + self.assertDictEqual({'dst-if': 'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G'}, + response['odu-connection'][0]['destination']) + self.assertDictEqual({'src-if': 'XPDR1-CLIENT1-ODU2e-SPDRA-SPDRC-10G'}, + response['odu-connection'][0]['source']) def test_20_check_interface_ODU2E_NETWORK(self): - response = test_utils.check_netconf_node_request("SPDR-SC1", "interface/XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") - self.assertEqual(response.status_code, requests.codes.ok) - res = response.json() + response = test_utils_rfc8040.check_node_attribute_request( + "SPDR-SC1", "interface", "XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G") + self.assertEqual(response['status_code'], requests.codes.ok) input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-SPDRA-SPDRC-10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', @@ -733,18 +688,18 @@ class TransportPCEtesting(unittest.TestCase): input_dict_3 = {'trib-port-number': 1} - self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), - res['interface'][0]) + self.assertDictEqual(dict(input_dict_1, **response['interface'][0]), + response['interface'][0]) self.assertDictEqual(dict(input_dict_2, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual(dict(input_dict_3, - **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']), - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']) self.assertIn(1, - res['interface'][0][ + response['interface'][0][ 'org-openroadm-otn-odu-interfaces:odu'][ 'parent-odu-allocation']['trib-slots']) @@ -752,14 +707,12 @@ class TransportPCEtesting(unittest.TestCase): # TODO: Delete interfaces (SPDR-A1, SPDR-C1) def test_21_disconnect_SPDR_SA1(self): - response = test_utils.unmount_device("SPDR-SA1") - self.assertEqual(response.status_code, requests.codes.ok, - test_utils.CODE_SHOULD_BE_200) + response = test_utils_rfc8040.unmount_device("SPDR-SA1") + self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) def test_22_disconnect_SPDR_SC1(self): - response = test_utils.unmount_device("SPDR-SC1") - self.assertEqual(response.status_code, requests.codes.ok, - test_utils.CODE_SHOULD_BE_200) + response = test_utils_rfc8040.unmount_device("SPDR-SC1") + self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content)) if __name__ == "__main__":