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
17 # pylint: disable=wrong-import-order
19 sys.path.append('transportpce_tests/common/')
20 # pylint: disable=wrong-import-position
21 # pylint: disable=import-error
22 import test_utils # nopep8
24 # pylint: disable=too-few-public-methods
29 # pylint: disable=invalid-name
34 # XPDR-A1-XPDR1+XPONDER Uuid is:
35 # 4378fc29-6408-39ec-8737-5008c3dc49e5
36 # XPDR-C1-XPDR1+XPONDER Uuid is:
37 # 1770bea4-b1da-3b20-abce-7d182c0ec0df
38 # ROADM-A1+PHOTONIC_MEDIA Uuid is:
39 # 3b726367-6f2d-3e3f-9033-d99b61459075
42 class TransportPCEFulltesting(unittest.TestCase):
44 cr_serv_input_data = {
47 "layer-protocol-name": "DSR",
48 "service-interface-point": {
49 "service-interface-point-uuid": "b1f4bd3b-7fa9-367b-a8ab-6e80293238df"
51 "administrative-state": "UNLOCKED",
52 "operational-state": "ENABLED",
53 "direction": "BIDIRECTIONAL",
55 "protection-role": "WORK",
56 "local-id": "XPDR-C1-XPDR1",
59 "value-name": "OpenROADM node id",
60 "value": "XPDR-C1-XPDR1"
65 "layer-protocol-name": "DSR",
66 "service-interface-point": {
67 "service-interface-point-uuid": "b5964ce9-274c-3f68-b4d1-83c0b61bc74e"
69 "administrative-state": "UNLOCKED",
70 "operational-state": "ENABLED",
71 "direction": "BIDIRECTIONAL",
73 "protection-role": "WORK",
74 "local-id": "XPDR-A1-XPDR1",
77 "value-name": "OpenROADM node id",
78 "value": "XPDR-A1-XPDR1"
83 "connectivity-constraint": {
84 # "service-layer": "ETH",
85 "service-type": "POINT_TO_POINT_CONNECTIVITY",
86 "service-level": "Some service-level",
87 "requested-capacity": {
90 "unit": "tapi-common:CAPACITY_UNIT_GBPS"
94 "topology-constraint": [
96 "local-id": "localIdTopoConstraint",
99 "value-name": "Dumb constraint",
100 "value": "for debug1"
106 "layer-protocol-name": "ETH"
109 del_serv_input_data = {"uuid": "TBD"}
111 tapi_topo = {"topology-id": "TBD"}
114 "topology-id": "TBD",
118 tapi_serv_details = {"uuid": "TBD"}
121 uuid_services = UuidServices()
122 WAITING = 25 # nominal value is 300
123 NODE_VERSION_221 = '2.2.1'
127 # pylint: disable=unsubscriptable-object
128 cls.init_failed = False
129 os.environ['JAVA_MIN_MEM'] = '1024M'
130 os.environ['JAVA_MAX_MEM'] = '4096M'
131 cls.processes = test_utils.start_tpce()
132 # TAPI feature is not installed by default in Karaf
133 if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
134 print("installing tapi feature...")
135 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
136 if result.returncode != 0:
137 cls.init_failed = True
138 print("Restarting OpenDaylight...")
139 test_utils.shutdown_process(cls.processes[0])
140 cls.processes[0] = test_utils.start_karaf()
141 test_utils.process_list[0] = cls.processes[0]
142 cls.init_failed = not test_utils.wait_until_log_contains(
143 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
145 print("tapi installation feature failed...")
146 test_utils.shutdown_process(cls.processes[0])
148 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
149 ('roadma', cls.NODE_VERSION_221),
150 ('roadmc', cls.NODE_VERSION_221),
151 ('xpdrc', cls.NODE_VERSION_221)])
154 def tearDownClass(cls):
155 # pylint: disable=not-an-iterable
156 for process in cls.processes:
157 test_utils.shutdown_process(process)
158 print("all processes killed")
160 def setUp(self): # instruction executed before each test method
161 # pylint: disable=consider-using-f-string
162 print("execution of {}".format(self.id().split(".")[-1]))
165 def test_01_connect_xpdrA(self):
166 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
167 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
169 def test_02_connect_xpdrC(self):
170 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
171 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
173 def test_03_connect_rdmA(self):
174 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
175 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
177 def test_04_connect_rdmC(self):
178 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
179 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
181 def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
182 response = test_utils.transportpce_api_rpc_request(
183 'transportpce-networkutils', 'init-xpdr-rdm-links',
184 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
185 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
186 self.assertEqual(response['status_code'], requests.codes.ok)
187 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
190 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
191 response = test_utils.transportpce_api_rpc_request(
192 'transportpce-networkutils', 'init-rdm-xpdr-links',
193 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
194 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
195 self.assertEqual(response['status_code'], requests.codes.ok)
196 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
199 def test_07_connect_xpdrC_N1_to_roadmC_PP1(self):
200 response = test_utils.transportpce_api_rpc_request(
201 'transportpce-networkutils', 'init-xpdr-rdm-links',
202 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
203 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
204 self.assertEqual(response['status_code'], requests.codes.ok)
205 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
208 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
209 response = test_utils.transportpce_api_rpc_request(
210 'transportpce-networkutils', 'init-rdm-xpdr-links',
211 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
212 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
213 self.assertEqual(response['status_code'], requests.codes.ok)
214 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
217 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
218 # Config ROADMA-ROADMC 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-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
231 self.assertEqual(response.status_code, requests.codes.created)
233 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
234 # Config ROADMC-ROADMA oms-attributes
236 "auto-spanloss": "true",
237 "spanloss-base": 11.4,
238 "spanloss-current": 12,
239 "engineered-spanloss": 12.2,
240 "link-concatenation": [{
243 "SRLG-length": 100000,
245 response = test_utils.add_oms_attr_request(
246 "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
247 self.assertEqual(response.status_code, requests.codes.created)
249 # test service-create for Eth service from xpdr to xpdr
250 def test_11_create_connectivity_service_Ethernet(self):
251 response = test_utils.transportpce_api_rpc_request(
252 'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
253 time.sleep(self.WAITING)
254 self.uuid_services.eth = response['output']['service']['uuid']
255 # pylint: disable=consider-using-f-string
257 input_dict_1 = {'administrative-state': 'LOCKED',
258 'lifecycle-state': 'PLANNED',
259 'operational-state': 'DISABLED',
260 # 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
261 # 'service-layer': 'ETH',
262 'layer-protocol-name': 'ETH',
263 # 'connectivity-direction': 'BIDIRECTIONAL'
264 'direction': 'BIDIRECTIONAL'
266 input_dict_2 = {'value-name': 'OpenROADM node id',
267 'value': 'XPDR-C1-XPDR1'}
268 input_dict_3 = {'value-name': 'OpenROADM node id',
269 'value': 'XPDR-A1-XPDR1'}
271 self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
272 response['output']['service'])
273 self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
274 response['output']['service']['end-point'][0]['name'][0])
275 self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
276 response['output']['service']['end-point'][1]['name'][0])
277 # If the gate fails is because of the waiting time not being enough
278 # time.sleep(self.WAITING)
280 def test_12_get_service_Ethernet(self):
281 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
282 self.assertEqual(response['status_code'], requests.codes.ok)
283 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
284 self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
285 self.assertEqual(response['services'][0]['connection-type'], 'service')
286 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
289 def test_13_get_connectivity_service_Ethernet(self):
290 self.tapi_serv_details["uuid"] = str(self.uuid_services.eth)
291 response = test_utils.transportpce_api_rpc_request(
292 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
293 self.assertEqual(response['status_code'], requests.codes.ok)
294 self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
295 self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
296 self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
297 self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
299 def test_14_change_status_line_port_xpdrc(self):
300 self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
304 "administrative-state": "outOfService",
305 "port-qual": "xpdr-network"
309 def test_15_check_update_portmapping(self):
310 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
311 self.assertEqual(response['status_code'], requests.codes.ok)
312 mapping_list = response['nodes'][0]['mapping']
313 for mapping in mapping_list:
314 if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
315 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
316 "Operational State should be 'OutOfService'")
317 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
318 "Administrative State should be 'OutOfService'")
320 self.assertEqual(mapping['port-oper-state'], 'InService',
321 "Operational State should be 'InService'")
322 self.assertEqual(mapping['port-admin-state'], 'InService',
323 "Administrative State should be 'InService'")
326 def test_16_check_update_openroadm_topo(self):
327 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
328 self.assertEqual(response['status_code'], requests.codes.ok)
329 node_list = response['network'][0]['node']
331 for node in node_list:
332 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
333 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
334 tp_list = node['ietf-network-topology:termination-point']
336 if node['node-id'] == 'XPDR-C1-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
337 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
338 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
341 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
342 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
343 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
345 link_list = response['network'][0]['ietf-network-topology:link']
346 updated_links = ['XPDR-C1-XPDR1-XPDR1-NETWORK1toROADM-C1-SRG1-SRG1-PP1-TXRX',
347 'ROADM-C1-SRG1-SRG1-PP1-TXRXtoXPDR-C1-XPDR1-XPDR1-NETWORK1']
349 for link in link_list:
350 if link['link-id'] in updated_links:
351 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
352 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
355 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
356 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
357 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
360 def test_17_check_update_tapi_neps(self):
361 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
362 # self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi"
363 self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df"
364 response = test_utils.transportpce_api_rpc_request(
365 'tapi-topology', 'get-node-details', self.node_details)
366 self.assertEqual(response['status_code'], requests.codes.ok)
367 nep_list = response['output']['node']['owned-node-edge-point']
370 if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
371 self.assertEqual(nep['operational-state'], 'DISABLED',
372 "Operational State should be 'DISABLED'")
373 self.assertEqual(nep['administrative-state'], 'LOCKED',
374 "Administrative State should be 'LOCKED'")
377 self.assertEqual(nep['operational-state'], 'ENABLED',
378 "Operational State should be 'ENABLED'")
379 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
380 "Administrative State should be 'UNLOCKED'")
381 self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR"
382 # response = test_utils.transportpce_api_rpc_request(
383 # 'tapi-topology', 'get-node-details', self.node_details)
384 # self.assertEqual(response['status_code'], requests.codes.ok)
385 # nep_list = response['output']['node']['owned-node-edge-point']
386 # for nep in nep_list:
387 # if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
388 # self.assertEqual(nep['operational-state'], 'DISABLED',
389 # "Operational State should be 'DISABLED'")
390 # self.assertEqual(nep['administrative-state'], 'LOCKED',
391 # "Administrative State should be 'LOCKED'")
392 # nb_updated_neps += 1
394 # self.assertEqual(nep['operational-state'], 'ENABLED',
395 # "Operational State should be 'ENABLED'")
396 # self.assertEqual(nep['administrative-state'], 'UNLOCKED',
397 # "Administrative State should be 'UNLOCKED'")
398 self.assertEqual(nb_updated_neps, 3, "3 xponder neps (OTS, OTSI_MC, eODU) should have been modified")
401 def test_18_check_update_tapi_links(self):
402 self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
403 response = test_utils.transportpce_api_rpc_request(
404 'tapi-topology', 'get-topology-details', self.tapi_topo)
406 self.assertEqual(response['status_code'], requests.codes.ok)
407 link_list = response['output']['topology']['link']
408 print(response['output']['topology']['link'])
410 for link in link_list:
411 if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']):
412 self.assertEqual(link['operational-state'], 'DISABLED')
413 self.assertEqual(link['administrative-state'], 'LOCKED')
416 self.assertEqual(link['operational-state'], 'ENABLED')
417 self.assertEqual(link['administrative-state'], 'UNLOCKED')
418 self.assertEqual(nb_updated_link, 1,
419 "Only one xponder-output/input bidirectional link should have been modified")
422 def test_19_check_update_service_Ethernet(self):
423 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
424 self.assertEqual(response['status_code'], requests.codes.ok)
425 self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
426 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
428 def test_20_check_update_connectivity_service_Ethernet(self):
429 self.tapi_serv_details["uuid"] = str(self.uuid_services.eth)
430 print(str(self.uuid_services.eth))
431 response = test_utils.transportpce_api_rpc_request(
432 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
433 self.assertEqual(response['status_code'], requests.codes.ok)
434 self.assertEqual(response['output']['service']['operational-state'], 'DISABLED')
435 self.assertEqual(response['output']['service']['administrative-state'], 'LOCKED')
438 def test_21_restore_status_line_port_xpdrc(self):
439 self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
443 "administrative-state": "inService",
444 "port-qual": "xpdr-network"
448 def test_22_check_update_portmapping_ok(self):
449 response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
450 self.assertEqual(response['status_code'], requests.codes.ok)
451 mapping_list = response['nodes'][0]['mapping']
452 for mapping in mapping_list:
453 self.assertEqual(mapping['port-oper-state'], 'InService',
454 "Operational State should be 'InService'")
455 self.assertEqual(mapping['port-admin-state'], 'InService',
456 "Administrative State should be 'InService'")
459 def test_23_check_update_openroadm_topo_ok(self):
460 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
461 self.assertEqual(response['status_code'], requests.codes.ok)
462 node_list = response['network'][0]['node']
463 for node in node_list:
464 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
465 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
466 tp_list = node['ietf-network-topology:termination-point']
468 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
469 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
471 link_list = response['network'][0]['ietf-network-topology:link']
472 for link in link_list:
473 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
474 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
477 def test_24_check_update_tapi_neps_ok(self):
478 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
479 # self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi"
480 self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df"
481 response = test_utils.transportpce_api_rpc_request(
482 'tapi-topology', 'get-node-details', self.node_details)
483 self.assertEqual(response['status_code'], requests.codes.ok)
484 nep_list = response['output']['node']['owned-node-edge-point']
486 self.assertEqual(nep['operational-state'], 'ENABLED',
487 "Operational State should be 'ENABLED'")
488 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
489 "Administrative State should be 'UNLOCKED'")
491 # self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR"
492 # response = test_utils.transportpce_api_rpc_request(
493 # 'tapi-topology', 'get-node-details', self.node_details)
494 # self.assertEqual(response['status_code'], requests.codes.ok)
495 # nep_list = response['output']['node']['owned-node-edge-point']
496 # for nep in nep_list:
497 # self.assertEqual(nep['operational-state'], 'ENABLED',
498 # "Operational State should be 'ENABLED'")
499 # self.assertEqual(nep['administrative-state'], 'UNLOCKED',
500 # "Administrative State should be 'UNLOCKED'")
503 def test_25_check_update_tapi_links_ok(self):
504 self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
505 response = test_utils.transportpce_api_rpc_request(
506 'tapi-topology', 'get-topology-details', self.tapi_topo)
508 link_list = response['output']['topology']['link']
509 for link in link_list:
510 self.assertEqual(link['operational-state'], 'ENABLED')
511 self.assertEqual(link['administrative-state'], 'UNLOCKED')
514 def test_26_check_update_service1_ok(self):
515 self.test_12_get_service_Ethernet()
517 def test_27_check_update_connectivity_service_Ethernet_ok(self):
518 self.test_13_get_connectivity_service_Ethernet()
520 def test_28_change_status_port_roadma_srg(self):
521 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
524 "logical-connection-point": "SRG1-PP1",
525 "port-type": "client",
526 "circuit-id": "SRG1",
527 "administrative-state": "outOfService",
528 "port-qual": "roadm-external"
532 def test_29_check_update_portmapping(self):
533 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
534 self.assertEqual(response['status_code'], requests.codes.ok)
535 mapping_list = response['nodes'][0]['mapping']
536 for mapping in mapping_list:
537 if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
538 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
539 "Operational State should be 'OutOfService'")
540 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
541 "Administrative State should be 'OutOfService'")
543 self.assertEqual(mapping['port-oper-state'], 'InService',
544 "Operational State should be 'InService'")
545 self.assertEqual(mapping['port-admin-state'], 'InService',
546 "Administrative State should be 'InService'")
549 def test_30_check_update_openroadm_topo(self):
550 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
551 self.assertEqual(response['status_code'], requests.codes.ok)
552 node_list = response['network'][0]['node']
554 for node in node_list:
555 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
556 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
557 tp_list = node['ietf-network-topology:termination-point']
559 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
560 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
561 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
564 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
565 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
566 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
568 link_list = response['network'][0]['ietf-network-topology:link']
569 updated_links = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
570 'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
572 for link in link_list:
573 if link['link-id'] in updated_links:
574 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
575 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
578 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
579 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
580 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
583 def test_31_check_update_tapi_neps(self):
584 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
585 # self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
586 self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
587 response = test_utils.transportpce_api_rpc_request(
588 'tapi-topology', 'get-node-details', self.node_details)
589 self.assertEqual(response['status_code'], requests.codes.ok)
590 nep_list = response['output']['node']['owned-node-edge-point']
593 if 'SRG1-PP1-TXRX' in nep['name'][0]['value']:
594 self.assertEqual(nep['operational-state'], 'DISABLED',
595 "Operational State should be 'DISABLED'")
596 self.assertEqual(nep['administrative-state'], 'LOCKED',
597 "Administrative State should be 'LOCKED'")
600 self.assertEqual(nep['operational-state'], 'ENABLED',
601 "Operational State should be 'ENABLED'")
602 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
603 "Administrative State should be 'UNLOCKED'")
604 self.assertEqual(nb_updated_neps, 2, "Only 2 roadm SRG-PP nep (OTS/MC)should have been modified")
607 def test_32_check_update_tapi_links(self):
608 self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
609 response = test_utils.transportpce_api_rpc_request(
610 'tapi-topology', 'get-topology-details', self.tapi_topo)
612 link_list = response['output']['topology']['link']
614 for link in link_list:
615 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP1-TXRX']):
616 self.assertEqual(link['operational-state'], 'DISABLED')
617 self.assertEqual(link['administrative-state'], 'LOCKED')
620 self.assertEqual(link['operational-state'], 'ENABLED')
621 self.assertEqual(link['administrative-state'], 'UNLOCKED')
622 self.assertEqual(nb_updated_link, 1,
623 "Only one xponder-output/input link should have been modified")
626 def test_33_check_update_service_Ethernet(self):
627 self.test_19_check_update_service_Ethernet()
629 def test_34_check_update_connectivity_service_Ethernet(self):
630 self.test_20_check_update_connectivity_service_Ethernet()
632 def test_35_restore_status_port_roadma_srg(self):
633 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
636 "logical-connection-point": "SRG1-PP1",
637 "port-type": "client",
638 "circuit-id": "SRG1",
639 "administrative-state": "inService",
640 "port-qual": "roadm-external"
644 def test_36_check_update_portmapping_ok(self):
645 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
646 self.assertEqual(response['status_code'], requests.codes.ok)
647 mapping_list = response['nodes'][0]['mapping']
648 for mapping in mapping_list:
649 self.assertEqual(mapping['port-oper-state'], 'InService',
650 "Operational State should be 'InService'")
651 self.assertEqual(mapping['port-admin-state'], 'InService',
652 "Administrative State should be 'InService'")
655 def test_37_check_update_openroadm_topo_ok(self):
656 self.test_23_check_update_openroadm_topo_ok()
658 def test_38_check_update_tapi_neps_ok(self):
659 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
660 # self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
661 self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
662 response = test_utils.transportpce_api_rpc_request(
663 'tapi-topology', 'get-node-details', self.node_details)
664 self.assertEqual(response['status_code'], requests.codes.ok)
665 nep_list = response['output']['node']['owned-node-edge-point']
667 self.assertEqual(nep['operational-state'], 'ENABLED',
668 "Operational State should be 'ENABLED'")
669 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
670 "Administrative State should be 'UNLOCKED'")
674 def test_39_check_update_tapi_links_ok(self):
675 self.test_25_check_update_tapi_links_ok()
677 def test_40_check_update_service1_ok(self):
678 self.test_12_get_service_Ethernet()
680 def test_41_check_update_connectivity_service_Ethernet_ok(self):
681 self.test_13_get_connectivity_service_Ethernet()
683 def test_42_change_status_line_port_roadma_deg(self):
684 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
687 "logical-connection-point": "DEG2-TTP-TXRX",
690 "administrative-state": "outOfService",
691 "port-qual": "roadm-external"
695 def test_43_check_update_portmapping(self):
696 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
697 self.assertEqual(response['status_code'], requests.codes.ok)
698 mapping_list = response['nodes'][0]['mapping']
699 for mapping in mapping_list:
700 if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
701 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
702 "Operational State should be 'OutOfService'")
703 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
704 "Administrative State should be 'OutOfService'")
706 self.assertEqual(mapping['port-oper-state'], 'InService',
707 "Operational State should be 'InService'")
708 self.assertEqual(mapping['port-admin-state'], 'InService',
709 "Administrative State should be 'InService'")
712 def test_44_check_update_openroadm_topo(self):
713 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
714 self.assertEqual(response['status_code'], requests.codes.ok)
715 node_list = response['network'][0]['node']
717 for node in node_list:
718 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
719 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
720 tp_list = node['ietf-network-topology:termination-point']
722 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
723 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
724 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
727 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
728 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
729 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
731 link_list = response['network'][0]['ietf-network-topology:link']
732 updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
733 'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
735 for link in link_list:
736 if link['link-id'] in updated_links:
737 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
738 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
741 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
742 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
743 self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
746 def test_45_check_update_tapi_neps(self):
747 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
748 # self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
749 self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
750 response = test_utils.transportpce_api_rpc_request(
751 'tapi-topology', 'get-node-details', self.node_details)
752 self.assertEqual(response['status_code'], requests.codes.ok)
753 nep_list = response['output']['node']['owned-node-edge-point']
756 if 'DEG2-TTP-TXRX' in nep['name'][0]['value']:
757 self.assertEqual(nep['operational-state'], 'DISABLED',
758 "Operational State should be 'DISABLED'")
759 self.assertEqual(nep['administrative-state'], 'LOCKED',
760 "Administrative State should be 'LOCKED'")
763 self.assertEqual(nep['operational-state'], 'ENABLED',
764 "Operational State should be 'ENABLED'")
765 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
766 "Administrative State should be 'UNLOCKED'")
767 self.assertEqual(nb_updated_neps, 4, "4 roadm NEPS should have been modified (OTS/OMS/MC/OTSI_MC")
770 def test_46_check_update_tapi_links(self):
771 self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
772 response = test_utils.transportpce_api_rpc_request(
773 'tapi-topology', 'get-topology-details', self.tapi_topo)
775 self.assertEqual(response['status_code'], requests.codes.ok)
776 link_list = response['output']['topology']['link']
778 for link in link_list:
779 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'DEG2-TTP-TXRX']):
780 self.assertEqual(link['operational-state'], 'DISABLED')
781 self.assertEqual(link['administrative-state'], 'LOCKED')
784 self.assertEqual(link['operational-state'], 'ENABLED')
785 self.assertEqual(link['administrative-state'], 'UNLOCKED')
786 self.assertEqual(nb_updated_link, 1,
787 "Only one rdm-rdm link should have been modified")
790 def test_47_check_update_service_Ethernet(self):
791 self.test_19_check_update_service_Ethernet()
793 def test_48_check_update_connectivity_service_Ethernet(self):
794 self.test_20_check_update_connectivity_service_Ethernet()
796 def test_49_restore_status_line_port_roadma_deg(self):
797 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
800 "logical-connection-point": "DEG2-TTP-TXRX",
803 "administrative-state": "inService",
804 "port-qual": "roadm-external"
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 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
830 "logical-connection-point": "SRG1-PP2",
831 "port-type": "client",
832 "circuit-id": "SRG1",
833 "administrative-state": "outOfService",
834 "port-qual": "roadm-external"
838 def test_57_check_update_portmapping(self):
839 response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
840 self.assertEqual(response['status_code'], requests.codes.ok)
841 mapping_list = response['nodes'][0]['mapping']
842 for mapping in mapping_list:
843 if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
844 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
845 "Operational State should be 'OutOfService'")
846 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
847 "Administrative State should be 'OutOfService'")
849 self.assertEqual(mapping['port-oper-state'], 'InService',
850 "Operational State should be 'InService'")
851 self.assertEqual(mapping['port-admin-state'], 'InService',
852 "Administrative State should be 'InService'")
855 def test_58_check_update_openroadm_topo(self):
856 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
857 self.assertEqual(response['status_code'], requests.codes.ok)
858 node_list = response['network'][0]['node']
860 for node in node_list:
861 self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
862 self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
863 tp_list = node['ietf-network-topology:termination-point']
865 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
866 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
867 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
870 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
871 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
872 self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
874 link_list = response['network'][0]['ietf-network-topology:link']
876 for link in link_list:
877 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
878 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
879 self.assertEqual(nb_updated_link, 0, "No link should have been modified")
882 def test_59_check_update_tapi_neps(self):
883 self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
884 # self.node_details["node-id"] = uuid.UUID(bytes="ROADM-A1+PHOTONIC_MEDIA".bytes)
885 self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
886 response = test_utils.transportpce_api_rpc_request(
887 'tapi-topology', 'get-node-details', self.node_details)
888 self.assertEqual(response['status_code'], requests.codes.ok)
889 nep_list = response['output']['node']['owned-node-edge-point']
892 if 'SRG1-PP2-TXRX' in nep['name'][0]['value']:
893 self.assertEqual(nep['operational-state'], 'DISABLED',
894 "Operational State should be 'DISABLED'")
895 self.assertEqual(nep['administrative-state'], 'LOCKED',
896 "Administrative State should be 'LOCKED'")
899 self.assertEqual(nep['operational-state'], 'ENABLED',
900 "Operational State should be 'ENABLED'")
901 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
902 "Administrative State should be 'UNLOCKED'")
903 self.assertEqual(nb_updated_neps, 1, "Only 1 roadm neps OTS should have been modified for SRG2PP")
906 def test_60_check_update_tapi_links(self):
907 self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
908 response = test_utils.transportpce_api_rpc_request(
909 'tapi-topology', 'get-topology-details', self.tapi_topo)
911 self.assertEqual(response['status_code'], requests.codes.ok)
912 link_list = response['output']['topology']['link']
914 for link in link_list:
915 if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP2-TXRX']):
916 self.assertEqual(link['operational-state'], 'DISABLED')
917 self.assertEqual(link['administrative-state'], 'LOCKED')
920 self.assertEqual(link['operational-state'], 'ENABLED')
921 self.assertEqual(link['administrative-state'], 'UNLOCKED')
922 self.assertEqual(nb_updated_link, 0,
923 "No link should have been modified")
926 def test_61_check_update_service1_ok(self):
927 self.test_12_get_service_Ethernet()
929 def test_62_check_update_connectivity_service_Ethernet_ok(self):
930 self.test_13_get_connectivity_service_Ethernet()
932 def test_63_delete_connectivity_service_Ethernet(self):
933 self.del_serv_input_data["uuid"] = str(self.uuid_services.eth)
934 response = test_utils.transportpce_api_rpc_request(
935 'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
936 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
937 time.sleep(self.WAITING)
939 def test_64_disconnect_xponders_from_roadm(self):
940 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
941 self.assertEqual(response['status_code'], requests.codes.ok)
942 links = response['network'][0]['ietf-network-topology:link']
944 if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
945 response = test_utils.del_ietf_network_link_request(
946 'openroadm-topology', link['link-id'], 'config')
947 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
949 def test_65_disconnect_XPDRA(self):
950 response = test_utils.unmount_device("XPDR-A1")
951 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
953 def test_66_disconnect_XPDRC(self):
954 response = test_utils.unmount_device("XPDR-C1")
955 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
957 def test_67_disconnect_ROADMA(self):
958 response = test_utils.unmount_device("ROADM-A1")
959 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
961 def test_68_disconnect_ROADMC(self):
962 response = test_utils.unmount_device("ROADM-C1")
963 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
965 def test_69_restore_status_port_roadma_srg(self):
966 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
969 "logical-connection-point": "SRG1-PP2",
970 "port-type": "client",
971 "circuit-id": "SRG1",
972 "administrative-state": "inService",
973 "port-qual": "roadm-external"
977 def test_70_clean_openroadm_topology(self):
978 response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
979 self.assertEqual(response['status_code'], requests.codes.ok)
980 links = response['network'][0]['ietf-network-topology:link']
982 if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT', 'ROADM-TO-ROADM'):
983 response = test_utils.del_ietf_network_link_request('openroadm-topology', link['link-id'], 'config')
984 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
987 if __name__ == "__main__":
988 unittest.main(verbosity=2)