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 cls.init_failed = not test_utils.wait_until_log_contains(
126 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
128 print("tapi installation feature failed...")
129 test_utils.shutdown_process(cls.processes[0])
131 cls.processes = test_utils.start_tpce()
132 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
133 ('roadma', cls.NODE_VERSION_221),
134 ('roadmc', cls.NODE_VERSION_221),
135 ('xpdrc', cls.NODE_VERSION_221)])
138 def tearDownClass(cls):
139 # pylint: disable=not-an-iterable
140 for process in cls.processes:
141 test_utils.shutdown_process(process)
142 print("all processes killed")
144 def setUp(self): # instruction executed before each test method
145 # pylint: disable=consider-using-f-string
146 print("execution of {}".format(self.id().split(".")[-1]))
149 def test_01_connect_xpdrA(self):
150 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
151 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
153 def test_02_connect_xpdrC(self):
154 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
155 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
157 def test_03_connect_rdmA(self):
158 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
159 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
161 def test_04_connect_rdmC(self):
162 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
163 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
165 def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
166 response = test_utils.transportpce_api_rpc_request(
167 'transportpce-networkutils', 'init-xpdr-rdm-links',
168 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
169 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
170 self.assertEqual(response['status_code'], requests.codes.ok)
171 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
174 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
175 response = test_utils.transportpce_api_rpc_request(
176 'transportpce-networkutils', 'init-rdm-xpdr-links',
177 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
178 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
179 self.assertEqual(response['status_code'], requests.codes.ok)
180 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
183 def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
184 response = test_utils.transportpce_api_rpc_request(
185 'transportpce-networkutils', 'init-xpdr-rdm-links',
186 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
187 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
188 self.assertEqual(response['status_code'], requests.codes.ok)
189 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
192 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
193 response = test_utils.transportpce_api_rpc_request(
194 'transportpce-networkutils', 'init-rdm-xpdr-links',
195 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
196 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
197 self.assertEqual(response['status_code'], requests.codes.ok)
198 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
201 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
202 # Config ROADMA-ROADMC oms-attributes
204 "auto-spanloss": "true",
205 "spanloss-base": 11.4,
206 "spanloss-current": 12,
207 "engineered-spanloss": 12.2,
208 "link-concatenation": [{
211 "SRLG-length": 100000,
213 response = test_utils.add_oms_attr_request(
214 "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
215 self.assertEqual(response.status_code, requests.codes.created)
217 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
218 # Config ROADMC-ROADMA oms-attributes
220 "auto-spanloss": "true",
221 "spanloss-base": 11.4,
222 "spanloss-current": 12,
223 "engineered-spanloss": 12.2,
224 "link-concatenation": [{
227 "SRLG-length": 100000,
229 response = test_utils.add_oms_attr_request(
230 "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
231 self.assertEqual(response.status_code, requests.codes.created)
233 # test service-create for Eth service from xpdr to xpdr
234 def test_11_create_connectivity_service_Ethernet(self):
235 response = test_utils.transportpce_api_rpc_request(
236 'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
237 time.sleep(self.WAITING)
238 self.uuid_services.eth = response['output']['service']['uuid']
239 # pylint: disable=consider-using-f-string
241 input_dict_1 = {'administrative-state': 'LOCKED',
242 'lifecycle-state': 'PLANNED',
243 'operational-state': 'DISABLED',
244 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
245 'service-layer': 'ETH',
246 'connectivity-direction': 'BIDIRECTIONAL'
248 input_dict_2 = {'value-name': 'OpenROADM node id',
249 'value': 'XPDR-C1-XPDR1'}
250 input_dict_3 = {'value-name': 'OpenROADM node id',
251 'value': 'XPDR-A1-XPDR1'}
253 self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
254 response['output']['service'])
255 self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
256 response['output']['service']['end-point'][0]['name'][0])
257 self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
258 response['output']['service']['end-point'][1]['name'][0])
259 # If the gate fails is because of the waiting time not being enough
260 # time.sleep(self.WAITING)
262 def test_12_get_service_Ethernet(self):
263 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
264 self.assertEqual(response['status_code'], requests.codes.ok)
265 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
266 self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
267 self.assertEqual(response['services'][0]['connection-type'], 'service')
268 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
271 def test_13_get_connectivity_service_Ethernet(self):
272 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
273 response = test_utils.transportpce_api_rpc_request(
274 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
275 self.assertEqual(response['status_code'], requests.codes.ok)
276 self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
277 self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
278 self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
279 self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
281 def test_14_change_status_line_port_xpdrc(self):
282 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
286 "administrative-state": "outOfService",
287 "port-qual": "xpdr-network"}]}
288 response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
289 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
290 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
291 timeout=test_utils.REQUEST_TIMEOUT)
292 self.assertEqual(response.status_code, requests.codes.ok)
295 def test_15_check_update_portmapping(self):
296 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
297 self.assertEqual(response['status_code'], requests.codes.ok)
298 mapping_list = response['nodes'][0]['mapping']
299 for mapping in mapping_list:
300 if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
301 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
302 "Operational State should be 'OutOfService'")
303 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
304 "Administrative State should be 'OutOfService'")
306 self.assertEqual(mapping['port-oper-state'], 'InService',
307 "Operational State should be 'InService'")
308 self.assertEqual(mapping['port-admin-state'], 'InService',
309 "Administrative State should be 'InService'")
312 def test_16_check_update_openroadm_topo(self):
313 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
314 self.assertEqual(response['status_code'], requests.codes.ok)
315 node_list = response['network'][0]['node']
317 for node in node_list:
318 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
319 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
320 tp_list = node['ietf-network-topology:termination-point']
322 if node['node-id'] == 'XPDR-C1-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
323 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
324 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
327 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
328 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
329 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
331 link_list = response['network'][0]['ietf-network-topology:link']
332 updated_links = ['XPDR-C1-XPDR1-XPDR1-NETWORK1toROADM-C1-SRG1-SRG1-PP1-TXRX',
333 'ROADM-C1-SRG1-SRG1-PP1-TXRXtoXPDR-C1-XPDR1-XPDR1-NETWORK1']
335 for link in link_list:
336 if link['link-id'] in updated_links:
337 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
338 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
341 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
342 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
343 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
346 def test_17_check_update_tapi_neps(self):
347 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
348 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi"
349 response = test_utils.transportpce_api_rpc_request(
350 'tapi-topology', 'get-node-details', self.node_details)
351 self.assertEqual(response['status_code'], requests.codes.ok)
352 nep_list = response['output']['node']['owned-node-edge-point']
355 if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
356 self.assertEqual(nep['operational-state'], 'DISABLED',
357 "Operational State should be 'DISABLED'")
358 self.assertEqual(nep['administrative-state'], 'LOCKED',
359 "Administrative State should be 'LOCKED'")
362 self.assertEqual(nep['operational-state'], 'ENABLED',
363 "Operational State should be 'ENABLED'")
364 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
365 "Administrative State should be 'UNLOCKED'")
366 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR"
367 response = test_utils.transportpce_api_rpc_request(
368 'tapi-topology', 'get-node-details', self.node_details)
369 self.assertEqual(response['status_code'], requests.codes.ok)
370 nep_list = response['output']['node']['owned-node-edge-point']
372 if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
373 self.assertEqual(nep['operational-state'], 'DISABLED',
374 "Operational State should be 'DISABLED'")
375 self.assertEqual(nep['administrative-state'], 'LOCKED',
376 "Administrative State should be 'LOCKED'")
379 self.assertEqual(nep['operational-state'], 'ENABLED',
380 "Operational State should be 'ENABLED'")
381 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
382 "Administrative State should be 'UNLOCKED'")
383 self.assertEqual(nb_updated_neps, 4, "Only two xponder neps should have been modified")
386 def test_18_check_update_tapi_links(self):
387 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
388 response = test_utils.transportpce_api_rpc_request(
389 'tapi-topology', 'get-topology-details', self.tapi_topo)
391 self.assertEqual(response['status_code'], requests.codes.ok)
392 link_list = response['output']['topology']['link']
394 for link in link_list:
395 if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']):
396 self.assertEqual(link['operational-state'], 'DISABLED')
397 self.assertEqual(link['administrative-state'], 'LOCKED')
400 self.assertEqual(link['operational-state'], 'ENABLED')
401 self.assertEqual(link['administrative-state'], 'UNLOCKED')
402 self.assertEqual(nb_updated_link, 2,
403 "Only two xponder-output/input & xponder-transi links should have been modified")
406 def test_19_check_update_service_Ethernet(self):
407 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
408 self.assertEqual(response['status_code'], requests.codes.ok)
409 self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
410 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
412 def test_20_check_update_connectivity_service_Ethernet(self):
413 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
414 response = test_utils.transportpce_api_rpc_request(
415 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
416 self.assertEqual(response['status_code'], requests.codes.ok)
417 self.assertEqual(response['output']['service']['operational-state'], 'DISABLED')
418 self.assertEqual(response['output']['service']['administrative-state'], 'LOCKED')
421 def test_21_restore_status_line_port_xpdrc(self):
422 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
426 "administrative-state": "inService",
427 "port-qual": "xpdr-network"}]}
428 response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
429 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
430 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
431 timeout=test_utils.REQUEST_TIMEOUT)
432 self.assertEqual(response.status_code, requests.codes.ok)
435 def test_22_check_update_portmapping_ok(self):
436 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
437 self.assertEqual(response['status_code'], requests.codes.ok)
438 mapping_list = response['nodes'][0]['mapping']
439 for mapping in mapping_list:
440 self.assertEqual(mapping['port-oper-state'], 'InService',
441 "Operational State should be 'InService'")
442 self.assertEqual(mapping['port-admin-state'], 'InService',
443 "Administrative State should be 'InService'")
446 def test_23_check_update_openroadm_topo_ok(self):
447 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
448 self.assertEqual(response['status_code'], requests.codes.ok)
449 node_list = response['network'][0]['node']
450 for node in node_list:
451 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
452 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
453 tp_list = node['ietf-network-topology:termination-point']
455 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
456 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
458 link_list = response['network'][0]['ietf-network-topology:link']
459 for link in link_list:
460 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
461 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
464 def test_24_check_update_tapi_neps_ok(self):
465 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
466 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+OTSi"
467 response = test_utils.transportpce_api_rpc_request(
468 'tapi-topology', 'get-node-details', self.node_details)
469 self.assertEqual(response['status_code'], requests.codes.ok)
470 nep_list = response['output']['node']['owned-node-edge-point']
472 self.assertEqual(nep['operational-state'], 'ENABLED',
473 "Operational State should be 'ENABLED'")
474 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
475 "Administrative State should be 'UNLOCKED'")
477 self.node_details["node-id-or-name"] = "XPDR-C1-XPDR1+DSR"
478 response = test_utils.transportpce_api_rpc_request(
479 'tapi-topology', 'get-node-details', self.node_details)
480 self.assertEqual(response['status_code'], requests.codes.ok)
481 nep_list = response['output']['node']['owned-node-edge-point']
483 self.assertEqual(nep['operational-state'], 'ENABLED',
484 "Operational State should be 'ENABLED'")
485 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
486 "Administrative State should be 'UNLOCKED'")
489 def test_25_check_update_tapi_links_ok(self):
490 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
491 response = test_utils.transportpce_api_rpc_request(
492 'tapi-topology', 'get-topology-details', self.tapi_topo)
494 link_list = response['output']['topology']['link']
495 for link in link_list:
496 self.assertEqual(link['operational-state'], 'ENABLED')
497 self.assertEqual(link['administrative-state'], 'UNLOCKED')
500 def test_26_check_update_service1_ok(self):
501 self.test_12_get_service_Ethernet()
503 def test_27_check_update_connectivity_service_Ethernet_ok(self):
504 self.test_13_get_connectivity_service_Ethernet()
506 def test_28_change_status_port_roadma_srg(self):
507 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
510 "logical-connection-point": "SRG1-PP1",
511 "port-type": "client",
512 "circuit-id": "SRG1",
513 "administrative-state": "outOfService",
514 "port-qual": "roadm-external"}]}
515 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
516 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
517 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
518 timeout=test_utils.REQUEST_TIMEOUT)
519 self.assertEqual(response.status_code, requests.codes.ok)
522 def test_29_check_update_portmapping(self):
523 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
524 self.assertEqual(response['status_code'], requests.codes.ok)
525 mapping_list = response['nodes'][0]['mapping']
526 for mapping in mapping_list:
527 if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
528 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
529 "Operational State should be 'OutOfService'")
530 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
531 "Administrative State should be 'OutOfService'")
533 self.assertEqual(mapping['port-oper-state'], 'InService',
534 "Operational State should be 'InService'")
535 self.assertEqual(mapping['port-admin-state'], 'InService',
536 "Administrative State should be 'InService'")
539 def test_30_check_update_openroadm_topo(self):
540 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
541 self.assertEqual(response['status_code'], requests.codes.ok)
542 node_list = response['network'][0]['node']
544 for node in node_list:
545 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
546 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
547 tp_list = node['ietf-network-topology:termination-point']
549 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
550 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
551 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
554 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
555 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
556 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
558 link_list = response['network'][0]['ietf-network-topology:link']
559 updated_links = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
560 'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
562 for link in link_list:
563 if link['link-id'] in updated_links:
564 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
565 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
568 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
569 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
570 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
573 def test_31_check_update_tapi_neps(self):
574 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
575 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
576 response = test_utils.transportpce_api_rpc_request(
577 'tapi-topology', 'get-node-details', self.node_details)
578 self.assertEqual(response['status_code'], requests.codes.ok)
579 nep_list = response['output']['node']['owned-node-edge-point']
582 if 'SRG1-PP1-TXRX' in nep['name'][0]['value']:
583 self.assertEqual(nep['operational-state'], 'DISABLED',
584 "Operational State should be 'DISABLED'")
585 self.assertEqual(nep['administrative-state'], 'LOCKED',
586 "Administrative State should be 'LOCKED'")
589 self.assertEqual(nep['operational-state'], 'ENABLED',
590 "Operational State should be 'ENABLED'")
591 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
592 "Administrative State should be 'UNLOCKED'")
593 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
596 def test_32_check_update_tapi_links(self):
597 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
598 response = test_utils.transportpce_api_rpc_request(
599 'tapi-topology', 'get-topology-details', self.tapi_topo)
601 link_list = response['output']['topology']['link']
603 for link in link_list:
604 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP1-TXRX']):
605 self.assertEqual(link['operational-state'], 'DISABLED')
606 self.assertEqual(link['administrative-state'], 'LOCKED')
609 self.assertEqual(link['operational-state'], 'ENABLED')
610 self.assertEqual(link['administrative-state'], 'UNLOCKED')
611 self.assertEqual(nb_updated_link, 1,
612 "Only one xponder-output/input link should have been modified")
615 def test_33_check_update_service_Ethernet(self):
616 self.test_19_check_update_service_Ethernet()
618 def test_34_check_update_connectivity_service_Ethernet(self):
619 self.test_20_check_update_connectivity_service_Ethernet()
621 def test_35_restore_status_port_roadma_srg(self):
622 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
625 "logical-connection-point": "SRG1-PP1",
626 "port-type": "client",
627 "circuit-id": "SRG1",
628 "administrative-state": "inService",
629 "port-qual": "roadm-external"}]}
630 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
631 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
632 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
633 timeout=test_utils.REQUEST_TIMEOUT)
634 self.assertEqual(response.status_code, requests.codes.ok)
637 def test_36_check_update_portmapping_ok(self):
638 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
639 self.assertEqual(response['status_code'], requests.codes.ok)
640 mapping_list = response['nodes'][0]['mapping']
641 for mapping in mapping_list:
642 self.assertEqual(mapping['port-oper-state'], 'InService',
643 "Operational State should be 'InService'")
644 self.assertEqual(mapping['port-admin-state'], 'InService',
645 "Administrative State should be 'InService'")
648 def test_37_check_update_openroadm_topo_ok(self):
649 self.test_23_check_update_openroadm_topo_ok()
651 def test_38_check_update_tapi_neps_ok(self):
652 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
653 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
654 response = test_utils.transportpce_api_rpc_request(
655 'tapi-topology', 'get-node-details', self.node_details)
656 self.assertEqual(response['status_code'], requests.codes.ok)
657 nep_list = response['output']['node']['owned-node-edge-point']
659 self.assertEqual(nep['operational-state'], 'ENABLED',
660 "Operational State should be 'ENABLED'")
661 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
662 "Administrative State should be 'UNLOCKED'")
666 def test_39_check_update_tapi_links_ok(self):
667 self.test_25_check_update_tapi_links_ok()
669 def test_40_check_update_service1_ok(self):
670 self.test_12_get_service_Ethernet()
672 def test_41_check_update_connectivity_service_Ethernet_ok(self):
673 self.test_13_get_connectivity_service_Ethernet()
675 def test_42_change_status_line_port_roadma_deg(self):
676 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
679 "logical-connection-point": "DEG2-TTP-TXRX",
682 "administrative-state": "outOfService",
683 "port-qual": "roadm-external"}]}
684 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
685 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
686 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
687 timeout=test_utils.REQUEST_TIMEOUT)
688 self.assertEqual(response.status_code, requests.codes.ok)
691 def test_43_check_update_portmapping(self):
692 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
693 self.assertEqual(response['status_code'], requests.codes.ok)
694 mapping_list = response['nodes'][0]['mapping']
695 for mapping in mapping_list:
696 if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
697 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
698 "Operational State should be 'OutOfService'")
699 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
700 "Administrative State should be 'OutOfService'")
702 self.assertEqual(mapping['port-oper-state'], 'InService',
703 "Operational State should be 'InService'")
704 self.assertEqual(mapping['port-admin-state'], 'InService',
705 "Administrative State should be 'InService'")
708 def test_44_check_update_openroadm_topo(self):
709 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
710 self.assertEqual(response['status_code'], requests.codes.ok)
711 node_list = response['network'][0]['node']
713 for node in node_list:
714 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
715 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
716 tp_list = node['ietf-network-topology:termination-point']
718 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
719 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
720 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
723 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
724 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
725 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
727 link_list = response['network'][0]['ietf-network-topology:link']
728 updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
729 'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
731 for link in link_list:
732 if link['link-id'] in updated_links:
733 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
734 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
737 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
738 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
739 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
742 def test_45_check_update_tapi_neps(self):
743 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
744 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
745 response = test_utils.transportpce_api_rpc_request(
746 'tapi-topology', 'get-node-details', self.node_details)
747 self.assertEqual(response['status_code'], requests.codes.ok)
748 nep_list = response['output']['node']['owned-node-edge-point']
751 if 'DEG2-TTP-TXRX' in nep['name'][0]['value']:
752 self.assertEqual(nep['operational-state'], 'DISABLED',
753 "Operational State should be 'DISABLED'")
754 self.assertEqual(nep['administrative-state'], 'LOCKED',
755 "Administrative State should be 'LOCKED'")
758 self.assertEqual(nep['operational-state'], 'ENABLED',
759 "Operational State should be 'ENABLED'")
760 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
761 "Administrative State should be 'UNLOCKED'")
762 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
765 def test_46_check_update_tapi_links(self):
766 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
767 response = test_utils.transportpce_api_rpc_request(
768 'tapi-topology', 'get-topology-details', self.tapi_topo)
770 self.assertEqual(response['status_code'], requests.codes.ok)
771 link_list = response['output']['topology']['link']
773 for link in link_list:
774 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'DEG2-TTP-TXRX']):
775 self.assertEqual(link['operational-state'], 'DISABLED')
776 self.assertEqual(link['administrative-state'], 'LOCKED')
779 self.assertEqual(link['operational-state'], 'ENABLED')
780 self.assertEqual(link['administrative-state'], 'UNLOCKED')
781 self.assertEqual(nb_updated_link, 1,
782 "Only one rdm-rdm link should have been modified")
785 def test_47_check_update_service_Ethernet(self):
786 self.test_19_check_update_service_Ethernet()
788 def test_48_check_update_connectivity_service_Ethernet(self):
789 self.test_20_check_update_connectivity_service_Ethernet()
791 def test_49_restore_status_line_port_roadma_deg(self):
792 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
795 "logical-connection-point": "DEG2-TTP-TXRX",
798 "administrative-state": "inService",
799 "port-qual": "roadm-external"}]}
800 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
801 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
802 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
803 timeout=test_utils.REQUEST_TIMEOUT)
804 self.assertEqual(response.status_code, requests.codes.ok)
807 def test_50_check_update_portmapping_ok(self):
808 self.test_36_check_update_portmapping_ok()
810 def test_51_check_update_openroadm_topo_ok(self):
811 self.test_23_check_update_openroadm_topo_ok()
813 def test_52_check_update_tapi_neps_ok(self):
814 self.test_38_check_update_tapi_neps_ok()
816 def test_53_check_update_tapi_links_ok(self):
817 self.test_25_check_update_tapi_links_ok()
819 def test_54_check_update_service1_ok(self):
820 self.test_12_get_service_Ethernet()
822 def test_55_check_update_connectivity_service_Ethernet_ok(self):
823 self.test_13_get_connectivity_service_Ethernet()
825 def test_56_change_status_port_roadma_srg(self):
826 url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
829 "logical-connection-point": "SRG1-PP2",
830 "port-type": "client",
831 "circuit-id": "SRG1",
832 "administrative-state": "outOfService",
833 "port-qual": "roadm-external"}]}
834 response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
835 data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
836 auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
837 timeout=test_utils.REQUEST_TIMEOUT)
838 self.assertEqual(response.status_code, requests.codes.ok)
841 def test_57_check_update_portmapping(self):
842 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
843 self.assertEqual(response['status_code'], requests.codes.ok)
844 mapping_list = response['nodes'][0]['mapping']
845 for mapping in mapping_list:
846 if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
847 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
848 "Operational State should be 'OutOfService'")
849 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
850 "Administrative State should be 'OutOfService'")
852 self.assertEqual(mapping['port-oper-state'], 'InService',
853 "Operational State should be 'InService'")
854 self.assertEqual(mapping['port-admin-state'], 'InService',
855 "Administrative State should be 'InService'")
858 def test_58_check_update_openroadm_topo(self):
859 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
860 self.assertEqual(response['status_code'], requests.codes.ok)
861 node_list = response['network'][0]['node']
863 for node in node_list:
864 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
865 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
866 tp_list = node['ietf-network-topology:termination-point']
868 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
869 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
870 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
873 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
874 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
875 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
877 link_list = response['network'][0]['ietf-network-topology:link']
879 for link in link_list:
880 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
881 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
882 self.assertEqual(nb_updated_link, 0, "No link should have been modified")
885 def test_59_check_update_tapi_neps(self):
886 self.node_details["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
887 self.node_details["node-id-or-name"] = "ROADM-A1+PHOTONIC_MEDIA"
888 response = test_utils.transportpce_api_rpc_request(
889 'tapi-topology', 'get-node-details', self.node_details)
890 self.assertEqual(response['status_code'], requests.codes.ok)
891 nep_list = response['output']['node']['owned-node-edge-point']
894 if 'SRG1-PP2-TXRX' in nep['name'][0]['value']:
895 self.assertEqual(nep['operational-state'], 'DISABLED',
896 "Operational State should be 'DISABLED'")
897 self.assertEqual(nep['administrative-state'], 'LOCKED',
898 "Administrative State should be 'LOCKED'")
901 self.assertEqual(nep['operational-state'], 'ENABLED',
902 "Operational State should be 'ENABLED'")
903 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
904 "Administrative State should be 'UNLOCKED'")
905 self.assertEqual(nb_updated_neps, 3, "Only three roadm neps should have been modified")
908 def test_60_check_update_tapi_links(self):
909 self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
910 response = test_utils.transportpce_api_rpc_request(
911 'tapi-topology', 'get-topology-details', self.tapi_topo)
913 self.assertEqual(response['status_code'], requests.codes.ok)
914 link_list = response['output']['topology']['link']
916 for link in link_list:
917 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP2-TXRX']):
918 self.assertEqual(link['operational-state'], 'DISABLED')
919 self.assertEqual(link['administrative-state'], 'LOCKED')
922 self.assertEqual(link['operational-state'], 'ENABLED')
923 self.assertEqual(link['administrative-state'], 'UNLOCKED')
924 self.assertEqual(nb_updated_link, 0,
925 "No link should have been modified")
928 def test_61_check_update_service1_ok(self):
929 self.test_12_get_service_Ethernet()
931 def test_62_check_update_connectivity_service_Ethernet_ok(self):
932 self.test_13_get_connectivity_service_Ethernet()
934 def test_63_delete_connectivity_service_Ethernet(self):
935 self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.eth)
936 response = test_utils.transportpce_api_rpc_request(
937 'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
938 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
939 time.sleep(self.WAITING)
941 def test_64_disconnect_xponders_from_roadm(self):
942 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
943 self.assertEqual(response['status_code'], requests.codes.ok)
944 links = response['network'][0]['ietf-network-topology:link']
946 if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
947 response = test_utils.del_ietf_network_link_request(
948 'openroadm-topology', link['link-id'], 'config')
949 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
951 def test_65_disconnect_XPDRA(self):
952 response = test_utils.unmount_device("XPDR-A1")
953 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
955 def test_66_disconnect_XPDRC(self):
956 response = test_utils.unmount_device("XPDR-C1")
957 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
959 def test_67_disconnect_ROADMA(self):
960 response = test_utils.unmount_device("ROADM-A1")
961 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
963 def test_68_disconnect_ROADMC(self):
964 response = test_utils.unmount_device("ROADM-C1")
965 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
968 if __name__ == "__main__":
969 unittest.main(verbosity=2)