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 from common import test_utils
21 class TransportOlmTesting(unittest.TestCase):
27 cls.processes = test_utils.start_tpce()
28 cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc'])
31 def tearDownClass(cls):
32 # pylint: disable=not-an-iterable
33 for process in cls.processes:
34 test_utils.shutdown_process(process)
35 print("all processes killed")
38 print("execution of {}".format(self.id().split(".")[-1]))
41 def test_01_xpdrA_device_connected(self):
42 response = test_utils.mount_device("XPDR-A1", 'xpdra')
43 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
45 def test_02_xpdrC_device_connected(self):
46 response = test_utils.mount_device("XPDR-C1", 'xpdrc')
47 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
49 def test_03_rdmA_device_connected(self):
50 response = test_utils.mount_device("ROADM-A1", 'roadma')
51 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
53 def test_04_rdmC_device_connected(self):
54 response = test_utils.mount_device("ROADM-C1", 'roadmc')
55 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
57 def test_05_connect_xprdA_to_roadmA(self):
58 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
59 "ROADM-A1", "1", "SRG1-PP1-TXRX")
60 self.assertEqual(response.status_code, requests.codes.ok)
62 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
64 def test_06_connect_roadmA_to_xpdrA(self):
65 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
66 "ROADM-A1", "1", "SRG1-PP1-TXRX")
67 self.assertEqual(response.status_code, requests.codes.ok)
69 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
71 def test_07_connect_xprdC_to_roadmC(self):
72 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
73 "ROADM-C1", "1", "SRG1-PP1-TXRX")
74 self.assertEqual(response.status_code, requests.codes.ok)
76 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
78 def test_08_connect_roadmC_to_xpdrC(self):
79 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
80 "ROADM-C1", "1", "SRG1-PP1-TXRX")
81 self.assertEqual(response.status_code, requests.codes.ok)
83 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
85 def test_09_create_OTS_ROADMA(self):
86 response = test_utils.create_ots_oms_request("ROADM-A1", "DEG1-TTP-TXRX")
88 self.assertEqual(response.status_code, requests.codes.ok)
90 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADM-A1',
91 res["output"]["result"])
93 def test_10_create_OTS_ROADMC(self):
94 response = test_utils.create_ots_oms_request("ROADM-C1", "DEG2-TTP-TXRX")
95 self.assertEqual(response.status_code, requests.codes.ok)
97 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADM-C1',
98 res["output"]["result"])
100 def test_11_get_PM_ROADMA(self):
101 url = "{}/operations/transportpce-olm:get-pm"
104 "node-id": "ROADM-A1",
105 "resource-type": "interface",
106 "granularity": "15min",
107 "resource-identifier": {
108 "resource-name": "OTS-DEG2-TTP-TXRX"
112 response = test_utils.post_request(url, data)
113 self.assertEqual(response.status_code, requests.codes.ok)
114 res = response.json()
116 "pmparameter-name": "OpticalPowerOutput",
117 "pmparameter-value": "2.5"
118 }, res["output"]["measurements"])
120 "pmparameter-name": "OpticalReturnLoss",
121 "pmparameter-value": "40"
122 }, res["output"]["measurements"])
124 "pmparameter-name": "OpticalPowerInput",
125 "pmparameter-value": "-21.1"
126 }, res["output"]["measurements"])
128 def test_12_get_PM_ROADMC(self):
129 url = "{}/operations/transportpce-olm:get-pm"
132 "node-id": "ROADM-C1",
133 "resource-type": "interface",
134 "granularity": "15min",
135 "resource-identifier": {
136 "resource-name": "OTS-DEG1-TTP-TXRX"
140 response = test_utils.post_request(url, data)
141 self.assertEqual(response.status_code, requests.codes.ok)
142 res = response.json()
144 "pmparameter-name": "OpticalPowerOutput",
145 "pmparameter-value": "4.6"
146 }, res["output"]["measurements"])
148 "pmparameter-name": "OpticalReturnLoss",
149 "pmparameter-value": "49.1"
150 }, res["output"]["measurements"])
152 "pmparameter-name": "OpticalPowerInput",
153 "pmparameter-value": "-15.1"
154 }, res["output"]["measurements"])
156 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
157 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
161 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
164 response = test_utils.post_request(url, data)
165 self.assertEqual(response.status_code, requests.codes.ok)
166 res = response.json()
167 self.assertIn('Success',
168 res["output"]["result"])
171 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
172 }, res["output"]["spans"])
175 def test_14_calculate_span_loss_base_all(self):
176 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
182 response = test_utils.post_request(url, data)
183 self.assertEqual(response.status_code, requests.codes.ok)
184 res = response.json()
185 self.assertIn('Success',
186 res["output"]["result"])
189 "link-id": "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX"
190 }, res["output"]["spans"])
193 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
194 }, res["output"]["spans"])
197 def test_15_get_OTS_DEG2_TTP_TXRX_ROADMA(self):
198 response = test_utils.check_netconf_node_request(
200 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
201 self.assertEqual(response.status_code, requests.codes.ok)
202 res = response.json()
203 self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
204 self.assertEqual(25.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
206 def test_16_get_OTS_DEG1_TTP_TXRX_ROADMC(self):
207 response = test_utils.check_netconf_node_request(
209 "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
210 self.assertEqual(response.status_code, requests.codes.ok)
211 res = response.json()
212 self.assertEqual(25.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
213 self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
215 def test_17_servicePath_create_AToZ(self):
216 response = test_utils.service_path_request("create", "test", "1",
217 [{"node-id": "XPDR-A1",
218 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
219 {"node-id": "ROADM-A1",
220 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
221 {"node-id": "ROADM-C1",
222 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG1-TTP-TXRX"},
223 {"node-id": "XPDR-C1",
224 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
225 196.1, 40, 196.075, 196.125, 761,
227 self.assertEqual(response.status_code, requests.codes.ok)
228 res = response.json()
229 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
233 def test_18_servicePath_create_ZToA(self):
234 response = test_utils.service_path_request("create", "test", "1",
235 [{"node-id": "XPDR-C1",
236 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
237 {"node-id": "ROADM-C1",
238 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
239 {"node-id": "ROADM-A1",
240 "src-tp": "DEG2-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
241 {"node-id": "XPDR-A1",
242 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
243 196.1, 40, 196.075, 196.125, 761,
245 self.assertEqual(response.status_code, requests.codes.ok)
246 res = response.json()
247 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
251 def test_19_service_power_setup_XPDRA_XPDRC(self):
252 url = "{}/operations/transportpce-olm:service-power-setup"
255 "service-name": "test",
259 "dest-tp": "XPDR1-NETWORK1",
260 "src-tp": "XPDR1-CLIENT1",
264 "dest-tp": "DEG2-TTP-TXRX",
265 "src-tp": "SRG1-PP1-TXRX",
266 "node-id": "ROADM-A1"
269 "dest-tp": "SRG1-PP1-TXRX",
270 "src-tp": "DEG1-TTP-TXRX",
271 "node-id": "ROADM-C1"
274 "dest-tp": "XPDR1-CLIENT1",
275 "src-tp": "XPDR1-NETWORK1",
279 "lower-spectral-slot-number": 761,
280 "higher-spectral-slot-number": 768
283 response = test_utils.post_request(url, data)
284 self.assertEqual(response.status_code, requests.codes.ok)
285 res = response.json()
286 self.assertIn('Success', res["output"]["result"])
288 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
289 response = test_utils.check_netconf_node_request(
291 "interface/XPDR1-NETWORK1-761:768/org-openroadm-optical-channel-interfaces:och")
292 self.assertEqual(response.status_code, requests.codes.ok)
293 res = response.json()
294 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
295 self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
297 def test_21_get_roadmconnection_ROADMA(self):
298 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
299 self.assertEqual(response.status_code, requests.codes.ok)
300 res = response.json()
301 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
302 self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power'])
304 def test_22_get_roadmconnection_ROADMC(self):
305 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
306 self.assertEqual(response.status_code, requests.codes.ok)
307 res = response.json()
308 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
310 def test_23_service_power_setup_XPDRC_XPDRA(self):
311 url = "{}/operations/transportpce-olm:service-power-setup"
314 "service-name": "test",
318 "dest-tp": "XPDR1-NETWORK1",
319 "src-tp": "XPDR1-CLIENT1",
323 "dest-tp": "DEG1-TTP-TXRX",
324 "src-tp": "SRG1-PP1-TXRX",
325 "node-id": "ROADM-C1"
328 "src-tp": "DEG2-TTP-TXRX",
329 "dest-tp": "SRG1-PP1-TXRX",
330 "node-id": "ROADM-A1"
333 "src-tp": "XPDR1-NETWORK1",
334 "dest-tp": "XPDR1-CLIENT1",
338 "lower-spectral-slot-number": 761,
339 "higher-spectral-slot-number": 768
342 response = test_utils.post_request(url, data)
343 self.assertEqual(response.status_code, requests.codes.ok)
344 res = response.json()
345 self.assertIn('Success', res["output"]["result"])
347 def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
348 response = test_utils.check_netconf_node_request(
350 "interface/XPDR1-NETWORK1-761:768/org-openroadm-optical-channel-interfaces:och")
351 self.assertEqual(response.status_code, requests.codes.ok)
352 res = response.json()
353 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
354 self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
356 def test_25_get_roadmconnection_ROADMC(self):
357 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
358 self.assertEqual(response.status_code, requests.codes.ok)
359 res = response.json()
360 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
361 self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power'])
363 def test_26_service_power_turndown_XPDRA_XPDRC(self):
364 url = "{}/operations/transportpce-olm:service-power-turndown"
367 "service-name": "test",
371 "dest-tp": "XPDR1-NETWORK1",
372 "src-tp": "XPDR1-CLIENT1",
376 "dest-tp": "DEG2-TTP-TXRX",
377 "src-tp": "SRG1-PP1-TXRX",
378 "node-id": "ROADM-A1"
381 "dest-tp": "SRG1-PP1-TXRX",
382 "src-tp": "DEG1-TTP-TXRX",
383 "node-id": "ROADM-C1"
386 "dest-tp": "XPDR1-CLIENT1",
387 "src-tp": "XPDR1-NETWORK1",
391 "lower-spectral-slot-number": 761,
392 "higher-spectral-slot-number": 768
395 response = test_utils.post_request(url, data)
396 self.assertEqual(response.status_code, requests.codes.ok)
397 res = response.json()
398 self.assertIn('Success', res["output"]["result"])
400 def test_27_get_roadmconnection_ROADMA(self):
401 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
402 self.assertEqual(response.status_code, requests.codes.ok)
403 res = response.json()
404 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
405 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
407 def test_28_get_roadmconnection_ROADMC(self):
408 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
409 self.assertEqual(response.status_code, requests.codes.ok)
410 res = response.json()
411 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
413 def test_29_servicePath_delete_AToZ(self):
414 response = test_utils.service_path_request("delete", "test", "1",
415 [{"node-id": "XPDR-A1",
416 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
417 {"node-id": "ROADM-A1",
418 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
419 {"node-id": "ROADM-C1",
420 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG1-TTP-TXRX"},
421 {"node-id": "XPDR-C1",
422 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
423 196.1, 40, 196.075, 196.125, 761,
425 self.assertEqual(response.status_code, requests.codes.ok)
426 res = response.json()
427 self.assertIn('Request processed', res["output"]["result"])
430 def test_30_servicePath_delete_ZToA(self):
431 response = test_utils.service_path_request("delete", "test", "1",
432 [{"node-id": "XPDR-C1",
433 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
434 {"node-id": "ROADM-C1",
435 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
436 {"node-id": "ROADM-A1",
437 "src-tp": "DEG2-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
438 {"node-id": "XPDR-A1",
439 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
440 196.053125, 40, 196.025, 196.08125, 761,
442 self.assertEqual(response.status_code, requests.codes.ok)
443 res = response.json()
444 self.assertIn('Request processed', res["output"]["result"])
447 #"""to test case where SRG where the xpdr is connected to has no optical range data"""
449 def test_31_connect_xprdA_to_roadmA(self):
450 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
451 "ROADM-A1", "1", "SRG1-PP2-TXRX")
452 self.assertEqual(response.status_code, requests.codes.ok)
453 res = response.json()
454 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
456 def test_32_connect_roadmA_to_xpdrA(self):
457 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
458 "ROADM-A1", "1", "SRG1-PP2-TXRX")
459 self.assertEqual(response.status_code, requests.codes.ok)
460 res = response.json()
461 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
463 def test_33_servicePath_create_AToZ(self):
464 response = test_utils.service_path_request("create", "test2", "2",
465 [{"node-id": "XPDR-A1",
466 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
467 {"node-id": "ROADM-A1",
468 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
469 196.1, 40, 196.075, 196.125, 753,
471 self.assertEqual(response.status_code, requests.codes.ok)
472 res = response.json()
473 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
477 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
478 response = test_utils.check_netconf_node_request(
480 "interface/XPDR1-NETWORK2-753:760/org-openroadm-optical-channel-interfaces:och")
481 self.assertEqual(response.status_code, requests.codes.ok)
482 res = response.json()
483 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
484 # self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
486 def test_35_servicePath_delete_AToZ(self):
487 response = test_utils.service_path_request("delete", "test", "1",
488 [{"node-id": "XPDR-A1",
489 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
490 {"node-id": "ROADM-A1",
491 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
492 196.053125, 40, 196.025, 196.08125, 761,
494 self.assertEqual(response.status_code, requests.codes.ok)
495 res = response.json()
496 self.assertIn('Request processed', res["output"]["result"])
499 def test_36_xpdrA_device_disconnected(self):
500 response = test_utils.unmount_device("XPDR-A1")
501 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
503 def test_37_xpdrC_device_disconnected(self):
504 response = test_utils.unmount_device("XPDR-C1")
505 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
507 def test_38_calculate_span_loss_current(self):
508 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
509 response = test_utils.post_request(url, None)
510 self.assertEqual(response.status_code, requests.codes.ok)
511 res = response.json()
512 self.assertIn('Success',
513 res["output"]["result"])
516 def test_39_rdmA_device_disconnected(self):
517 response = test_utils.unmount_device("ROADM-A1")
518 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
520 def test_40_rdmC_device_disconnected(self):
521 response = test_utils.unmount_device("ROADM-C1")
522 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
525 if __name__ == "__main__":
526 unittest.main(verbosity=2)