X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fhybrid%2Ftest02_B100G_end2end.py;h=9ebcffcd914ba868decb767773f90087978ba971;hb=6a0209f13ac0bb62053128e4a2281871caba91de;hp=f6e4e3a983775a3c7a2de00da20ac92edc0a6391;hpb=83f3af2d91fb44be21d2d6a17b3539784068211c;p=transportpce.git diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index f6e4e3a98..9ebcffcd9 100644 --- a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py +++ b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py @@ -9,16 +9,22 @@ # 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): @@ -145,7 +151,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)) @@ -277,7 +283,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) @@ -291,17 +297,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'}, + 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', @@ -323,15 +329,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, @@ -343,7 +351,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) @@ -357,17 +365,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'}, + 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', @@ -392,14 +400,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, @@ -429,8 +437,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) @@ -519,7 +527,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): @@ -546,7 +554,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): @@ -583,7 +591,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': @@ -632,16 +640,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', + 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 = {u'speed': 100000} + 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']), @@ -655,7 +663,7 @@ class TransportPCEtesting(unittest.TestCase): input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4-service-100GE', '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 = { @@ -669,7 +677,7 @@ 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): @@ -687,7 +695,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]) @@ -717,23 +725,23 @@ class TransportPCEtesting(unittest.TestCase): 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-service-100GE'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR2-CLIENT1-ODU4-service-100GE'}, + self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4-service-100GE'}, 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', 'supporting-port': 'C1' } - input_dict_2 = {u'speed': 100000} + 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']), @@ -747,7 +755,7 @@ class TransportPCEtesting(unittest.TestCase): input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4-service-100GE', '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 = { @@ -761,7 +769,7 @@ 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): @@ -815,9 +823,9 @@ class TransportPCEtesting(unittest.TestCase): 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-service-100GE'}, res['odu-connection'][0]['destination']) - self.assertDictEqual({u'src-if': u'XPDR2-CLIENT1-ODU4-service-100GE'}, + self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4-service-100GE'}, res['odu-connection'][0]['source']) def test_39_check_otn_topo_links(self): @@ -839,7 +847,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': @@ -913,7 +921,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': @@ -969,7 +977,7 @@ class TransportPCEtesting(unittest.TestCase): 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): @@ -991,7 +999,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': @@ -1027,7 +1035,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': @@ -1057,14 +1065,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): @@ -1080,7 +1088,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): @@ -1090,7 +1098,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) @@ -1099,7 +1107,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) @@ -1108,7 +1116,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) @@ -1117,7 +1125,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) @@ -1126,15 +1134,14 @@ 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-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" response = test_utils.service_create_request(self.cr_serv_sample_data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -1186,8 +1193,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)) @@ -1234,7 +1241,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) @@ -1245,7 +1252,7 @@ class TransportPCEtesting(unittest.TestCase): 'type': 'org-openroadm-interfaces:ethernetCsmacd', 'supporting-port': 'C1' } - input_dict_2 = {u'speed': 400000} + 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']), @@ -1279,10 +1286,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', @@ -1305,15 +1312,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]) @@ -1329,9 +1339,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, @@ -1349,6 +1361,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") @@ -1380,7 +1447,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): @@ -1400,8 +1467,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)