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 response = test_utils.create_ots_oms_request("ROADMA01", "DEG1-TTP-TXRX")
84 self.assertEqual(response.status_code, requests.codes.ok)
86 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
87 res["output"]["result"])
89 def test_10_create_OTS_ROADMC(self):
90 response = test_utils.create_ots_oms_request("ROADMC01", "DEG2-TTP-TXRX")
91 self.assertEqual(response.status_code, requests.codes.ok)
93 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
94 res["output"]["result"])
96 def test_11_get_PM_ROADMA(self):
97 url = "{}/operations/transportpce-olm:get-pm"
100 "node-id": "ROADMA01",
101 "resource-type": "interface",
102 "granularity": "15min",
103 "resource-identifier": {
104 "resource-name": "OTS-DEG1-TTP-TXRX"
108 response = test_utils.post_request(url, data)
109 self.assertEqual(response.status_code, requests.codes.ok)
110 res = response.json()
112 "pmparameter-name": "OpticalPowerOutput",
113 "pmparameter-value": "2.5"
114 }, res["output"]["measurements"])
116 "pmparameter-name": "OpticalReturnLoss",
117 "pmparameter-value": "49.9"
118 }, res["output"]["measurements"])
120 "pmparameter-name": "OpticalPowerInput",
121 "pmparameter-value": "3"
122 }, res["output"]["measurements"])
124 def test_12_get_PM_ROADMC(self):
125 url = "{}/operations/transportpce-olm:get-pm"
128 "node-id": "ROADMC01",
129 "resource-type": "interface",
130 "granularity": "15min",
131 "resource-identifier": {
132 "resource-name": "OTS-DEG2-TTP-TXRX"
136 response = test_utils.post_request(url, data)
137 self.assertEqual(response.status_code, requests.codes.ok)
138 res = response.json()
140 "pmparameter-name": "OpticalPowerOutput",
141 "pmparameter-value": "18.1"
142 }, res["output"]["measurements"])
144 "pmparameter-name": "OpticalReturnLoss",
145 "pmparameter-value": "48.8"
146 }, res["output"]["measurements"])
148 "pmparameter-name": "OpticalPowerInput",
149 "pmparameter-value": "-3.2"
150 }, res["output"]["measurements"])
152 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
153 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
157 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
160 response = test_utils.post_request(url, data)
161 self.assertEqual(response.status_code, requests.codes.ok)
162 res = response.json()
163 self.assertIn('Success',
164 res["output"]["result"])
167 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
168 }, res["output"]["spans"])
171 def test_14_calculate_span_loss_base_all(self):
172 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
178 response = test_utils.post_request(url, data)
179 self.assertEqual(response.status_code, requests.codes.ok)
180 res = response.json()
181 self.assertIn('Success',
182 res["output"]["result"])
185 "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
186 }, res["output"]["spans"])
189 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
190 }, res["output"]["spans"])
193 def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
194 response = test_utils.check_netconf_node_request(
196 "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
197 self.assertEqual(response.status_code, requests.codes.ok)
198 res = response.json()
199 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
200 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
202 def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
203 response = test_utils.check_netconf_node_request(
205 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
206 self.assertEqual(response.status_code, requests.codes.ok)
207 res = response.json()
208 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
209 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
211 def test_17_servicePath_create_AToZ(self):
212 response = test_utils.service_path_request("create", "test", "1",
213 [{"node-id": "XPDRA01",
214 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
215 {"node-id": "ROADMA01",
216 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
217 {"node-id": "ROADMC01",
218 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
219 {"node-id": "XPDRC01",
220 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
221 self.assertEqual(response.status_code, requests.codes.ok)
222 res = response.json()
223 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
227 def test_18_servicePath_create_ZToA(self):
228 response = test_utils.service_path_request("create", "test", "1",
229 [{"node-id": "XPDRC01",
230 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
231 {"node-id": "ROADMC01",
232 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
233 {"node-id": "ROADMA01",
234 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
235 {"node-id": "XPDRA01",
236 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
237 self.assertEqual(response.status_code, requests.codes.ok)
238 res = response.json()
239 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
243 def test_19_service_power_setup_XPDRA_XPDRC(self):
244 url = "{}/operations/transportpce-olm:service-power-setup"
247 "service-name": "test",
251 "dest-tp": "XPDR1-NETWORK1",
252 "src-tp": "XPDR1-CLIENT1",
256 "dest-tp": "DEG1-TTP-TXRX",
257 "src-tp": "SRG1-PP1-TXRX",
258 "node-id": "ROADMA01"
261 "dest-tp": "SRG1-PP1-TXRX",
262 "src-tp": "DEG2-TTP-TXRX",
263 "node-id": "ROADMC01"
266 "dest-tp": "XPDR1-CLIENT1",
267 "src-tp": "XPDR1-NETWORK1",
273 response = test_utils.post_request(url, data)
274 self.assertEqual(response.status_code, requests.codes.ok)
275 res = response.json()
276 self.assertIn('Success', res["output"]["result"])
278 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
279 response = test_utils.check_netconf_node_request(
281 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
282 self.assertEqual(response.status_code, requests.codes.ok)
283 res = response.json()
284 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
285 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
287 def test_21_get_roadmconnection_ROADMA(self):
288 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
289 self.assertEqual(response.status_code, requests.codes.ok)
290 res = response.json()
291 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
292 self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
294 def test_22_get_roadmconnection_ROADMC(self):
295 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
296 self.assertEqual(response.status_code, requests.codes.ok)
297 res = response.json()
298 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
300 def test_23_service_power_setup_XPDRC_XPDRA(self):
301 url = "{}/operations/transportpce-olm:service-power-setup"
304 "service-name": "test",
308 "dest-tp": "XPDR1-NETWORK1",
309 "src-tp": "XPDR1-CLIENT1",
313 "dest-tp": "DEG2-TTP-TXRX",
314 "src-tp": "SRG1-PP1-TXRX",
315 "node-id": "ROADMC01"
318 "src-tp": "DEG1-TTP-TXRX",
319 "dest-tp": "SRG1-PP1-TXRX",
320 "node-id": "ROADMA01"
323 "src-tp": "XPDR1-NETWORK1",
324 "dest-tp": "XPDR1-CLIENT1",
330 response = test_utils.post_request(url, data)
331 self.assertEqual(response.status_code, requests.codes.ok)
332 res = response.json()
333 self.assertIn('Success', res["output"]["result"])
335 def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
336 response = test_utils.check_netconf_node_request(
338 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
339 self.assertEqual(response.status_code, requests.codes.ok)
340 res = response.json()
341 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
342 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
344 def test_25_get_roadmconnection_ROADMC(self):
345 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
346 self.assertEqual(response.status_code, requests.codes.ok)
347 res = response.json()
348 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
349 self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
351 def test_26_service_power_turndown_XPDRA_XPDRC(self):
352 url = "{}/operations/transportpce-olm:service-power-turndown"
355 "service-name": "test",
359 "dest-tp": "XPDR1-NETWORK1",
360 "src-tp": "XPDR1-CLIENT1",
364 "dest-tp": "DEG1-TTP-TXRX",
365 "src-tp": "SRG1-PP1-TXRX",
366 "node-id": "ROADMA01"
369 "dest-tp": "SRG1-PP1-TXRX",
370 "src-tp": "DEG2-TTP-TXRX",
371 "node-id": "ROADMC01"
374 "dest-tp": "XPDR1-CLIENT1",
375 "src-tp": "XPDR1-NETWORK1",
381 response = test_utils.post_request(url, data)
382 self.assertEqual(response.status_code, requests.codes.ok)
383 res = response.json()
384 self.assertIn('Success', res["output"]["result"])
386 def test_27_get_roadmconnection_ROADMA(self):
387 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
388 self.assertEqual(response.status_code, requests.codes.ok)
389 res = response.json()
390 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
391 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
393 def test_28_get_roadmconnection_ROADMC(self):
394 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
395 self.assertEqual(response.status_code, requests.codes.ok)
396 res = response.json()
397 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
399 def test_29_servicePath_delete_AToZ(self):
400 response = test_utils.service_path_request("delete", "test", "1",
401 [{"node-id": "XPDRA01",
402 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
403 {"node-id": "ROADMA01",
404 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
405 {"node-id": "ROADMC01",
406 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
407 {"node-id": "XPDRC01",
408 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
409 self.assertEqual(response.status_code, requests.codes.ok)
410 res = response.json()
411 self.assertIn('Request processed', res["output"]["result"])
414 def test_30_servicePath_delete_ZToA(self):
415 response = test_utils.service_path_request("delete", "test", "1",
416 [{"node-id": "XPDRC01",
417 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
418 {"node-id": "ROADMC01",
419 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
420 {"node-id": "ROADMA01",
421 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
422 {"node-id": "XPDRA01",
423 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
424 self.assertEqual(response.status_code, requests.codes.ok)
425 res = response.json()
426 self.assertIn('Request processed', res["output"]["result"])
429 """to test case where SRG where the xpdr is connected to has no optical range data"""
431 def test_31_connect_xprdA_to_roadmA(self):
432 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
433 "ROADMA01", "1", "SRG1-PP2-TXRX")
434 self.assertEqual(response.status_code, requests.codes.ok)
435 res = response.json()
436 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
438 def test_32_connect_roadmA_to_xpdrA(self):
439 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
440 "ROADMA01", "1", "SRG1-PP2-TXRX")
441 self.assertEqual(response.status_code, requests.codes.ok)
442 res = response.json()
443 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
445 def test_33_servicePath_create_AToZ(self):
446 response = test_utils.service_path_request("create", "test2", "2",
447 [{"node-id": "XPDRA01",
448 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
449 {"node-id": "ROADMA01",
450 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
451 self.assertEqual(response.status_code, requests.codes.ok)
452 res = response.json()
453 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
457 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
458 response = test_utils.check_netconf_node_request(
460 "interface/XPDR1-NETWORK2-2/org-openroadm-optical-channel-interfaces:och")
461 self.assertEqual(response.status_code, requests.codes.ok)
462 res = response.json()
463 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
464 self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
466 def test_35_servicePath_delete_AToZ(self):
467 response = test_utils.service_path_request("delete", "test", "1",
468 [{"node-id": "XPDRA01",
469 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
470 {"node-id": "ROADMA01",
471 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
472 self.assertEqual(response.status_code, requests.codes.ok)
473 res = response.json()
474 self.assertIn('Request processed', res["output"]["result"])
477 def test_36_xpdrA_device_disconnected(self):
478 response = test_utils.unmount_device("XPDRA01")
479 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
481 def test_37_xpdrC_device_disconnected(self):
482 response = test_utils.unmount_device("XPDRC01")
483 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
485 def test_38_calculate_span_loss_current(self):
486 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
487 response = test_utils.post_request(url, None)
488 self.assertEqual(response.status_code, requests.codes.ok)
489 res = response.json()
490 self.assertIn('Success',
491 res["output"]["result"])
494 def test_39_rdmA_device_disconnected(self):
495 response = test_utils.unmount_device("ROADMA01")
496 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
498 def test_40_rdmC_device_disconnected(self):
499 response = test_utils.unmount_device("ROADMC01")
500 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
503 if __name__ == "__main__":
504 unittest.main(verbosity=2)