X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fhybrid%2Ftest02_B100G_end2end.py;h=2bd319118ee204569bec52104d6bfc11a5c752f9;hb=56d1795b647693ec1d620d6aca2c4003e3b302ca;hp=b5e8c70eb2a2b6e03f01c8f4f6d59841ecc3677b;hpb=e48ffbcb7d5c3ecbfb438bd3a684c29f54d02bd9;p=transportpce.git diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index b5e8c70eb..2bd319118 100644 --- a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py +++ b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py @@ -9,16 +9,21 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +# pylint: disable=invalid-name # pylint: disable=no-member # pylint: disable=too-many-public-methods +# pylint: disable=too-many-lines import base64 import unittest import time import requests +# pylint: disable=wrong-import-order import sys sys.path.append('transportpce_tests/common/') -import test_utils +# pylint: disable=wrong-import-position +# pylint: disable=import-error +import test_utils # nopep8 class TransportPCEtesting(unittest.TestCase): @@ -43,13 +48,7 @@ class TransportPCEtesting(unittest.TestCase): "service-format": "OTU", "otu-service-rate": "org-openroadm-otn-common-types:OTUCn", "clli": "NodeSA", - "subrate-eth-sla": { - "subrate-eth-sla": { - "committed-info-rate": "100000", - "committed-burst-size": "64" - } - }, - "tx-direction": { + "tx-direction": [{ "port": { "port-device-name": "XPDR-A2-XPDR2", "port-type": "fixed", @@ -62,9 +61,10 @@ class TransportPCEtesting(unittest.TestCase): "lgx-port-name": "Some lgx-port-name", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" - } - }, - "rx-direction": { + }, + "index": 0 + }], + "rx-direction": [{ "port": { "port-device-name": "XPDR-A2-XPDR2", "port-type": "fixed", @@ -77,8 +77,9 @@ class TransportPCEtesting(unittest.TestCase): "lgx-port-name": "Some lgx-port-name", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" - } - }, + }, + "index": 0 + }], "optic-type": "gray" }, "service-z-end": { @@ -87,13 +88,7 @@ class TransportPCEtesting(unittest.TestCase): "service-format": "OTU", "otu-service-rate": "org-openroadm-otn-common-types:OTUCn", "clli": "NodeSC", - "subrate-eth-sla": { - "subrate-eth-sla": { - "committed-info-rate": "100000", - "committed-burst-size": "64" - } - }, - "tx-direction": { + "tx-direction": [{ "port": { "port-device-name": "XPDR-C2-XPDR2", "port-type": "fixed", @@ -106,9 +101,10 @@ class TransportPCEtesting(unittest.TestCase): "lgx-port-name": "Some lgx-port-name", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" - } - }, - "rx-direction": { + }, + "index": 0 + }], + "rx-direction": [{ "port": { "port-device-name": "XPDR-C2-XPDR2", "port-type": "fixed", @@ -121,8 +117,9 @@ class TransportPCEtesting(unittest.TestCase): "lgx-port-name": "Some lgx-port-name", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" - } - }, + }, + "index": 0 + }], "optic-type": "gray" }, "due-date": "2018-06-15T00:00:01Z", @@ -146,7 +143,7 @@ class TransportPCEtesting(unittest.TestCase): print("all processes killed") def setUp(self): - time.sleep(5) + time.sleep(2) def test_01_connect_xpdra2(self): response = test_utils.mount_device("XPDR-A2", ('xpdra2', self.NODE_VERSION_71)) @@ -242,7 +239,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.ok) res = response.json() nbNode = len(res['network'][0]['node']) - self.assertEqual(nbNode, 4, 'There should be 4 nodes') + self.assertEqual(nbNode, 6, 'There should be 6 nodes') self.assertNotIn('ietf-network-topology:link', res['network'][0]) def test_12_create_OTUC4_service(self): @@ -278,7 +275,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) # Check correct configuration of devices - def test_14_check_interface_och_xpdra2(self): + def test_14_check_interface_otsi_xpdra2(self): response = test_utils.check_netconf_node_request( "XPDR-A2", "interface/XPDR2-NETWORK1-755:768") self.assertEqual(response.status_code, requests.codes.ok) @@ -292,17 +289,17 @@ class TransportPCEtesting(unittest.TestCase): res['interface'][0]) self.assertDictEqual( - dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi', - u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, - **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), + dict({'frequency': 196.0812, 'otsi-rate': 'org-openroadm-common-optical-channel-types:R400G-otsi', + 'transmit-power': -5, 'modulation-format': 'dp-qam16'}, + **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']) def test_15_check_interface_OTSI_GROUP_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-NETWORK1-OTSI-GROUP") + "XPDR-A2", "interface/XPDR2-NETWORK1-OTSIGROUP-400G") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-NETWORK1-OTUC4', + input_dict_1 = {'name': 'XPDR2-NETWORK1-OTSIGROUP-400G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', 'supporting-interface-list': 'XPDR2-NETWORK1-755:768', @@ -324,15 +321,17 @@ class TransportPCEtesting(unittest.TestCase): "XPDR-A2", "interface/XPDR2-NETWORK1-OTUC4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-NETWORK1-OTSI-GROUP', + input_dict_1 = {'name': 'XPDR2-NETWORK1-OTUC4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', - 'supporting-interface-list': 'XPDR2-NETWORK1-OTSI-GROUP', + 'supporting-interface-list': 'XPDR2-NETWORK1-OTSIGROUP-400G', 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'L1' } - input_dict_2 = {'tx-sapi': 'LY9PxYJqUbw=', - 'expected-dapi': 'LY9PxYJqUbw=', + input_dict_2 = {'tx-sapi': 'G54UFNImtOE=', + 'expected-dapi': 'G54UFNImtOE=', + 'tx-dapi': 'J/FIUzQc+4M=', + 'expected-sapi': 'J/FIUzQc+4M=', 'rate': 'org-openroadm-otn-common-types:OTUCn', 'degthr-percentage': 100, 'degm-intervals': 2, @@ -344,7 +343,7 @@ class TransportPCEtesting(unittest.TestCase): res['interface'][0] ['org-openroadm-otn-otu-interfaces:otu']) - def test_17_check_interface_och_xpdrc2(self): + def test_17_check_interface_otsi_xpdrc2(self): response = test_utils.check_netconf_node_request( "XPDR-C2", "interface/XPDR2-NETWORK1-755:768") self.assertEqual(response.status_code, requests.codes.ok) @@ -358,17 +357,17 @@ class TransportPCEtesting(unittest.TestCase): res['interface'][0]) self.assertDictEqual( - dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi', - u'transmit-power': -5, u'modulation-format': 'dp-qam16'}, - **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), + dict({'frequency': 196.0812, 'otsi-rate': 'org-openroadm-common-optical-channel-types:R400G-otsi', + 'transmit-power': -5, 'modulation-format': 'dp-qam16'}, + **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']), res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']) def test_18_check_interface_OTSI_GROUP_xpdrc2(self): response = test_utils.check_netconf_node_request( - "XPDR-C2", "interface/XPDR2-NETWORK1-OTSI-GROUP") + "XPDR-C2", "interface/XPDR2-NETWORK1-OTSIGROUP-400G") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-NETWORK1-OTSI-GROUP', + input_dict_1 = {'name': 'XPDR2-NETWORK1-OTSIGROUP-400G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', 'supporting-interface-list': 'XPDR2-NETWORK1-755:768', @@ -393,14 +392,14 @@ class TransportPCEtesting(unittest.TestCase): input_dict_1 = {'name': 'XPDR1-NETWORK1-OTUC4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', - 'supporting-interface-list': 'XPDR2-NETWORK1-OTSI-GROUP', + 'supporting-interface-list': 'XPDR2-NETWORK1-OTSIGROUP-400G', 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'L1' } - input_dict_2 = {'tx-dapi': 'LY9PxYJqUbw=', - 'expected-sapi': 'LY9PxYJqUbw=', - 'tx-sapi': 'Nmbu2MNHvc4=', - 'expected-dapi': 'Nmbu2MNHvc4=', + input_dict_2 = {'tx-dapi': 'G54UFNImtOE=', + 'expected-sapi': 'G54UFNImtOE=', + 'tx-sapi': 'J/FIUzQc+4M=', + 'expected-dapi': 'J/FIUzQc+4M=', 'rate': 'org-openroadm-otn-common-types:OTUCn', 'degthr-percentage': 100, 'degm-intervals': 2, @@ -430,8 +429,8 @@ class TransportPCEtesting(unittest.TestCase): res = response.json() ele = res['node'][0]['ietf-network-topology:termination-point'][0] self.assertEqual('XPDR2-NETWORK1', ele['tp-id']) - self.assertEqual({u'frequency': 196.08125, - u'width': 75}, + self.assertEqual({'frequency': 196.08125, + 'width': 75}, ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) time.sleep(3) @@ -446,7 +445,7 @@ class TransportPCEtesting(unittest.TestCase): for link in res['network'][0]['ietf-network-topology:link']: self.assertIn(link['link-id'], listLinkId) self.assertEqual( - link['transportpce-topology:otn-link-type'], 'OTUC4') + link['transportpce-networkutils:otn-link-type'], 'OTUC4') self.assertEqual( link['org-openroadm-common-network:link-type'], 'OTN-LINK') self.assertEqual( @@ -520,7 +519,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'22', u'exp-payload-type': u'22'}, + {'payload-type': '22', 'exp-payload-type': '22'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_26_check_interface_ODUC4_xpdrc2(self): @@ -547,7 +546,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'22', u'exp-payload-type': u'22'}, + {'payload-type': '22', 'exp-payload-type': '22'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_27_check_otn_topo_links(self): @@ -571,7 +570,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual( link['org-openroadm-otn-network-topology:used-bandwidth'], 0) self.assertEqual( - link['transportpce-topology:otn-link-type'], 'ODUC4') + link['transportpce-networkutils:otn-link-type'], 'ODUC4') self.assertEqual( link['org-openroadm-common-network:link-type'], 'OTN-LINK') self.assertIn(link['org-openroadm-common-network:opposite-link'], @@ -584,7 +583,7 @@ class TransportPCEtesting(unittest.TestCase): response = test_utils.get_otn_topo_request() res = response.json() for node in res['network'][0]['node']: - if node['node-id'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2': + if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'): tpList = node['ietf-network-topology:termination-point'] for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': @@ -602,13 +601,13 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "100" self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "Ethernet" del self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] - self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"]["port"]["port-name"] = "XPDR2-CLIENT1" - self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"]["port"]["port-name"] = "XPDR2-CLIENT1" + self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT1" + self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT1" self.cr_serv_sample_data["input"]["service-z-end"]["service-rate"] = "100" self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "Ethernet" del self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] - self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"]["port"]["port-name"] = "XPDR2-CLIENT1" - self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"]["port"]["port-name"] = "XPDR2-CLIENT1" + self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT1" + self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT1" response = test_utils.service_create_request(self.cr_serv_sample_data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -633,16 +632,16 @@ class TransportPCEtesting(unittest.TestCase): def test_31_check_interface_100GE_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-CLIENT1-ETHERNET100G") + "XPDR-A2", "interface/XPDR2-CLIENT1-ETHERNET-100G") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } - input_dict_2 = {u'speed': 100000} + input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET-100G', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } + input_dict_2 = {'speed': 100000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']), @@ -650,13 +649,13 @@ class TransportPCEtesting(unittest.TestCase): def test_32_check_interface_ODU4_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-CLIENT1-ODU4-service-100GE") + "XPDR-A2", "interface/XPDR2-CLIENT1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4-service-100GE', + input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'supporting-interface-list': 'XPDR2-CLIENT1-ETHERNET100G', + 'supporting-interface-list': 'XPDR2-CLIENT1-ETHERNET-100G', 'type': 'org-openroadm-interfaces:otnOdu', 'supporting-port': 'C1'} input_dict_2 = { @@ -670,15 +669,15 @@ class TransportPCEtesting(unittest.TestCase): **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'07', u'exp-payload-type': u'07'}, + {'payload-type': '07', 'exp-payload-type': '07'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_33_check_interface_ODU4_NETWORK_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-NETWORK1-ODU4-service-100GE") + "XPDR-A2", "interface/XPDR2-NETWORK1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4-service-100GE', + input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', 'supporting-interface-list': 'XPDR2-NETWORK1-ODUC4', @@ -688,7 +687,7 @@ class TransportPCEtesting(unittest.TestCase): 'odu-function': 'org-openroadm-otn-common-types:ODU-CTP', 'rate': 'org-openroadm-otn-common-types:ODU4', 'monitoring-mode': 'not-terminated'} - input_dict_3 = {u'trib-port-number': 1} + input_dict_3 = {'trib-port-number': 1} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) @@ -707,48 +706,48 @@ class TransportPCEtesting(unittest.TestCase): def test_34_check_ODU4_connection_xpdra2(self): response = test_utils.check_netconf_node_request( "XPDR-A2", - "odu-connection/XPDR2-CLIENT1-ODU4-service-100GE-x-XPDR2-NETWORK1-ODU4-service-100GE") + "odu-connection/XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = { 'connection-name': - 'XPDR2-CLIENT1-ODU4-service-100GE-x-XPDR2-NETWORK1-ODU4-service-100GE', + 'XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4', 'direction': 'bidirectional' } self.assertDictEqual(dict(input_dict_1, **res['odu-connection'][0]), res['odu-connection'][0]) - self.assertDictEqual({u'dst-if': u'XPDR2-NETWORK1-ODU4-service-100GE'}, + self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR2-CLIENT1-ODU4-service-100GE'}, + self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4'}, res['odu-connection'][0]['source']) def test_35_check_interface_100GE_CLIENT_xpdrc2(self): response = test_utils.check_netconf_node_request( - "XPDR-C2", "interface/XPDR2-CLIENT1-ETHERNET100G") + "XPDR-C2", "interface/XPDR2-CLIENT1-ETHERNET-100G") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } - input_dict_2 = {u'speed': 100000} + input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET-100G', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } + input_dict_2 = {'speed': 100000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']), - res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) + res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) def test_36_check_interface_ODU4_CLIENT_xpdrc2(self): response = test_utils.check_netconf_node_request( - "XPDR-C2", "interface/XPDR2-CLIENT1-ODU4-service-100GE") + "XPDR-C2", "interface/XPDR2-CLIENT1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4-service-100GE', + input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT', - 'supporting-interface-list': 'XPDR2-CLIENT1-ETHERNET100G', + 'supporting-interface-list': 'XPDR2-CLIENT1-ETHERNET-100G', 'type': 'org-openroadm-interfaces:otnOdu', 'supporting-port': 'C1'} input_dict_2 = { @@ -762,15 +761,15 @@ class TransportPCEtesting(unittest.TestCase): **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']), res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) self.assertDictEqual( - {u'payload-type': u'07', u'exp-payload-type': u'07'}, + {'payload-type': '07', 'exp-payload-type': '07'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) def test_37_check_interface_ODU4_NETWORK_xpdrc2(self): response = test_utils.check_netconf_node_request( - "XPDR-C2", "interface/XPDR2-NETWORK1-ODU4-service-100GE") + "XPDR-C2", "interface/XPDR2-NETWORK1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4-service-100GE', + input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/2/2-PLUG-NET', 'supporting-interface-list': 'XPDR2-NETWORK1-ODUC4', @@ -805,20 +804,20 @@ class TransportPCEtesting(unittest.TestCase): def test_38_check_ODU4_connection_xpdrc2(self): response = test_utils.check_netconf_node_request( "XPDR-C2", - "odu-connection/XPDR2-CLIENT1-ODU4-service-100GE-x-XPDR2-NETWORK1-ODU4-service-100GE") + "odu-connection/XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = { 'connection-name': - 'XPDR2-CLIENT1-ODU4-service-100GE-x-XPDR2-NETWORK1-ODU4-service-100GE', + 'XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4', 'direction': 'bidirectional' } self.assertDictEqual(dict(input_dict_1, **res['odu-connection'][0]), res['odu-connection'][0]) - self.assertDictEqual({u'dst-if': u'XPDR2-NETWORK1-ODU4-service-100GE'}, + self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR2-CLIENT1-ODU4-service-100GE'}, + self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4'}, res['odu-connection'][0]['source']) def test_39_check_otn_topo_links(self): @@ -840,7 +839,7 @@ class TransportPCEtesting(unittest.TestCase): response = test_utils.get_otn_topo_request() res = response.json() for node in res['network'][0]['node']: - if node['node-id'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2': + if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'): tpList = node['ietf-network-topology:termination-point'] for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': @@ -860,12 +859,12 @@ class TransportPCEtesting(unittest.TestCase): self.cr_serv_sample_data["input"]["connection-type"] = "service" self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "100" self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "Ethernet" - self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"]["port"]["port-name"] = "XPDR2-CLIENT2" - self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"]["port"]["port-name"] = "XPDR2-CLIENT2" + self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT2" + self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT2" self.cr_serv_sample_data["input"]["service-z-end"]["service-rate"] = "100" self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "Ethernet" - self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"]["port"]["port-name"] = "XPDR2-CLIENT2" - self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"]["port"]["port-name"] = "XPDR2-CLIENT2" + self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT2" + self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR2-CLIENT2" response = test_utils.service_create_request(self.cr_serv_sample_data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -914,7 +913,7 @@ class TransportPCEtesting(unittest.TestCase): response = test_utils.get_otn_topo_request() res = response.json() for node in res['network'][0]['node']: - if node['node-id'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2': + if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'): tpList = node['ietf-network-topology:termination-point'] for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': @@ -960,17 +959,17 @@ class TransportPCEtesting(unittest.TestCase): def test_50_check_no_interface_ODU4_NETWORK_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-NETWORK1-ODU4-service-100GE") + "XPDR-A2", "interface/XPDR2-NETWORK1-ODU4") self.assertEqual(response.status_code, requests.codes.conflict) def test_51_check_no_interface_ODU4_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-CLIENT1-ODU4-service-100GE") + "XPDR-A2", "interface/XPDR2-CLIENT1-ODU4") self.assertEqual(response.status_code, requests.codes.conflict) def test_52_check_no_interface_100GE_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-CLIENT1-ETHERNET100G") + "XPDR-A2", "interface/XPDR2-CLIENT1-ETHERNET-100G") self.assertEqual(response.status_code, requests.codes.conflict) def test_53_check_otn_topo_links(self): @@ -992,7 +991,7 @@ class TransportPCEtesting(unittest.TestCase): response = test_utils.get_otn_topo_request() res = response.json() for node in res['network'][0]['node']: - if (node['node-id'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2'): + if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'): tpList = node['ietf-network-topology:termination-point'] for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': @@ -1028,12 +1027,12 @@ class TransportPCEtesting(unittest.TestCase): response = test_utils.get_otn_topo_request() res = response.json() for node in res['network'][0]['node']: - if node['node-id'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2': + if node['node-id'] in ('XPDR-A2-XPDR2', 'XPDR-C2-XPDR2'): tpList = node['ietf-network-topology:termination-point'] for tp in tpList: if tp['tp-id'] == 'XPDR2-NETWORK1': self.assertNotIn('org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes', - dict.keys(tp)) + dict.keys(tp)) def test_60_delete_OTUC4_service(self): response = test_utils.service_delete_request("service1-OTUC4") @@ -1058,14 +1057,14 @@ class TransportPCEtesting(unittest.TestCase): "XPDR-A2", "interface/XPDR2-NETWORK1-OTUC4") self.assertEqual(response.status_code, requests.codes.conflict) - def test_63_check_no_interface_OCH_xpdra2(self): + def test_63_check_no_interface_OTSI_xpdra2(self): response = test_utils.check_netconf_node_request( "XPDR-A2", "interface/XPDR2-NETWORK1-755:768") self.assertEqual(response.status_code, requests.codes.conflict) - def test_64_check_no_interface_OTSI_xpdra2(self): + def test_64_check_no_interface_OTSIG_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR2-NETWORK1-OTSI-GROUP") + "XPDR-A2", "interface/XPDR2-NETWORK1-OTSIGROUP-400G") self.assertEqual(response.status_code, requests.codes.conflict) def test_65_getLinks_OtnTopology(self): @@ -1081,7 +1080,7 @@ class TransportPCEtesting(unittest.TestCase): tp = res['node'][0]['ietf-network-topology:termination-point'][0] self.assertEqual('XPDR2-NETWORK1', tp['tp-id']) self.assertNotIn('wavelength', dict.keys( - tp[u'org-openroadm-network-topology:xpdr-network-attributes'])) + tp['org-openroadm-network-topology:xpdr-network-attributes'])) time.sleep(3) def test_67_check_openroadm_topology(self): @@ -1091,7 +1090,7 @@ class TransportPCEtesting(unittest.TestCase): links = res['network'][0]['ietf-network-topology:link'] self.assertEqual(22, len(links), 'Topology should contain 22 links') - def test_68_connect_xprda2_2_N1_to_roadma_PP2(self): + def test_68_connect_xprda2_1_N1_to_roadma_PP2(self): response = test_utils.connect_xpdr_to_rdm_request("XPDR-A2", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -1100,7 +1099,7 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_69_connect_roadma_PP2_to_xpdra2_2_N1(self): + def test_69_connect_roadma_PP2_to_xpdra2_1_N1(self): response = test_utils.connect_rdm_to_xpdr_request("XPDR-A2", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -1109,7 +1108,7 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_70_connect_xprdc2_2_N1_to_roadmc_PP2(self): + def test_70_connect_xprdc2_1_N1_to_roadmc_PP2(self): response = test_utils.connect_xpdr_to_rdm_request("XPDR-C2", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -1118,7 +1117,7 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_71_connect_roadmc_PP2_to_xpdrc2_2_N1(self): + def test_71_connect_roadmc_PP2_to_xpdrc2_1_N1(self): response = test_utils.connect_rdm_to_xpdr_request("XPDR-C2", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -1127,15 +1126,24 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) + # test service-create for 400GE service from xpdra2 to xpdrc2 + + def test_72_create_400GE_service(self): self.cr_serv_sample_data["input"]["service-name"] = "service-400GE" self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "400" - self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1" - self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1" + self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" + self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"][0]["port"]["port-name"] = "XPDR1-CLIENT1" self.cr_serv_sample_data["input"]["service-z-end"]["service-rate"] = "400" - self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1" - self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1" + del self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"][0]["port"]["port-name"] + del self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"][0]["port"]["port-device-name"] + del self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"][0]["port"]["port-name"] + del self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"][0]["port"]["port-device-name"] + del self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"][0]["port"]["port-name"] + del self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"][0]["port"]["port-device-name"] + del self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"][0]["port"]["port-name"] + del self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"][0]["port"]["port-device-name"] response = test_utils.service_create_request(self.cr_serv_sample_data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -1159,7 +1167,8 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) def test_74_check_xc1_roadma(self): - response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768") + response = test_utils.check_netconf_node_request( + "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 @@ -1186,8 +1195,8 @@ class TransportPCEtesting(unittest.TestCase): liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'XPDR1-NETWORK1': - self.assertEqual({u'frequency': 196.08125, - u'width': 75}, + self.assertEqual({'frequency': 196.08125, + 'width': 75}, ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) if ele['tp-id'] == 'XPDR1-CLIENT1': self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele)) @@ -1228,25 +1237,24 @@ class TransportPCEtesting(unittest.TestCase): freq_map_array = [int(x) for x in freq_map] self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available") if ele['tp-id'] == 'DEG2-TTP-TXRX': - freq_map = base64.b64decode( - 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, "Index 1 should not be available") + freq_map = base64.b64decode( + 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, "Index 1 should not be available") time.sleep(3) - - def test_78_check_interface_100GE_CLIENT_xpdra2(self): + def test_78_check_interface_400GE_CLIENT_xpdra2(self): response = test_utils.check_netconf_node_request( "XPDR-A2", "interface/XPDR1-CLIENT1-ETHERNET") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() input_dict_1 = {'name': 'XPDR1-CLIENT1-ETHERNET', - 'administrative-state': 'inService', - 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT', - 'type': 'org-openroadm-interfaces:ethernetCsmacd', - 'supporting-port': 'C1' - } - input_dict_2 = {u'speed': 400000} + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'C1' + } + input_dict_2 = {'speed': 400000} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']), @@ -1280,10 +1288,10 @@ class TransportPCEtesting(unittest.TestCase): def test_80_check_interface_OTSI_GROUP_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR1-NETWORK1-OTSI-GROUP") + "XPDR-A2", "interface/XPDR1-NETWORK1-OTSIGROUP-400G") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e-service1-10GE', + input_dict_1 = {'name': 'XPDR1-NETWORK1-OTSIGROUP-400G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/1/2-PLUG-NET', ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-755:768', @@ -1306,15 +1314,18 @@ class TransportPCEtesting(unittest.TestCase): input_dict_1 = {'name': 'XPDR1-NETWORK1-OTUC4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/1/2-PLUG-NET', - ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-OTSI-GROUP', + ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-OTSIGROUP-400G', 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'L1'} - input_dict_2 = {"tx-sapi": "AIGiVAQ4gDil", "rate": "org-openroadm-otn-common-types:OTUCn", + input_dict_2 = {"tx-sapi": "ANeUjNzWtDLV", + "expected-dapi": "ANeUjNzWtDLV", + 'tx-dapi': 'AKsqPmWceByv', + 'expected-sapi': 'AKsqPmWceByv', + "rate": "org-openroadm-otn-common-types:OTUCn", "degthr-percentage": 100, "tim-detect-mode": "Disabled", "otucn-n-rate": 4, - "degm-intervals": 2, - "expected-dapi": "AIGiVAQ4gDil"} + "degm-intervals": 2} self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), res['interface'][0]) @@ -1330,9 +1341,11 @@ class TransportPCEtesting(unittest.TestCase): input_dict_1 = {'name': 'XPDR1-NETWORK1-ODUC4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': '1/1/2-PLUG-NET', - ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-OTSI-GROUP', + ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-OTUC4', 'type': 'org-openroadm-interfaces:otnOdu', - 'supporting-port': 'L1'} + 'supporting-port': 'L1', + 'circuit-id': 'TBD', + 'description': 'TBD'} input_dict_2 = {"odu-function": "org-openroadm-otn-common-types:ODU-TTP", "tim-detect-mode": "Disabled", "degm-intervals": 2, @@ -1350,6 +1363,61 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual(dict(input_dict_3, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']), res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + self.assertEqual('XPDR1-NETWORK1-OTUC4', res['interface'][0]['supporting-interface-list'][0]) + + def test_82a_check_interface_ODUFLEX_xpdra2(self): + response = test_utils.check_netconf_node_request( + "XPDR-A2", "interface/XPDR1-NETWORK1-ODUFLEX") + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + input_dict_1 = {'name': 'XPDR1-NETWORK1-ODUFLEX', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': '1/1/2-PLUG-NET', + ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-ODUC4', + 'type': 'org-openroadm-interfaces:otnOdu', + 'supporting-port': 'L1', + 'circuit-id': 'TBD', + 'description': 'TBD'} + input_dict_2 = {"odu-function": "org-openroadm-otn-common-types:ODU-TTP-CTP", + "tim-detect-mode": "Disabled", + "degm-intervals": 2, + "degthr-percentage": 100, + "monitoring-mode": "terminated", + "rate": "org-openroadm-otn-common-types:ODUflex-cbr", + "oduflex-cbr-service": "org-openroadm-otn-common-types:ODUflex-cbr-400G" + } + input_dict_3 = {"exp-payload-type": "32", "payload-type": "32"} + input_dict_4 = {'trib-port-number': 1} + + self.assertDictEqual(dict(input_dict_1, **res['interface'][0]), + res['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']) + self.assertDictEqual(dict(input_dict_3, + **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']), + res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) + self.assertDictEqual(dict(input_dict_4, + **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][ + 'parent-odu-allocation']), + res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation']) + self.assertIn('1.1', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('1.20', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('2.1', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('2.20', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('3.1', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('3.20', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('4.1', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertIn('4.20', res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'] + ['opucn-trib-slots']) + self.assertEqual('XPDR1-NETWORK1-ODUC4', res['interface'][0]['supporting-interface-list'][0]) def test_83_delete_400GE_service(self): response = test_utils.service_delete_request("service-400GE") @@ -1381,7 +1449,7 @@ class TransportPCEtesting(unittest.TestCase): def test_87_check_no_interface_OTSI_GROUP_xpdra2(self): response = test_utils.check_netconf_node_request( - "XPDR-A2", "interface/XPDR1-NETWORK1-OTSI-GROUP") + "XPDR-A2", "interface/XPDR1-NETWORK1-OTSIGROUP-400G") self.assertEqual(response.status_code, requests.codes.conflict) def test_88_check_no_interface_OTSI_xpdra2(self): @@ -1401,8 +1469,7 @@ class TransportPCEtesting(unittest.TestCase): res = response.json() links = res['network'][0]['ietf-network-topology:link'] for link in links: - if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or - link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"): + if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'): link_name = link["link-id"] response = test_utils.delete_request(url+link_name) self.assertEqual(response.status_code, requests.codes.ok)