2 ##############################################################################
3 # Copyright (c) 2021 Orange, Inc. and others. All rights reserved.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 ##############################################################################
11 # pylint: disable=no-member
12 # pylint: disable=too-many-public-methods
18 # pylint: disable=wrong-import-order
20 sys.path.append('transportpce_tests/common/')
21 # pylint: disable=wrong-import-position
22 # pylint: disable=import-error
23 import test_utils # nopep8
25 # pylint: disable=too-few-public-methods
30 # pylint: disable=invalid-name
37 class TransportPCEFulltesting(unittest.TestCase):
39 cr_serv_input_data = {
42 "layer-protocol-name": "DSR",
43 "service-interface-point": {
44 "service-interface-point-uuid": "b1f4bd3b-7fa9-367b-a8ab-6e80293238df"
46 "administrative-state": "UNLOCKED",
47 "operational-state": "ENABLED",
48 "direction": "BIDIRECTIONAL",
50 "protection-role": "WORK",
51 "local-id": "XPDR-C1-XPDR1",
54 "value-name": "OpenROADM node id",
55 "value": "XPDR-C1-XPDR1"
60 "layer-protocol-name": "DSR",
61 "service-interface-point": {
62 "service-interface-point-uuid": "b5964ce9-274c-3f68-b4d1-83c0b61bc74e"
64 "administrative-state": "UNLOCKED",
65 "operational-state": "ENABLED",
66 "direction": "BIDIRECTIONAL",
68 "protection-role": "WORK",
69 "local-id": "XPDR-A1-XPDR1",
72 "value-name": "OpenROADM node id",
73 "value": "XPDR-A1-XPDR1"
78 "connectivity-constraint": {
79 "service-layer": "ETH",
80 "service-type": "POINT_TO_POINT_CONNECTIVITY",
81 "service-level": "Some service-level",
82 "requested-capacity": {
92 del_serv_input_data = {"service-id-or-name": "TBD"}
94 tapi_topo = {"topology-id-or-name": "TBD"}
97 "topology-id-or-name": "TBD",
98 "node-id-or-name": "TBD"
101 tapi_serv_details = {"service-id-or-name": "TBD"}
104 uuid_services = UuidServices()
105 WAITING = 25 # nominal value is 300
106 NODE_VERSION_221 = '2.2.1'
110 # pylint: disable=unsubscriptable-object
111 cls.init_failed = False
112 os.environ['JAVA_MIN_MEM'] = '1024M'
113 os.environ['JAVA_MAX_MEM'] = '4096M'
114 cls.processes = test_utils.start_tpce()
115 # TAPI feature is not installed by default in Karaf
116 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
117 print("installing tapi feature...")
118 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
119 if result.returncode != 0:
120 cls.init_failed = True
121 print("Restarting OpenDaylight...")
122 test_utils.shutdown_process(cls.processes[0])
123 cls.processes[0] = test_utils.start_karaf()
124 test_utils.process_list[0] = cls.processes[0]
125 # cause troubles after Chlorine bump
126 # cls.init_failed = not test_utils.wait_until_log_contains(
127 # test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
129 print("tapi installation feature failed...")
130 test_utils.shutdown_process(cls.processes[0])
132 cls.processes = test_utils.start_tpce()
133 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
134 ('roadma', cls.NODE_VERSION_221),
135 ('roadmc', cls.NODE_VERSION_221),
136 ('xpdrc', cls.NODE_VERSION_221)])
139 def tearDownClass(cls):
140 # pylint: disable=not-an-iterable
141 for process in cls.processes:
142 test_utils.shutdown_process(process)
143 print("all processes killed")
145 def setUp(self): # instruction executed before each test method
146 # pylint: disable=consider-using-f-string
147 print("execution of {}".format(self.id().split(".")[-1]))
150 def test_01_connect_xpdrA(self):
151 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
152 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
154 def test_02_connect_xpdrC(self):
155 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
156 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
158 def test_03_connect_rdmA(self):
159 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
160 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
162 def test_04_connect_rdmC(self):
163 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
164 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
166 def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
167 response = test_utils.transportpce_api_rpc_request(
168 'transportpce-networkutils', 'init-xpdr-rdm-links',
169 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
170 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
171 self.assertEqual(response['status_code'], requests.codes.ok)
172 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
175 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
176 response = test_utils.transportpce_api_rpc_request(
177 'transportpce-networkutils', 'init-rdm-xpdr-links',
178 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
179 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
180 self.assertEqual(response['status_code'], requests.codes.ok)
181 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
184 def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
185 response = test_utils.transportpce_api_rpc_request(
186 'transportpce-networkutils', 'init-xpdr-rdm-links',
187 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
188 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
189 self.assertEqual(response['status_code'], requests.codes.ok)
190 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
193 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
194 response = test_utils.transportpce_api_rpc_request(
195 'transportpce-networkutils', 'init-rdm-xpdr-links',
196 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
197 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
198 self.assertEqual(response['status_code'], requests.codes.ok)
199 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
202 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
203 # Config ROADMA-ROADMC oms-attributes
205 "auto-spanloss": "true",
206 "spanloss-base": 11.4,
207 "spanloss-current": 12,
208 "engineered-spanloss": 12.2,
209 "link-concatenation": [{
212 "SRLG-length": 100000,
214 response = test_utils.add_oms_attr_request(
215 "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
216 self.assertEqual(response.status_code, requests.codes.created)
218 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
219 # Config ROADMC-ROADMA oms-attributes
221 "auto-spanloss": "true",
222 "spanloss-base": 11.4,
223 "spanloss-current": 12,
224 "engineered-spanloss": 12.2,
225 "link-concatenation": [{
228 "SRLG-length": 100000,
230 response = test_utils.add_oms_attr_request(
231 "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
232 self.assertEqual(response.status_code, requests.codes.created)
234 # test service-create for Eth service from xpdr to xpdr
235 def test_11_create_connectivity_service_Ethernet(self):
236 response = test_utils.transportpce_api_rpc_request(
237 'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
238 time.sleep(self.WAITING)
239 self.uuid_services.eth = response['output']['service']['uuid']
240 # pylint: disable=consider-using-f-string
242 input_dict_1 = {'administrative-state': 'LOCKED',
243 'lifecycle-state': 'PLANNED',
244 'operational-state': 'DISABLED',
245 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
246 'service-layer': 'ETH',
247 'connectivity-direction': 'BIDIRECTIONAL'
249 input_dict_2 = {'value-name': 'OpenROADM node id',
250 'value': 'XPDR-C1-XPDR1'}
251 input_dict_3 = {'value-name': 'OpenROADM node id',
252 'value': 'XPDR-A1-XPDR1'}
254 self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
255 response['output']['service'])
256 self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
257 response['output']['service']['end-point'][0]['name'][0])
258 self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
259 response['output']['service']['end-point'][1]['name'][0])
260 # If the gate fails is because of the waiting time not being enough
261 # time.sleep(self.WAITING)
263 def test_12_get_service_Ethernet(self):
264 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
265 self.assertEqual(response['status_code'], requests.codes.ok)
266 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
267 self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
268 self.assertEqual(response['services'][0]['connection-type'], 'service')
269 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
272 def test_13_get_connectivity_service_Ethernet(self):
273 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
274 response = test_utils.transportpce_api_rpc_request(
275 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
276 self.assertEqual(response['status_code'], requests.codes.ok)
277 self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
278 self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
279 self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
280 self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
282 def test_14_change_status_line_port_xpdrc(self):
283 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
287 "administrative-state": "outOfService",
288 "port-qual": "xpdr-network"}]}
289 response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
290 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
291 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
292 timeout=test_utils.REQUEST_TIMEOUT)
293 self.assertEqual(response.status_code, requests.codes.ok)
296 def test_15_check_update_portmapping(self):
297 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
298 self.assertEqual(response['status_code'], requests.codes.ok)
299 mapping_list = response['nodes'][0]['mapping']
300 for mapping in mapping_list:
301 if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
302 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
303 "Operational State should be 'OutOfService'")
304 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
305 "Administrative State should be 'OutOfService'")
307 self.assertEqual(mapping['port-oper-state'], 'InService',
308 "Operational State should be 'InService'")
309 self.assertEqual(mapping['port-admin-state'], 'InService',
310 "Administrative State should be 'InService'")
313 def test_16_check_update_openroadm_topo(self):
314 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
315 self.assertEqual(response['status_code'], requests.codes.ok)
316 node_list = response['network'][0]['node']
318 for node in node_list:
319 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
320 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
321 tp_list = node['ietf-network-topology:termination-point']
323 if node['node-id'] == 'XPDR-C1-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
324 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
325 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
328 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
329 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
330 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
332 link_list = response['network'][0]['ietf-network-topology:link']
333 updated_links = ['XPDR-C1-XPDR1-XPDR1-NETWORK1toROADM-C1-SRG1-SRG1-PP1-TXRX',
334 'ROADM-C1-SRG1-SRG1-PP1-TXRXtoXPDR-C1-XPDR1-XPDR1-NETWORK1']
336 for link in link_list:
337 if link['link-id'] in updated_links:
338 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
339 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
342 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
343 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
344 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
347 def test_17_check_update_tapi_neps(self):
348 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
349 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi"
350 response = test_utils.transportpce_api_rpc_request(
351 'tapi-topology', 'get-node-details', self.node_details)
352 self.assertEqual(response['status_code'], requests.codes.ok)
353 nep_list = response['output']['node']['owned-node-edge-point']
356 if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
357 self.assertEqual(nep['operational-state'], 'DISABLED',
358 "Operational State should be 'DISABLED'")
359 self.assertEqual(nep['administrative-state'], 'LOCKED',
360 "Administrative State should be 'LOCKED'")
363 self.assertEqual(nep['operational-state'], 'ENABLED',
364 "Operational State should be 'ENABLED'")
365 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
366 "Administrative State should be 'UNLOCKED'")
367 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR"
368 response = test_utils.transportpce_api_rpc_request(
369 'tapi-topology', 'get-node-details', self.node_details)
370 self.assertEqual(response['status_code'], requests.codes.ok)
371 nep_list = response['output']['node']['owned-node-edge-point']
373 if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
374 self.assertEqual(nep['operational-state'], 'DISABLED',
375 "Operational State should be 'DISABLED'")
376 self.assertEqual(nep['administrative-state'], 'LOCKED',
377 "Administrative State should be 'LOCKED'")
380 self.assertEqual(nep['operational-state'], 'ENABLED',
381 "Operational State should be 'ENABLED'")
382 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
383 "Administrative State should be 'UNLOCKED'")
384 self.assertEqual(nb_updated_neps, 4, "Only two xponder neps should have been modified")
387 def test_18_check_update_tapi_links(self):
388 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
389 response = test_utils.transportpce_api_rpc_request(
390 'tapi-topology', 'get-topology-details', self.tapi_topo)
392 self.assertEqual(response['status_code'], requests.codes.ok)
393 link_list = response['output']['topology']['link']
395 for link in link_list:
396 if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']):
397 self.assertEqual(link['operational-state'], 'DISABLED')
398 self.assertEqual(link['administrative-state'], 'LOCKED')
401 self.assertEqual(link['operational-state'], 'ENABLED')
402 self.assertEqual(link['administrative-state'], 'UNLOCKED')
403 self.assertEqual(nb_updated_link, 2,
404 "Only two xponder-output/input & xponder-transi links should have been modified")
407 def test_19_check_update_service_Ethernet(self):
408 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
409 self.assertEqual(response['status_code'], requests.codes.ok)
410 self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
411 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
413 def test_20_check_update_connectivity_service_Ethernet(self):
414 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
415 response = test_utils.transportpce_api_rpc_request(
416 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
417 self.assertEqual(response['status_code'], requests.codes.ok)
418 self.assertEqual(response['output']['service']['operational-state'], 'DISABLED')
419 self.assertEqual(response['output']['service']['administrative-state'], 'LOCKED')
422 def test_21_restore_status_line_port_xpdrc(self):
423 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
427 "administrative-state": "inService",
428 "port-qual": "xpdr-network"}]}
429 response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
430 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
431 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
432 timeout=test_utils.REQUEST_TIMEOUT)
433 self.assertEqual(response.status_code, requests.codes.ok)
436 def test_22_check_update_portmapping_ok(self):
437 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
438 self.assertEqual(response['status_code'], requests.codes.ok)
439 mapping_list = response['nodes'][0]['mapping']
440 for mapping in mapping_list:
441 self.assertEqual(mapping['port-oper-state'], 'InService',
442 "Operational State should be 'InService'")
443 self.assertEqual(mapping['port-admin-state'], 'InService',
444 "Administrative State should be 'InService'")
447 def test_23_check_update_openroadm_topo_ok(self):
448 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
449 self.assertEqual(response['status_code'], requests.codes.ok)
450 node_list = response['network'][0]['node']
451 for node in node_list:
452 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
453 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
454 tp_list = node['ietf-network-topology:termination-point']
456 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
457 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
459 link_list = response['network'][0]['ietf-network-topology:link']
460 for link in link_list:
461 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
462 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
465 def test_24_check_update_tapi_neps_ok(self):
466 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
467 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi"
468 response = test_utils.transportpce_api_rpc_request(
469 'tapi-topology', 'get-node-details', self.node_details)
470 self.assertEqual(response['status_code'], requests.codes.ok)
471 nep_list = response['output']['node']['owned-node-edge-point']
473 self.assertEqual(nep['operational-state'], 'ENABLED',
474 "Operational State should be 'ENABLED'")
475 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
476 "Administrative State should be 'UNLOCKED'")
478 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR"
479 response = test_utils.transportpce_api_rpc_request(
480 'tapi-topology', 'get-node-details', self.node_details)
481 self.assertEqual(response['status_code'], requests.codes.ok)
482 nep_list = response['output']['node']['owned-node-edge-point']
484 self.assertEqual(nep['operational-state'], 'ENABLED',
485 "Operational State should be 'ENABLED'")
486 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
487 "Administrative State should be 'UNLOCKED'")
490 def test_25_check_update_tapi_links_ok(self):
491 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
492 response = test_utils.transportpce_api_rpc_request(
493 'tapi-topology', 'get-topology-details', self.tapi_topo)
495 link_list = response['output']['topology']['link']
496 for link in link_list:
497 self.assertEqual(link['operational-state'], 'ENABLED')
498 self.assertEqual(link['administrative-state'], 'UNLOCKED')
501 def test_26_check_update_service1_ok(self):
502 self.test_12_get_service_Ethernet()
504 def test_27_check_update_connectivity_service_Ethernet_ok(self):
505 self.test_13_get_connectivity_service_Ethernet()
507 def test_28_change_status_port_roadma_srg(self):
508 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
511 "logical-connection-point": "SRG1-PP1",
512 "port-type": "client",
513 "circuit-id": "SRG1",
514 "administrative-state": "outOfService",
515 "port-qual": "roadm-external"}]}
516 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
517 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
518 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
519 timeout=test_utils.REQUEST_TIMEOUT)
520 self.assertEqual(response.status_code, requests.codes.ok)
523 def test_29_check_update_portmapping(self):
524 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
525 self.assertEqual(response['status_code'], requests.codes.ok)
526 mapping_list = response['nodes'][0]['mapping']
527 for mapping in mapping_list:
528 if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
529 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
530 "Operational State should be 'OutOfService'")
531 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
532 "Administrative State should be 'OutOfService'")
534 self.assertEqual(mapping['port-oper-state'], 'InService',
535 "Operational State should be 'InService'")
536 self.assertEqual(mapping['port-admin-state'], 'InService',
537 "Administrative State should be 'InService'")
540 def test_30_check_update_openroadm_topo(self):
541 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
542 self.assertEqual(response['status_code'], requests.codes.ok)
543 node_list = response['network'][0]['node']
545 for node in node_list:
546 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
547 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
548 tp_list = node['ietf-network-topology:termination-point']
550 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
551 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
552 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
555 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
556 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
557 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
559 link_list = response['network'][0]['ietf-network-topology:link']
560 updated_links = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
561 'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
563 for link in link_list:
564 if link['link-id'] in updated_links:
565 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
566 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
569 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
570 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
571 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
574 def test_31_check_update_tapi_neps(self):
575 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
576 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
577 response = test_utils.transportpce_api_rpc_request(
578 'tapi-topology', 'get-node-details', self.node_details)
579 self.assertEqual(response['status_code'], requests.codes.ok)
580 nep_list = response['output']['node']['owned-node-edge-point']
583 if 'SRG1-PP1-TXRX' in nep['name'][0]['value']:
584 self.assertEqual(nep['operational-state'], 'DISABLED',
585 "Operational State should be 'DISABLED'")
586 self.assertEqual(nep['administrative-state'], 'LOCKED',
587 "Administrative State should be 'LOCKED'")
590 self.assertEqual(nep['operational-state'], 'ENABLED',
591 "Operational State should be 'ENABLED'")
592 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
593 "Administrative State should be 'UNLOCKED'")
594 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
597 def test_32_check_update_tapi_links(self):
598 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
599 response = test_utils.transportpce_api_rpc_request(
600 'tapi-topology', 'get-topology-details', self.tapi_topo)
602 link_list = response['output']['topology']['link']
604 for link in link_list:
605 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP1-TXRX']):
606 self.assertEqual(link['operational-state'], 'DISABLED')
607 self.assertEqual(link['administrative-state'], 'LOCKED')
610 self.assertEqual(link['operational-state'], 'ENABLED')
611 self.assertEqual(link['administrative-state'], 'UNLOCKED')
612 self.assertEqual(nb_updated_link, 1,
613 "Only one xponder-output/input link should have been modified")
616 def test_33_check_update_service_Ethernet(self):
617 self.test_19_check_update_service_Ethernet()
619 def test_34_check_update_connectivity_service_Ethernet(self):
620 self.test_20_check_update_connectivity_service_Ethernet()
622 def test_35_restore_status_port_roadma_srg(self):
623 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
626 "logical-connection-point": "SRG1-PP1",
627 "port-type": "client",
628 "circuit-id": "SRG1",
629 "administrative-state": "inService",
630 "port-qual": "roadm-external"}]}
631 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
632 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
633 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
634 timeout=test_utils.REQUEST_TIMEOUT)
635 self.assertEqual(response.status_code, requests.codes.ok)
638 def test_36_check_update_portmapping_ok(self):
639 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
640 self.assertEqual(response['status_code'], requests.codes.ok)
641 mapping_list = response['nodes'][0]['mapping']
642 for mapping in mapping_list:
643 self.assertEqual(mapping['port-oper-state'], 'InService',
644 "Operational State should be 'InService'")
645 self.assertEqual(mapping['port-admin-state'], 'InService',
646 "Administrative State should be 'InService'")
649 def test_37_check_update_openroadm_topo_ok(self):
650 self.test_23_check_update_openroadm_topo_ok()
652 def test_38_check_update_tapi_neps_ok(self):
653 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
654 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
655 response = test_utils.transportpce_api_rpc_request(
656 'tapi-topology', 'get-node-details', self.node_details)
657 self.assertEqual(response['status_code'], requests.codes.ok)
658 nep_list = response['output']['node']['owned-node-edge-point']
660 self.assertEqual(nep['operational-state'], 'ENABLED',
661 "Operational State should be 'ENABLED'")
662 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
663 "Administrative State should be 'UNLOCKED'")
667 def test_39_check_update_tapi_links_ok(self):
668 self.test_25_check_update_tapi_links_ok()
670 def test_40_check_update_service1_ok(self):
671 self.test_12_get_service_Ethernet()
673 def test_41_check_update_connectivity_service_Ethernet_ok(self):
674 self.test_13_get_connectivity_service_Ethernet()
676 def test_42_change_status_line_port_roadma_deg(self):
677 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
680 "logical-connection-point": "DEG2-TTP-TXRX",
683 "administrative-state": "outOfService",
684 "port-qual": "roadm-external"}]}
685 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
686 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
687 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
688 timeout=test_utils.REQUEST_TIMEOUT)
689 self.assertEqual(response.status_code, requests.codes.ok)
692 def test_43_check_update_portmapping(self):
693 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
694 self.assertEqual(response['status_code'], requests.codes.ok)
695 mapping_list = response['nodes'][0]['mapping']
696 for mapping in mapping_list:
697 if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
698 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
699 "Operational State should be 'OutOfService'")
700 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
701 "Administrative State should be 'OutOfService'")
703 self.assertEqual(mapping['port-oper-state'], 'InService',
704 "Operational State should be 'InService'")
705 self.assertEqual(mapping['port-admin-state'], 'InService',
706 "Administrative State should be 'InService'")
709 def test_44_check_update_openroadm_topo(self):
710 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
711 self.assertEqual(response['status_code'], requests.codes.ok)
712 node_list = response['network'][0]['node']
714 for node in node_list:
715 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
716 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
717 tp_list = node['ietf-network-topology:termination-point']
719 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
720 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
721 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
724 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
725 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
726 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
728 link_list = response['network'][0]['ietf-network-topology:link']
729 updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
730 'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
732 for link in link_list:
733 if link['link-id'] in updated_links:
734 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
735 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
738 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
739 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
740 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
743 def test_45_check_update_tapi_neps(self):
744 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
745 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
746 response = test_utils.transportpce_api_rpc_request(
747 'tapi-topology', 'get-node-details', self.node_details)
748 self.assertEqual(response['status_code'], requests.codes.ok)
749 nep_list = response['output']['node']['owned-node-edge-point']
752 if 'DEG2-TTP-TXRX' in nep['name'][0]['value']:
753 self.assertEqual(nep['operational-state'], 'DISABLED',
754 "Operational State should be 'DISABLED'")
755 self.assertEqual(nep['administrative-state'], 'LOCKED',
756 "Administrative State should be 'LOCKED'")
759 self.assertEqual(nep['operational-state'], 'ENABLED',
760 "Operational State should be 'ENABLED'")
761 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
762 "Administrative State should be 'UNLOCKED'")
763 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
766 def test_46_check_update_tapi_links(self):
767 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
768 response = test_utils.transportpce_api_rpc_request(
769 'tapi-topology', 'get-topology-details', self.tapi_topo)
771 self.assertEqual(response['status_code'], requests.codes.ok)
772 link_list = response['output']['topology']['link']
774 for link in link_list:
775 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'DEG2-TTP-TXRX']):
776 self.assertEqual(link['operational-state'], 'DISABLED')
777 self.assertEqual(link['administrative-state'], 'LOCKED')
780 self.assertEqual(link['operational-state'], 'ENABLED')
781 self.assertEqual(link['administrative-state'], 'UNLOCKED')
782 self.assertEqual(nb_updated_link, 1,
783 "Only one rdm-rdm link should have been modified")
786 def test_47_check_update_service_Ethernet(self):
787 self.test_19_check_update_service_Ethernet()
789 def test_48_check_update_connectivity_service_Ethernet(self):
790 self.test_20_check_update_connectivity_service_Ethernet()
792 def test_49_restore_status_line_port_roadma_deg(self):
793 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
796 "logical-connection-point": "DEG2-TTP-TXRX",
799 "administrative-state": "inService",
800 "port-qual": "roadm-external"}]}
801 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
802 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
803 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
804 timeout=test_utils.REQUEST_TIMEOUT)
805 self.assertEqual(response.status_code, requests.codes.ok)
808 def test_50_check_update_portmapping_ok(self):
809 self.test_36_check_update_portmapping_ok()
811 def test_51_check_update_openroadm_topo_ok(self):
812 self.test_23_check_update_openroadm_topo_ok()
814 def test_52_check_update_tapi_neps_ok(self):
815 self.test_38_check_update_tapi_neps_ok()
817 def test_53_check_update_tapi_links_ok(self):
818 self.test_25_check_update_tapi_links_ok()
820 def test_54_check_update_service1_ok(self):
821 self.test_12_get_service_Ethernet()
823 def test_55_check_update_connectivity_service_Ethernet_ok(self):
824 self.test_13_get_connectivity_service_Ethernet()
826 def test_56_change_status_port_roadma_srg(self):
827 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
830 "logical-connection-point": "SRG1-PP2",
831 "port-type": "client",
832 "circuit-id": "SRG1",
833 "administrative-state": "outOfService",
834 "port-qual": "roadm-external"}]}
835 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
836 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
837 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
838 timeout=test_utils.REQUEST_TIMEOUT)
839 self.assertEqual(response.status_code, requests.codes.ok)
842 def test_57_check_update_portmapping(self):
843 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
844 self.assertEqual(response['status_code'], requests.codes.ok)
845 mapping_list = response['nodes'][0]['mapping']
846 for mapping in mapping_list:
847 if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
848 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
849 "Operational State should be 'OutOfService'")
850 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
851 "Administrative State should be 'OutOfService'")
853 self.assertEqual(mapping['port-oper-state'], 'InService',
854 "Operational State should be 'InService'")
855 self.assertEqual(mapping['port-admin-state'], 'InService',
856 "Administrative State should be 'InService'")
859 def test_58_check_update_openroadm_topo(self):
860 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
861 self.assertEqual(response['status_code'], requests.codes.ok)
862 node_list = response['network'][0]['node']
864 for node in node_list:
865 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
866 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
867 tp_list = node['ietf-network-topology:termination-point']
869 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
870 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
871 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
874 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
875 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
876 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
878 link_list = response['network'][0]['ietf-network-topology:link']
880 for link in link_list:
881 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
882 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
883 self.assertEqual(nb_updated_link, 0, "No link should have been modified")
886 def test_59_check_update_tapi_neps(self):
887 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
888 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
889 response = test_utils.transportpce_api_rpc_request(
890 'tapi-topology', 'get-node-details', self.node_details)
891 self.assertEqual(response['status_code'], requests.codes.ok)
892 nep_list = response['output']['node']['owned-node-edge-point']
895 if 'SRG1-PP2-TXRX' in nep['name'][0]['value']:
896 self.assertEqual(nep['operational-state'], 'DISABLED',
897 "Operational State should be 'DISABLED'")
898 self.assertEqual(nep['administrative-state'], 'LOCKED',
899 "Administrative State should be 'LOCKED'")
902 self.assertEqual(nep['operational-state'], 'ENABLED',
903 "Operational State should be 'ENABLED'")
904 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
905 "Administrative State should be 'UNLOCKED'")
906 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
909 def test_60_check_update_tapi_links(self):
910 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
911 response = test_utils.transportpce_api_rpc_request(
912 'tapi-topology', 'get-topology-details', self.tapi_topo)
914 self.assertEqual(response['status_code'], requests.codes.ok)
915 link_list = response['output']['topology']['link']
917 for link in link_list:
918 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP2-TXRX']):
919 self.assertEqual(link['operational-state'], 'DISABLED')
920 self.assertEqual(link['administrative-state'], 'LOCKED')
923 self.assertEqual(link['operational-state'], 'ENABLED')
924 self.assertEqual(link['administrative-state'], 'UNLOCKED')
925 self.assertEqual(nb_updated_link, 0,
926 "No link should have been modified")
929 def test_61_check_update_service1_ok(self):
930 self.test_12_get_service_Ethernet()
932 def test_62_check_update_connectivity_service_Ethernet_ok(self):
933 self.test_13_get_connectivity_service_Ethernet()
935 def test_63_delete_connectivity_service_Ethernet(self):
936 self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.eth)
937 response = test_utils.transportpce_api_rpc_request(
938 'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
939 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
940 time.sleep(self.WAITING)
942 def test_64_disconnect_xponders_from_roadm(self):
943 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
944 self.assertEqual(response['status_code'], requests.codes.ok)
945 links = response['network'][0]['ietf-network-topology:link']
947 if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
948 response = test_utils.del_ietf_network_link_request(
949 'openroadm-topology', link['link-id'], 'config')
950 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
952 def test_65_disconnect_XPDRA(self):
953 response = test_utils.unmount_device("XPDR-A1")
954 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
956 def test_66_disconnect_XPDRC(self):
957 response = test_utils.unmount_device("XPDR-C1")
958 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
960 def test_67_disconnect_ROADMA(self):
961 response = test_utils.unmount_device("ROADM-A1")
962 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
964 def test_68_disconnect_ROADMC(self):
965 response = test_utils.unmount_device("ROADM-C1")
966 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
969 if __name__ == "__main__":
970 unittest.main(verbosity=2)