X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fhybrid%2Ftest02_B100G_end2end.py;h=51dc2e60ba2bc940f6d203f8c191461ee3afc16d;hb=c9bab520323eb66d759c3a7b2f90daa61d021f6b;hp=db0a67c1e61e2b768ea02005b679683860fcbe79;hpb=53b009d500687374e012405b1dd4b891e3ad30be;p=transportpce.git diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index db0a67c1e..51dc2e60b 100644 --- a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py +++ b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py @@ -9,15 +9,20 @@ # 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/') +# pylint: disable=wrong-import-position +# pylint: disable=import-error import test_utils # nopep8 @@ -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)) @@ -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) @@ -299,10 +296,10 @@ class TransportPCEtesting(unittest.TestCase): 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) @@ -365,10 +364,10 @@ class TransportPCEtesting(unittest.TestCase): 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, @@ -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,10 +632,10 @@ 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', + 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', @@ -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 = { @@ -675,10 +674,10 @@ class TransportPCEtesting(unittest.TestCase): 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', @@ -707,28 +706,28 @@ 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({'dst-if': 'XPDR2-NETWORK1-ODU4-service-100GE'}, + self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({'src-if': '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', + 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', @@ -742,13 +741,13 @@ class TransportPCEtesting(unittest.TestCase): 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 = { @@ -767,10 +766,10 @@ class TransportPCEtesting(unittest.TestCase): 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({'dst-if': 'XPDR2-NETWORK1-ODU4-service-100GE'}, + self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({'src-if': '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,7 +1027,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': @@ -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): @@ -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() @@ -1235,7 +1243,7 @@ class TransportPCEtesting(unittest.TestCase): 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) @@ -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)