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 #############################################################################
16 from common import test_utils
19 class TransportOlmTesting(unittest.TestCase):
25 cls.processes = test_utils.start_tpce()
26 cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc'])
29 def tearDownClass(cls):
30 for process in cls.processes:
31 test_utils.shutdown_process(process)
32 print("all processes killed")
35 print("execution of {}".format(self.id().split(".")[-1]))
38 def test_01_xpdrA_device_connected(self):
39 response = test_utils.mount_device("XPDRA01", 'xpdra')
40 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
42 def test_02_xpdrC_device_connected(self):
43 response = test_utils.mount_device("XPDRC01", 'xpdrc')
44 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
46 def test_03_rdmA_device_connected(self):
47 response = test_utils.mount_device("ROADMA01", 'roadma-full')
48 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
50 def test_04_rdmC_device_connected(self):
51 response = test_utils.mount_device("ROADMC01", 'roadmc-full')
52 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
54 def test_05_connect_xprdA_to_roadmA(self):
55 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
56 "ROADMA01", "1", "SRG1-PP1-TXRX")
57 self.assertEqual(response.status_code, requests.codes.ok)
59 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
61 def test_06_connect_roadmA_to_xpdrA(self):
62 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
63 "ROADMA01", "1", "SRG1-PP1-TXRX")
64 self.assertEqual(response.status_code, requests.codes.ok)
66 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
68 def test_07_connect_xprdC_to_roadmC(self):
69 response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
70 "ROADMC01", "1", "SRG1-PP1-TXRX")
71 self.assertEqual(response.status_code, requests.codes.ok)
73 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
75 def test_08_connect_roadmC_to_xpdrC(self):
76 response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
77 "ROADMC01", "1", "SRG1-PP1-TXRX")
78 self.assertEqual(response.status_code, requests.codes.ok)
80 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
82 def test_09_create_OTS_ROADMA(self):
83 url = "{}/operations/transportpce-device-renderer:create-ots-oms"
86 "node-id": "ROADMA01",
87 "logical-connection-point": "DEG1-TTP-TXRX"
90 response = test_utils.post_request(url, data)
91 self.assertEqual(response.status_code, requests.codes.ok)
93 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
94 res["output"]["result"])
96 def test_10_create_OTS_ROADMC(self):
97 url = "{}/operations/transportpce-device-renderer:create-ots-oms"
100 "node-id": "ROADMC01",
101 "logical-connection-point": "DEG2-TTP-TXRX"
104 response = test_utils.post_request(url, data)
105 self.assertEqual(response.status_code, requests.codes.ok)
106 res = response.json()
107 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
108 res["output"]["result"])
110 def test_11_get_PM_ROADMA(self):
111 url = "{}/operations/transportpce-olm:get-pm"
114 "node-id": "ROADMA01",
115 "resource-type": "interface",
116 "granularity": "15min",
117 "resource-identifier": {
118 "resource-name": "OTS-DEG1-TTP-TXRX"
122 response = test_utils.post_request(url, data)
123 self.assertEqual(response.status_code, requests.codes.ok)
124 res = response.json()
126 "pmparameter-name": "OpticalPowerOutput",
127 "pmparameter-value": "2.5"
128 }, res["output"]["measurements"])
130 "pmparameter-name": "OpticalReturnLoss",
131 "pmparameter-value": "49.9"
132 }, res["output"]["measurements"])
134 "pmparameter-name": "OpticalPowerInput",
135 "pmparameter-value": "3"
136 }, res["output"]["measurements"])
138 def test_12_get_PM_ROADMC(self):
139 url = "{}/operations/transportpce-olm:get-pm"
142 "node-id": "ROADMC01",
143 "resource-type": "interface",
144 "granularity": "15min",
145 "resource-identifier": {
146 "resource-name": "OTS-DEG2-TTP-TXRX"
150 response = test_utils.post_request(url, data)
151 self.assertEqual(response.status_code, requests.codes.ok)
152 res = response.json()
154 "pmparameter-name": "OpticalPowerOutput",
155 "pmparameter-value": "18.1"
156 }, res["output"]["measurements"])
158 "pmparameter-name": "OpticalReturnLoss",
159 "pmparameter-value": "48.8"
160 }, res["output"]["measurements"])
162 "pmparameter-name": "OpticalPowerInput",
163 "pmparameter-value": "-3.2"
164 }, res["output"]["measurements"])
166 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
167 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
171 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
174 response = test_utils.post_request(url, data)
175 self.assertEqual(response.status_code, requests.codes.ok)
176 res = response.json()
177 self.assertIn('Success',
178 res["output"]["result"])
181 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
182 }, res["output"]["spans"])
185 def test_14_calculate_span_loss_base_all(self):
186 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
192 response = test_utils.post_request(url, data)
193 self.assertEqual(response.status_code, requests.codes.ok)
194 res = response.json()
195 self.assertIn('Success',
196 res["output"]["result"])
199 "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
200 }, res["output"]["spans"])
203 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
204 }, res["output"]["spans"])
207 def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
208 response = test_utils.check_netconf_node_request(
210 "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
211 self.assertEqual(response.status_code, requests.codes.ok)
212 res = response.json()
213 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
214 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
216 def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
217 response = test_utils.check_netconf_node_request(
219 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
220 self.assertEqual(response.status_code, requests.codes.ok)
221 res = response.json()
222 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
223 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
225 def test_17_servicePath_create_AToZ(self):
226 url = "{}/operations/transportpce-device-renderer:service-path"
229 "service-name": "test",
231 "modulation-format": "qpsk",
232 "operation": "create",
235 "dest-tp": "XPDR1-NETWORK1",
236 "src-tp": "XPDR1-CLIENT1",
240 "dest-tp": "DEG1-TTP-TXRX",
241 "src-tp": "SRG1-PP1-TXRX",
242 "node-id": "ROADMA01"
245 "dest-tp": "SRG1-PP1-TXRX",
246 "src-tp": "DEG2-TTP-TXRX",
247 "node-id": "ROADMC01"
250 "dest-tp": "XPDR1-CLIENT1",
251 "src-tp": "XPDR1-NETWORK1",
257 response = test_utils.post_request(url, data)
258 self.assertEqual(response.status_code, requests.codes.ok)
259 res = response.json()
260 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
264 def test_18_servicePath_create_ZToA(self):
265 url = "{}/operations/transportpce-device-renderer:service-path"
268 "service-name": "test",
270 "modulation-format": "qpsk",
271 "operation": "create",
274 "dest-tp": "XPDR1-NETWORK1",
275 "src-tp": "XPDR1-CLIENT1",
279 "dest-tp": "DEG2-TTP-TXRX",
280 "src-tp": "SRG1-PP1-TXRX",
281 "node-id": "ROADMC01"
284 "src-tp": "DEG1-TTP-TXRX",
285 "dest-tp": "SRG1-PP1-TXRX",
286 "node-id": "ROADMA01"
289 "src-tp": "XPDR1-NETWORK1",
290 "dest-tp": "XPDR1-CLIENT1",
296 response = test_utils.post_request(url, data)
297 self.assertEqual(response.status_code, requests.codes.ok)
298 res = response.json()
299 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
303 def test_19_service_power_setup_XPDRA_XPDRC(self):
304 url = "{}/operations/transportpce-olm:service-power-setup"
307 "service-name": "test",
311 "dest-tp": "XPDR1-NETWORK1",
312 "src-tp": "XPDR1-CLIENT1",
316 "dest-tp": "DEG1-TTP-TXRX",
317 "src-tp": "SRG1-PP1-TXRX",
318 "node-id": "ROADMA01"
321 "dest-tp": "SRG1-PP1-TXRX",
322 "src-tp": "DEG2-TTP-TXRX",
323 "node-id": "ROADMC01"
326 "dest-tp": "XPDR1-CLIENT1",
327 "src-tp": "XPDR1-NETWORK1",
333 response = test_utils.post_request(url, data)
334 self.assertEqual(response.status_code, requests.codes.ok)
335 res = response.json()
336 self.assertIn('Success', res["output"]["result"])
338 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
339 response = test_utils.check_netconf_node_request(
341 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
342 self.assertEqual(response.status_code, requests.codes.ok)
343 res = response.json()
344 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
345 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
347 def test_21_get_roadmconnection_ROADMA(self):
348 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
349 self.assertEqual(response.status_code, requests.codes.ok)
350 res = response.json()
351 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
352 self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
354 def test_22_get_roadmconnection_ROADMC(self):
355 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
356 self.assertEqual(response.status_code, requests.codes.ok)
357 res = response.json()
358 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
360 def test_23_service_power_setup_XPDRC_XPDRA(self):
361 url = "{}/operations/transportpce-olm:service-power-setup"
364 "service-name": "test",
368 "dest-tp": "XPDR1-NETWORK1",
369 "src-tp": "XPDR1-CLIENT1",
373 "dest-tp": "DEG2-TTP-TXRX",
374 "src-tp": "SRG1-PP1-TXRX",
375 "node-id": "ROADMC01"
378 "src-tp": "DEG1-TTP-TXRX",
379 "dest-tp": "SRG1-PP1-TXRX",
380 "node-id": "ROADMA01"
383 "src-tp": "XPDR1-NETWORK1",
384 "dest-tp": "XPDR1-CLIENT1",
390 response = test_utils.post_request(url, data)
391 self.assertEqual(response.status_code, requests.codes.ok)
392 res = response.json()
393 self.assertIn('Success', res["output"]["result"])
395 def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
396 response = test_utils.check_netconf_node_request(
398 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
399 self.assertEqual(response.status_code, requests.codes.ok)
400 res = response.json()
401 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
402 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
404 def test_25_get_roadmconnection_ROADMC(self):
405 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
406 self.assertEqual(response.status_code, requests.codes.ok)
407 res = response.json()
408 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
409 self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
411 def test_26_service_power_turndown_XPDRA_XPDRC(self):
412 url = "{}/operations/transportpce-olm:service-power-turndown"
415 "service-name": "test",
419 "dest-tp": "XPDR1-NETWORK1",
420 "src-tp": "XPDR1-CLIENT1",
424 "dest-tp": "DEG1-TTP-TXRX",
425 "src-tp": "SRG1-PP1-TXRX",
426 "node-id": "ROADMA01"
429 "dest-tp": "SRG1-PP1-TXRX",
430 "src-tp": "DEG2-TTP-TXRX",
431 "node-id": "ROADMC01"
434 "dest-tp": "XPDR1-CLIENT1",
435 "src-tp": "XPDR1-NETWORK1",
441 response = test_utils.post_request(url, data)
442 self.assertEqual(response.status_code, requests.codes.ok)
443 res = response.json()
444 self.assertIn('Success', res["output"]["result"])
446 def test_27_get_roadmconnection_ROADMA(self):
447 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
448 self.assertEqual(response.status_code, requests.codes.ok)
449 res = response.json()
450 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
451 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
453 def test_28_get_roadmconnection_ROADMC(self):
454 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
455 self.assertEqual(response.status_code, requests.codes.ok)
456 res = response.json()
457 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
459 def test_29_servicePath_delete_AToZ(self):
460 url = "{}/operations/transportpce-device-renderer:service-path"
463 "service-name": "test",
465 "modulation-format": "qpsk",
466 "operation": "delete",
469 "dest-tp": "XPDR1-NETWORK1",
470 "src-tp": "XPDR1-CLIENT1",
474 "dest-tp": "DEG1-TTP-TXRX",
475 "src-tp": "SRG1-PP1-TXRX",
476 "node-id": "ROADMA01"
479 "dest-tp": "SRG1-PP1-TXRX",
480 "src-tp": "DEG2-TTP-TXRX",
481 "node-id": "ROADMC01"
484 "dest-tp": "XPDR1-CLIENT1",
485 "src-tp": "XPDR1-NETWORK1",
491 response = test_utils.post_request(url, data)
492 self.assertEqual(response.status_code, requests.codes.ok)
493 res = response.json()
494 self.assertIn('Request processed', res["output"]["result"])
497 def test_30_servicePath_delete_ZToA(self):
498 url = "{}/operations/transportpce-device-renderer:service-path"
501 "service-name": "test",
503 "modulation-format": "qpsk",
504 "operation": "delete",
507 "dest-tp": "XPDR1-NETWORK1",
508 "src-tp": "XPDR1-CLIENT1",
512 "dest-tp": "DEG2-TTP-TXRX",
513 "src-tp": "SRG1-PP1-TXRX",
514 "node-id": "ROADMC01"
517 "src-tp": "DEG1-TTP-TXRX",
518 "dest-tp": "SRG1-PP1-TXRX",
519 "node-id": "ROADMA01"
522 "src-tp": "XPDR1-NETWORK1",
523 "dest-tp": "XPDR1-CLIENT1",
529 response = test_utils.post_request(url, data)
530 self.assertEqual(response.status_code, requests.codes.ok)
531 res = response.json()
532 self.assertIn('Request processed', res["output"]["result"])
535 """to test case where SRG where the xpdr is connected to has no optical range data"""
537 def test_31_connect_xprdA_to_roadmA(self):
538 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
539 "ROADMA01", "1", "SRG1-PP2-TXRX")
540 self.assertEqual(response.status_code, requests.codes.ok)
541 res = response.json()
542 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
544 def test_32_connect_roadmA_to_xpdrA(self):
545 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
546 "ROADMA01", "1", "SRG1-PP2-TXRX")
547 self.assertEqual(response.status_code, requests.codes.ok)
548 res = response.json()
549 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
551 def test_33_servicePath_create_AToZ(self):
552 url = "{}/operations/transportpce-device-renderer:service-path"
555 "service-name": "test2",
557 "modulation-format": "qpsk",
558 "operation": "create",
561 "dest-tp": "XPDR1-NETWORK2",
562 "src-tp": "XPDR1-CLIENT2",
566 "dest-tp": "DEG1-TTP-TXRX",
567 "src-tp": "SRG1-PP2-TXRX",
568 "node-id": "ROADMA01"
573 response = test_utils.post_request(url, data)
574 self.assertEqual(response.status_code, requests.codes.ok)
575 res = response.json()
576 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
580 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
581 response = test_utils.check_netconf_node_request(
583 "interface/XPDR1-NETWORK2-2/org-openroadm-optical-channel-interfaces:och")
584 self.assertEqual(response.status_code, requests.codes.ok)
585 res = response.json()
586 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
587 self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
589 def test_35_servicePath_delete_AToZ(self):
590 url = "{}/operations/transportpce-device-renderer:service-path"
593 "service-name": "test",
595 "modulation-format": "qpsk",
596 "operation": "delete",
599 "dest-tp": "XPDR1-NETWORK2",
600 "src-tp": "XPDR1-CLIENT2",
604 "dest-tp": "DEG1-TTP-TXRX",
605 "src-tp": "SRG1-PP2-TXRX",
606 "node-id": "ROADMA01"
611 response = test_utils.post_request(url, data)
612 self.assertEqual(response.status_code, requests.codes.ok)
613 res = response.json()
614 self.assertIn('Request processed', res["output"]["result"])
617 def test_36_xpdrA_device_disconnected(self):
618 response = test_utils.unmount_device("XPDRA01")
619 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
621 def test_37_xpdrC_device_disconnected(self):
622 response = test_utils.unmount_device("XPDRC01")
623 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
625 def test_38_calculate_span_loss_current(self):
626 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
627 response = test_utils.post_request(url, None)
628 self.assertEqual(response.status_code, requests.codes.ok)
629 res = response.json()
630 self.assertIn('Success',
631 res["output"]["result"])
634 def test_39_rdmA_device_disconnected(self):
635 response = test_utils.unmount_device("ROADMA01")
636 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
638 def test_40_rdmC_device_disconnected(self):
639 response = test_utils.unmount_device("ROADMC01")
640 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
643 if __name__ == "__main__":
644 unittest.main(verbosity=2)