3 #############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #############################################################################
12 # pylint: disable=no-member
13 # 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
26 class TransportOlmTesting(unittest.TestCase):
29 NODE_VERSION = '1.2.1'
33 cls.processes = test_utils.start_tpce()
34 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
35 ('roadma-full', cls.NODE_VERSION),
36 ('roadmc-full', cls.NODE_VERSION),
37 ('xpdrc', cls.NODE_VERSION)])
40 def tearDownClass(cls):
41 # pylint: disable=not-an-iterable
42 for process in cls.processes:
43 test_utils.shutdown_process(process)
44 print("all processes killed")
47 print("execution of {}".format(self.id().split(".")[-1]))
50 def test_01_xpdrA_device_connected(self):
51 response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION))
52 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
54 def test_02_xpdrC_device_connected(self):
55 response = test_utils.mount_device("XPDRC01", ('xpdrc', self.NODE_VERSION))
56 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
58 def test_03_rdmA_device_connected(self):
59 response = test_utils.mount_device("ROADMA01", ('roadma-full', self.NODE_VERSION))
60 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
62 def test_04_rdmC_device_connected(self):
63 response = test_utils.mount_device("ROADMC01", ('roadmc-full', self.NODE_VERSION))
64 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
66 def test_05_connect_xprdA_to_roadmA(self):
67 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
68 "ROADMA01", "1", "SRG1-PP1-TXRX")
69 self.assertEqual(response.status_code, requests.codes.ok)
71 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
73 def test_06_connect_roadmA_to_xpdrA(self):
74 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
75 "ROADMA01", "1", "SRG1-PP1-TXRX")
76 self.assertEqual(response.status_code, requests.codes.ok)
78 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
80 def test_07_connect_xprdC_to_roadmC(self):
81 response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
82 "ROADMC01", "1", "SRG1-PP1-TXRX")
83 self.assertEqual(response.status_code, requests.codes.ok)
85 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
87 def test_08_connect_roadmC_to_xpdrC(self):
88 response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
89 "ROADMC01", "1", "SRG1-PP1-TXRX")
90 self.assertEqual(response.status_code, requests.codes.ok)
92 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
94 def test_09_create_OTS_ROADMA(self):
95 response = test_utils.create_ots_oms_request("ROADMA01", "DEG1-TTP-TXRX")
96 self.assertEqual(response.status_code, requests.codes.ok)
98 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
99 res["output"]["result"])
101 def test_10_create_OTS_ROADMC(self):
102 response = test_utils.create_ots_oms_request("ROADMC01", "DEG2-TTP-TXRX")
103 self.assertEqual(response.status_code, requests.codes.ok)
104 res = response.json()
105 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
106 res["output"]["result"])
108 def test_11_get_PM_ROADMA(self):
109 url = "{}/operations/transportpce-olm:get-pm"
112 "node-id": "ROADMA01",
113 "resource-type": "interface",
114 "granularity": "15min",
115 "resource-identifier": {
116 "resource-name": "OTS-DEG1-TTP-TXRX"
120 response = test_utils.post_request(url, data)
121 self.assertEqual(response.status_code, requests.codes.ok)
122 res = response.json()
124 "pmparameter-name": "OpticalPowerOutput",
125 "pmparameter-value": "2.5"
126 }, res["output"]["measurements"])
128 "pmparameter-name": "OpticalReturnLoss",
129 "pmparameter-value": "49.9"
130 }, res["output"]["measurements"])
132 "pmparameter-name": "OpticalPowerInput",
133 "pmparameter-value": "3"
134 }, res["output"]["measurements"])
136 def test_12_get_PM_ROADMC(self):
137 url = "{}/operations/transportpce-olm:get-pm"
140 "node-id": "ROADMC01",
141 "resource-type": "interface",
142 "granularity": "15min",
143 "resource-identifier": {
144 "resource-name": "OTS-DEG2-TTP-TXRX"
148 response = test_utils.post_request(url, data)
149 self.assertEqual(response.status_code, requests.codes.ok)
150 res = response.json()
152 "pmparameter-name": "OpticalPowerOutput",
153 "pmparameter-value": "18.1"
154 }, res["output"]["measurements"])
156 "pmparameter-name": "OpticalReturnLoss",
157 "pmparameter-value": "48.8"
158 }, res["output"]["measurements"])
160 "pmparameter-name": "OpticalPowerInput",
161 "pmparameter-value": "-3.2"
162 }, res["output"]["measurements"])
164 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
165 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
169 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
172 response = test_utils.post_request(url, data)
173 self.assertEqual(response.status_code, requests.codes.ok)
174 res = response.json()
175 self.assertIn('Success',
176 res["output"]["result"])
179 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
180 }, res["output"]["spans"])
183 def test_14_calculate_span_loss_base_all(self):
184 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
190 response = test_utils.post_request(url, data)
191 self.assertEqual(response.status_code, requests.codes.ok)
192 res = response.json()
193 self.assertIn('Success',
194 res["output"]["result"])
197 "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
198 }, res["output"]["spans"])
201 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
202 }, res["output"]["spans"])
205 def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
206 response = test_utils.check_netconf_node_request(
208 "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
209 self.assertEqual(response.status_code, requests.codes.ok)
210 res = response.json()
211 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
212 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
214 def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
215 response = test_utils.check_netconf_node_request(
217 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
218 self.assertEqual(response.status_code, requests.codes.ok)
219 res = response.json()
220 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
221 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
223 def test_17_servicePath_create_AToZ(self):
224 response = test_utils.service_path_request("create", "test", "1",
225 [{"node-id": "XPDRA01",
226 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
227 {"node-id": "ROADMA01",
228 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
229 {"node-id": "ROADMC01",
230 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
231 {"node-id": "XPDRC01",
232 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
233 196.1, 40, 196.075, 196.125, 761,
235 self.assertEqual(response.status_code, requests.codes.ok)
236 res = response.json()
237 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
241 def test_18_servicePath_create_ZToA(self):
242 response = test_utils.service_path_request("create", "test", "1",
243 [{"node-id": "XPDRC01",
244 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
245 {"node-id": "ROADMC01",
246 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
247 {"node-id": "ROADMA01",
248 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
249 {"node-id": "XPDRA01",
250 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
251 196.1, 40, 196.075, 196.125, 761,
253 self.assertEqual(response.status_code, requests.codes.ok)
254 res = response.json()
255 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
259 def test_19_service_power_setup_XPDRA_XPDRC(self):
260 url = "{}/operations/transportpce-olm:service-power-setup"
263 "service-name": "test",
267 "dest-tp": "XPDR1-NETWORK1",
268 "src-tp": "XPDR1-CLIENT1",
272 "dest-tp": "DEG1-TTP-TXRX",
273 "src-tp": "SRG1-PP1-TXRX",
274 "node-id": "ROADMA01"
277 "dest-tp": "SRG1-PP1-TXRX",
278 "src-tp": "DEG2-TTP-TXRX",
279 "node-id": "ROADMC01"
282 "dest-tp": "XPDR1-CLIENT1",
283 "src-tp": "XPDR1-NETWORK1",
287 "center-freq": 196.1,
291 "lower-spectral-slot-number": 761,
292 "higher-spectral-slot-number": 768
295 response = test_utils.post_request(url, data)
296 self.assertEqual(response.status_code, requests.codes.ok)
297 res = response.json()
298 self.assertIn('Success', res["output"]["result"])
300 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
301 response = test_utils.check_netconf_node_request(
303 "interface/XPDR1-NETWORK1-761:768/org-openroadm-optical-channel-interfaces:och")
304 self.assertEqual(response.status_code, requests.codes.ok)
305 res = response.json()
306 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
307 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
309 def test_21_get_roadmconnection_ROADMA(self):
310 response = test_utils.check_netconf_node_request(
311 "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
312 self.assertEqual(response.status_code, requests.codes.ok)
313 res = response.json()
314 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
315 self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
317 def test_22_get_roadmconnection_ROADMC(self):
318 response = test_utils.check_netconf_node_request(
319 "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
320 self.assertEqual(response.status_code, requests.codes.ok)
321 res = response.json()
322 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
324 def test_23_service_power_setup_XPDRC_XPDRA(self):
325 url = "{}/operations/transportpce-olm:service-power-setup"
328 "service-name": "test",
332 "dest-tp": "XPDR1-NETWORK1",
333 "src-tp": "XPDR1-CLIENT1",
337 "dest-tp": "DEG2-TTP-TXRX",
338 "src-tp": "SRG1-PP1-TXRX",
339 "node-id": "ROADMC01"
342 "src-tp": "DEG1-TTP-TXRX",
343 "dest-tp": "SRG1-PP1-TXRX",
344 "node-id": "ROADMA01"
347 "src-tp": "XPDR1-NETWORK1",
348 "dest-tp": "XPDR1-CLIENT1",
352 "center-freq": 196.1,
356 "lower-spectral-slot-number": 761,
357 "higher-spectral-slot-number": 768
360 response = test_utils.post_request(url, data)
361 self.assertEqual(response.status_code, requests.codes.ok)
362 res = response.json()
363 self.assertIn('Success', res["output"]["result"])
365 def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
366 response = test_utils.check_netconf_node_request(
368 "interface/XPDR1-NETWORK1-761:768/org-openroadm-optical-channel-interfaces:och")
369 self.assertEqual(response.status_code, requests.codes.ok)
370 res = response.json()
371 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
372 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
374 def test_25_get_roadmconnection_ROADMC(self):
375 response = test_utils.check_netconf_node_request(
376 "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
377 self.assertEqual(response.status_code, requests.codes.ok)
378 res = response.json()
379 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
380 self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
382 def test_26_service_power_turndown_XPDRA_XPDRC(self):
383 url = "{}/operations/transportpce-olm:service-power-turndown"
386 "service-name": "test",
390 "dest-tp": "XPDR1-NETWORK1",
391 "src-tp": "XPDR1-CLIENT1",
395 "dest-tp": "DEG1-TTP-TXRX",
396 "src-tp": "SRG1-PP1-TXRX",
397 "node-id": "ROADMA01"
400 "dest-tp": "SRG1-PP1-TXRX",
401 "src-tp": "DEG2-TTP-TXRX",
402 "node-id": "ROADMC01"
405 "dest-tp": "XPDR1-CLIENT1",
406 "src-tp": "XPDR1-NETWORK1",
410 "center-freq": 196.1,
414 "lower-spectral-slot-number": 761,
415 "higher-spectral-slot-number": 768
418 response = test_utils.post_request(url, data)
419 print(response.json())
420 self.assertEqual(response.status_code, requests.codes.ok)
421 res = response.json()
422 self.assertIn('Success', res["output"]["result"])
424 def test_27_get_roadmconnection_ROADMA(self):
425 response = test_utils.check_netconf_node_request(
426 "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
427 self.assertEqual(response.status_code, requests.codes.ok)
428 res = response.json()
429 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
430 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
432 def test_28_get_roadmconnection_ROADMC(self):
433 response = test_utils.check_netconf_node_request(
434 "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
435 self.assertEqual(response.status_code, requests.codes.ok)
436 res = response.json()
437 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
439 def test_29_servicePath_delete_AToZ(self):
440 response = test_utils.service_path_request("delete", "test", "1",
441 [{"node-id": "XPDRA01",
442 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
443 {"node-id": "ROADMA01",
444 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
445 {"node-id": "ROADMC01",
446 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
447 {"node-id": "XPDRC01",
448 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
449 196.1, 40, 196.075, 196.125, 761,
451 self.assertEqual(response.status_code, requests.codes.ok)
452 res = response.json()
453 self.assertIn('Request processed', res["output"]["result"])
456 def test_30_servicePath_delete_ZToA(self):
457 response = test_utils.service_path_request("delete", "test", "1",
458 [{"node-id": "XPDRC01",
459 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
460 {"node-id": "ROADMC01",
461 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
462 {"node-id": "ROADMA01",
463 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
464 {"node-id": "XPDRA01",
465 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
466 196.1, 40, 196.075, 196.125, 761,
468 self.assertEqual(response.status_code, requests.codes.ok)
469 res = response.json()
470 self.assertIn('Request processed', res["output"]["result"])
473 #"""to test case where SRG where the xpdr is connected to has no optical range data"""
475 def test_31_connect_xprdA_to_roadmA(self):
476 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
477 "ROADMA01", "1", "SRG1-PP2-TXRX")
478 self.assertEqual(response.status_code, requests.codes.ok)
479 res = response.json()
480 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
482 def test_32_connect_roadmA_to_xpdrA(self):
483 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
484 "ROADMA01", "1", "SRG1-PP2-TXRX")
485 self.assertEqual(response.status_code, requests.codes.ok)
486 res = response.json()
487 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
489 def test_33_servicePath_create_AToZ(self):
490 response = test_utils.service_path_request("create", "test2", "2",
491 [{"node-id": "XPDRA01",
492 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
493 {"node-id": "ROADMA01",
494 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
495 196.05, 40, 196.025, 196.075, 753,
497 self.assertEqual(response.status_code, requests.codes.ok)
498 res = response.json()
499 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
503 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
504 response = test_utils.check_netconf_node_request(
506 "interface/XPDR1-NETWORK2-753:760/org-openroadm-optical-channel-interfaces:och")
507 self.assertEqual(response.status_code, requests.codes.ok)
508 res = response.json()
509 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
510 self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
512 def test_35_servicePath_delete_AToZ(self):
513 response = test_utils.service_path_request("delete", "test", "1",
514 [{"node-id": "XPDRA01",
515 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
516 {"node-id": "ROADMA01",
517 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
518 196.1, 40, 196.075, 196.125, 761,
520 self.assertEqual(response.status_code, requests.codes.ok)
521 res = response.json()
522 self.assertIn('Request processed', res["output"]["result"])
525 def test_36_xpdrA_device_disconnected(self):
526 response = test_utils.unmount_device("XPDRA01")
527 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
529 def test_37_xpdrC_device_disconnected(self):
530 response = test_utils.unmount_device("XPDRC01")
531 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
533 def test_38_calculate_span_loss_current(self):
534 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
535 response = test_utils.post_request(url, None)
536 self.assertEqual(response.status_code, requests.codes.ok)
537 res = response.json()
538 self.assertIn('Success',
539 res["output"]["result"])
542 def test_39_rdmA_device_disconnected(self):
543 response = test_utils.unmount_device("ROADMA01")
544 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
546 def test_40_rdmC_device_disconnected(self):
547 response = test_utils.unmount_device("ROADMC01")
548 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
551 if __name__ == "__main__":
552 unittest.main(verbosity=2)