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 for process in cls.processes:
33 test_utils.shutdown_process(process)
34 print("all processes killed")
37 print("execution of {}".format(self.id().split(".")[-1]))
40 def test_01_xpdrA_device_connected(self):
41 response = test_utils.mount_device("XPDR-A1", 'xpdra')
42 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
44 def test_02_xpdrC_device_connected(self):
45 response = test_utils.mount_device("XPDR-C1", 'xpdrc')
46 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
48 def test_03_rdmA_device_connected(self):
49 response = test_utils.mount_device("ROADM-A1", 'roadma')
50 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
52 def test_04_rdmC_device_connected(self):
53 response = test_utils.mount_device("ROADM-C1", 'roadmc')
54 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
56 def test_05_connect_xprdA_to_roadmA(self):
57 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
58 "ROADM-A1", "1", "SRG1-PP1-TXRX")
59 self.assertEqual(response.status_code, requests.codes.ok)
61 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
63 def test_06_connect_roadmA_to_xpdrA(self):
64 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
65 "ROADM-A1", "1", "SRG1-PP1-TXRX")
66 self.assertEqual(response.status_code, requests.codes.ok)
68 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
70 def test_07_connect_xprdC_to_roadmC(self):
71 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
72 "ROADM-C1", "1", "SRG1-PP1-TXRX")
73 self.assertEqual(response.status_code, requests.codes.ok)
75 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
77 def test_08_connect_roadmC_to_xpdrC(self):
78 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
79 "ROADM-C1", "1", "SRG1-PP1-TXRX")
80 self.assertEqual(response.status_code, requests.codes.ok)
82 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
84 def test_09_create_OTS_ROADMA(self):
85 response = test_utils.create_ots_oms_request("ROADM-A1", "DEG1-TTP-TXRX")
87 self.assertEqual(response.status_code, requests.codes.ok)
89 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADM-A1',
90 res["output"]["result"])
92 def test_10_create_OTS_ROADMC(self):
93 response = test_utils.create_ots_oms_request("ROADM-C1", "DEG2-TTP-TXRX")
94 self.assertEqual(response.status_code, requests.codes.ok)
96 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADM-C1',
97 res["output"]["result"])
99 def test_11_get_PM_ROADMA(self):
100 url = "{}/operations/transportpce-olm:get-pm"
103 "node-id": "ROADM-A1",
104 "resource-type": "interface",
105 "granularity": "15min",
106 "resource-identifier": {
107 "resource-name": "OTS-DEG2-TTP-TXRX"
111 response = test_utils.post_request(url, data)
112 self.assertEqual(response.status_code, requests.codes.ok)
113 res = response.json()
115 "pmparameter-name": "OpticalPowerOutput",
116 "pmparameter-value": "2.5"
117 }, res["output"]["measurements"])
119 "pmparameter-name": "OpticalReturnLoss",
120 "pmparameter-value": "40"
121 }, res["output"]["measurements"])
123 "pmparameter-name": "OpticalPowerInput",
124 "pmparameter-value": "-21.1"
125 }, res["output"]["measurements"])
127 def test_12_get_PM_ROADMC(self):
128 url = "{}/operations/transportpce-olm:get-pm"
131 "node-id": "ROADM-C1",
132 "resource-type": "interface",
133 "granularity": "15min",
134 "resource-identifier": {
135 "resource-name": "OTS-DEG1-TTP-TXRX"
139 response = test_utils.post_request(url, data)
140 self.assertEqual(response.status_code, requests.codes.ok)
141 res = response.json()
143 "pmparameter-name": "OpticalPowerOutput",
144 "pmparameter-value": "4.6"
145 }, res["output"]["measurements"])
147 "pmparameter-name": "OpticalReturnLoss",
148 "pmparameter-value": "49.1"
149 }, res["output"]["measurements"])
151 "pmparameter-name": "OpticalPowerInput",
152 "pmparameter-value": "-15.1"
153 }, res["output"]["measurements"])
155 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
156 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
160 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
163 response = test_utils.post_request(url, data)
164 self.assertEqual(response.status_code, requests.codes.ok)
165 res = response.json()
166 self.assertIn('Success',
167 res["output"]["result"])
170 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
171 }, res["output"]["spans"])
174 def test_14_calculate_span_loss_base_all(self):
175 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
181 response = test_utils.post_request(url, data)
182 self.assertEqual(response.status_code, requests.codes.ok)
183 res = response.json()
184 self.assertIn('Success',
185 res["output"]["result"])
188 "link-id": "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX"
189 }, res["output"]["spans"])
192 "link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
193 }, res["output"]["spans"])
196 def test_15_get_OTS_DEG2_TTP_TXRX_ROADMA(self):
197 response = test_utils.check_netconf_node_request(
199 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
200 self.assertEqual(response.status_code, requests.codes.ok)
201 res = response.json()
202 self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
203 self.assertEqual(25.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
205 def test_16_get_OTS_DEG1_TTP_TXRX_ROADMC(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(25.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
212 self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
214 def test_17_servicePath_create_AToZ(self):
215 response = test_utils.service_path_request("create", "test", "1",
216 [{"node-id": "XPDR-A1",
217 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
218 {"node-id": "ROADM-A1",
219 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
220 {"node-id": "ROADM-C1",
221 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG1-TTP-TXRX"},
222 {"node-id": "XPDR-C1",
223 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
224 self.assertEqual(response.status_code, requests.codes.ok)
225 res = response.json()
226 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
230 def test_18_servicePath_create_ZToA(self):
231 response = test_utils.service_path_request("create", "test", "1",
232 [{"node-id": "XPDR-C1",
233 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
234 {"node-id": "ROADM-C1",
235 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
236 {"node-id": "ROADM-A1",
237 "src-tp": "DEG2-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
238 {"node-id": "XPDR-A1",
239 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
240 self.assertEqual(response.status_code, requests.codes.ok)
241 res = response.json()
242 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
246 def test_19_service_power_setup_XPDRA_XPDRC(self):
247 url = "{}/operations/transportpce-olm:service-power-setup"
250 "service-name": "test",
254 "dest-tp": "XPDR1-NETWORK1",
255 "src-tp": "XPDR1-CLIENT1",
259 "dest-tp": "DEG2-TTP-TXRX",
260 "src-tp": "SRG1-PP1-TXRX",
261 "node-id": "ROADM-A1"
264 "dest-tp": "SRG1-PP1-TXRX",
265 "src-tp": "DEG1-TTP-TXRX",
266 "node-id": "ROADM-C1"
269 "dest-tp": "XPDR1-CLIENT1",
270 "src-tp": "XPDR1-NETWORK1",
276 response = test_utils.post_request(url, data)
277 self.assertEqual(response.status_code, requests.codes.ok)
278 res = response.json()
279 self.assertIn('Success', res["output"]["result"])
281 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
282 response = test_utils.check_netconf_node_request(
284 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
285 self.assertEqual(response.status_code, requests.codes.ok)
286 res = response.json()
287 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
288 self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
290 def test_21_get_roadmconnection_ROADMA(self):
291 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
292 self.assertEqual(response.status_code, requests.codes.ok)
293 res = response.json()
294 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
295 self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power'])
297 def test_22_get_roadmconnection_ROADMC(self):
298 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-1")
299 self.assertEqual(response.status_code, requests.codes.ok)
300 res = response.json()
301 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
303 def test_23_service_power_setup_XPDRC_XPDRA(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": "ROADM-C1"
321 "src-tp": "DEG2-TTP-TXRX",
322 "dest-tp": "SRG1-PP1-TXRX",
323 "node-id": "ROADM-A1"
326 "src-tp": "XPDR1-NETWORK1",
327 "dest-tp": "XPDR1-CLIENT1",
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_24_get_interface_XPDRC_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(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
345 self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
347 def test_25_get_roadmconnection_ROADMC(self):
348 response = test_utils.check_netconf_node_request("ROADM-C1", "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(2.0, res['roadm-connections'][0]['target-output-power'])
354 def test_26_service_power_turndown_XPDRA_XPDRC(self):
355 url = "{}/operations/transportpce-olm:service-power-turndown"
358 "service-name": "test",
362 "dest-tp": "XPDR1-NETWORK1",
363 "src-tp": "XPDR1-CLIENT1",
367 "dest-tp": "DEG2-TTP-TXRX",
368 "src-tp": "SRG1-PP1-TXRX",
369 "node-id": "ROADM-A1"
372 "dest-tp": "SRG1-PP1-TXRX",
373 "src-tp": "DEG1-TTP-TXRX",
374 "node-id": "ROADM-C1"
377 "dest-tp": "XPDR1-CLIENT1",
378 "src-tp": "XPDR1-NETWORK1",
384 response = test_utils.post_request(url, data)
385 self.assertEqual(response.status_code, requests.codes.ok)
386 res = response.json()
387 self.assertIn('Success', res["output"]["result"])
389 def test_27_get_roadmconnection_ROADMA(self):
390 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
391 self.assertEqual(response.status_code, requests.codes.ok)
392 res = response.json()
393 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
394 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
396 def test_28_get_roadmconnection_ROADMC(self):
397 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-1")
398 self.assertEqual(response.status_code, requests.codes.ok)
399 res = response.json()
400 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
402 def test_29_servicePath_delete_AToZ(self):
403 response = test_utils.service_path_request("delete", "test", "1",
404 [{"node-id": "XPDR-A1",
405 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
406 {"node-id": "ROADM-A1",
407 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
408 {"node-id": "ROADM-C1",
409 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG1-TTP-TXRX"},
410 {"node-id": "XPDR-C1",
411 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
412 self.assertEqual(response.status_code, requests.codes.ok)
413 res = response.json()
414 self.assertIn('Request processed', res["output"]["result"])
417 def test_30_servicePath_delete_ZToA(self):
418 response = test_utils.service_path_request("delete", "test", "1",
419 [{"node-id": "XPDR-C1",
420 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
421 {"node-id": "ROADM-C1",
422 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
423 {"node-id": "ROADM-A1",
424 "src-tp": "DEG2-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
425 {"node-id": "XPDR-A1",
426 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
427 self.assertEqual(response.status_code, requests.codes.ok)
428 res = response.json()
429 self.assertIn('Request processed', res["output"]["result"])
432 """to test case where SRG where the xpdr is connected to has no optical range data"""
434 def test_31_connect_xprdA_to_roadmA(self):
435 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
436 "ROADM-A1", "1", "SRG1-PP2-TXRX")
437 self.assertEqual(response.status_code, requests.codes.ok)
438 res = response.json()
439 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
441 def test_32_connect_roadmA_to_xpdrA(self):
442 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
443 "ROADM-A1", "1", "SRG1-PP2-TXRX")
444 self.assertEqual(response.status_code, requests.codes.ok)
445 res = response.json()
446 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
448 def test_33_servicePath_create_AToZ(self):
449 response = test_utils.service_path_request("create", "test2", "2",
450 [{"node-id": "XPDR-A1",
451 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
452 {"node-id": "ROADM-A1",
453 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
454 self.assertEqual(response.status_code, requests.codes.ok)
455 res = response.json()
456 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
460 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
461 response = test_utils.check_netconf_node_request(
463 "interface/XPDR1-NETWORK2-2/org-openroadm-optical-channel-interfaces:och")
464 self.assertEqual(response.status_code, requests.codes.ok)
465 res = response.json()
466 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
467 # self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
469 def test_35_servicePath_delete_AToZ(self):
470 response = test_utils.service_path_request("delete", "test", "1",
471 [{"node-id": "XPDR-A1",
472 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
473 {"node-id": "ROADM-A1",
474 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
475 self.assertEqual(response.status_code, requests.codes.ok)
476 res = response.json()
477 self.assertIn('Request processed', res["output"]["result"])
480 def test_36_xpdrA_device_disconnected(self):
481 response = test_utils.unmount_device("XPDR-A1")
482 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
484 def test_37_xpdrC_device_disconnected(self):
485 response = test_utils.unmount_device("XPDR-C1")
486 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
488 def test_38_calculate_span_loss_current(self):
489 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
490 response = test_utils.post_request(url, None)
491 self.assertEqual(response.status_code, requests.codes.ok)
492 res = response.json()
493 self.assertIn('Success',
494 res["output"]["result"])
497 def test_39_rdmA_device_disconnected(self):
498 response = test_utils.unmount_device("ROADM-A1")
499 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
501 def test_40_rdmC_device_disconnected(self):
502 response = test_utils.unmount_device("ROADM-C1")
503 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
506 if __name__ == "__main__":
507 unittest.main(verbosity=2)