X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fhybrid%2Ftest02_B100G_end2end.py;h=aabe50b0ac9c62bd2ae31c99baa5370266be35ef;hb=refs%2Fchanges%2F96%2F99196%2F4;hp=856713da0dd0239bb77d6f5597152684c22aa4dd;hpb=de894bc83aa112264418e7eb6a2f2dd674451745;p=transportpce.git diff --git a/tests/transportpce_tests/hybrid/test02_B100G_end2end.py b/tests/transportpce_tests/hybrid/test02_B100G_end2end.py index 856713da0..aabe50b0a 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 @@ -146,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)) @@ -292,8 +297,8 @@ 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']) @@ -331,8 +336,10 @@ class TransportPCEtesting(unittest.TestCase): '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, @@ -358,8 +365,8 @@ 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']) @@ -397,10 +404,10 @@ class TransportPCEtesting(unittest.TestCase): '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 +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) @@ -520,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): @@ -547,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): @@ -584,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': @@ -642,7 +649,7 @@ class TransportPCEtesting(unittest.TestCase): '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']), @@ -670,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): @@ -688,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]) @@ -718,9 +725,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_35_check_interface_100GE_CLIENT_xpdrc2(self): @@ -734,7 +741,7 @@ class TransportPCEtesting(unittest.TestCase): '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']), @@ -762,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): @@ -816,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): @@ -840,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': @@ -914,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': @@ -992,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': @@ -1028,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': @@ -1081,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): @@ -1091,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) @@ -1100,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) @@ -1109,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) @@ -1118,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) @@ -1127,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() @@ -1187,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)) @@ -1235,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) @@ -1246,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']), @@ -1309,12 +1315,15 @@ class TransportPCEtesting(unittest.TestCase): ['supporting-interface-list'][0]: 'XPDR1-NETWORK1-OTSI-GROUP', '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 +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, @@ -1350,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") @@ -1401,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)