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-full', 'roadmc-full', '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("XPDRA01", '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("XPDRC01", '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("ROADMA01", 'roadma-full')
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("ROADMC01", 'roadmc-full')
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("XPDRA01", "1", "1",
58 "ROADMA01", "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("XPDRA01", "1", "1",
65 "ROADMA01", "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("XPDRC01", "1", "1",
72 "ROADMC01", "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("XPDRC01", "1", "1",
79 "ROADMC01", "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("ROADMA01", "DEG1-TTP-TXRX")
86 self.assertEqual(response.status_code, requests.codes.ok)
88 self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
89 res["output"]["result"])
91 def test_10_create_OTS_ROADMC(self):
92 response = test_utils.create_ots_oms_request("ROADMC01", "DEG2-TTP-TXRX")
93 self.assertEqual(response.status_code, requests.codes.ok)
95 self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
96 res["output"]["result"])
98 def test_11_get_PM_ROADMA(self):
99 url = "{}/operations/transportpce-olm:get-pm"
102 "node-id": "ROADMA01",
103 "resource-type": "interface",
104 "granularity": "15min",
105 "resource-identifier": {
106 "resource-name": "OTS-DEG1-TTP-TXRX"
110 response = test_utils.post_request(url, data)
111 self.assertEqual(response.status_code, requests.codes.ok)
112 res = response.json()
114 "pmparameter-name": "OpticalPowerOutput",
115 "pmparameter-value": "2.5"
116 }, res["output"]["measurements"])
118 "pmparameter-name": "OpticalReturnLoss",
119 "pmparameter-value": "49.9"
120 }, res["output"]["measurements"])
122 "pmparameter-name": "OpticalPowerInput",
123 "pmparameter-value": "3"
124 }, res["output"]["measurements"])
126 def test_12_get_PM_ROADMC(self):
127 url = "{}/operations/transportpce-olm:get-pm"
130 "node-id": "ROADMC01",
131 "resource-type": "interface",
132 "granularity": "15min",
133 "resource-identifier": {
134 "resource-name": "OTS-DEG2-TTP-TXRX"
138 response = test_utils.post_request(url, data)
139 self.assertEqual(response.status_code, requests.codes.ok)
140 res = response.json()
142 "pmparameter-name": "OpticalPowerOutput",
143 "pmparameter-value": "18.1"
144 }, res["output"]["measurements"])
146 "pmparameter-name": "OpticalReturnLoss",
147 "pmparameter-value": "48.8"
148 }, res["output"]["measurements"])
150 "pmparameter-name": "OpticalPowerInput",
151 "pmparameter-value": "-3.2"
152 }, res["output"]["measurements"])
154 def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
155 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
159 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
162 response = test_utils.post_request(url, data)
163 self.assertEqual(response.status_code, requests.codes.ok)
164 res = response.json()
165 self.assertIn('Success',
166 res["output"]["result"])
169 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
170 }, res["output"]["spans"])
173 def test_14_calculate_span_loss_base_all(self):
174 url = "{}/operations/transportpce-olm:calculate-spanloss-base"
180 response = test_utils.post_request(url, data)
181 self.assertEqual(response.status_code, requests.codes.ok)
182 res = response.json()
183 self.assertIn('Success',
184 res["output"]["result"])
187 "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
188 }, res["output"]["spans"])
191 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
192 }, res["output"]["spans"])
195 def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
196 response = test_utils.check_netconf_node_request(
198 "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
199 self.assertEqual(response.status_code, requests.codes.ok)
200 res = response.json()
201 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
202 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
204 def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
205 response = test_utils.check_netconf_node_request(
207 "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
208 self.assertEqual(response.status_code, requests.codes.ok)
209 res = response.json()
210 self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
211 self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
213 def test_17_servicePath_create_AToZ(self):
214 response = test_utils.service_path_request("create", "test", "1",
215 [{"node-id": "XPDRA01",
216 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
217 {"node-id": "ROADMA01",
218 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
219 {"node-id": "ROADMC01",
220 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
221 {"node-id": "XPDRC01",
222 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
223 self.assertEqual(response.status_code, requests.codes.ok)
224 res = response.json()
225 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
229 def test_18_servicePath_create_ZToA(self):
230 response = test_utils.service_path_request("create", "test", "1",
231 [{"node-id": "XPDRC01",
232 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
233 {"node-id": "ROADMC01",
234 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
235 {"node-id": "ROADMA01",
236 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
237 {"node-id": "XPDRA01",
238 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
239 self.assertEqual(response.status_code, requests.codes.ok)
240 res = response.json()
241 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
245 def test_19_service_power_setup_XPDRA_XPDRC(self):
246 url = "{}/operations/transportpce-olm:service-power-setup"
249 "service-name": "test",
253 "dest-tp": "XPDR1-NETWORK1",
254 "src-tp": "XPDR1-CLIENT1",
258 "dest-tp": "DEG1-TTP-TXRX",
259 "src-tp": "SRG1-PP1-TXRX",
260 "node-id": "ROADMA01"
263 "dest-tp": "SRG1-PP1-TXRX",
264 "src-tp": "DEG2-TTP-TXRX",
265 "node-id": "ROADMC01"
268 "dest-tp": "XPDR1-CLIENT1",
269 "src-tp": "XPDR1-NETWORK1",
275 response = test_utils.post_request(url, data)
276 self.assertEqual(response.status_code, requests.codes.ok)
277 res = response.json()
278 self.assertIn('Success', res["output"]["result"])
280 def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
281 response = test_utils.check_netconf_node_request(
283 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
284 self.assertEqual(response.status_code, requests.codes.ok)
285 res = response.json()
286 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
287 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
289 def test_21_get_roadmconnection_ROADMA(self):
290 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
291 self.assertEqual(response.status_code, requests.codes.ok)
292 res = response.json()
293 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
294 self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
296 def test_22_get_roadmconnection_ROADMC(self):
297 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
298 self.assertEqual(response.status_code, requests.codes.ok)
299 res = response.json()
300 self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
302 def test_23_service_power_setup_XPDRC_XPDRA(self):
303 url = "{}/operations/transportpce-olm:service-power-setup"
306 "service-name": "test",
310 "dest-tp": "XPDR1-NETWORK1",
311 "src-tp": "XPDR1-CLIENT1",
315 "dest-tp": "DEG2-TTP-TXRX",
316 "src-tp": "SRG1-PP1-TXRX",
317 "node-id": "ROADMC01"
320 "src-tp": "DEG1-TTP-TXRX",
321 "dest-tp": "SRG1-PP1-TXRX",
322 "node-id": "ROADMA01"
325 "src-tp": "XPDR1-NETWORK1",
326 "dest-tp": "XPDR1-CLIENT1",
332 response = test_utils.post_request(url, data)
333 self.assertEqual(response.status_code, requests.codes.ok)
334 res = response.json()
335 self.assertIn('Success', res["output"]["result"])
337 def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
338 response = test_utils.check_netconf_node_request(
340 "interface/XPDR1-NETWORK1-1/org-openroadm-optical-channel-interfaces:och")
341 self.assertEqual(response.status_code, requests.codes.ok)
342 res = response.json()
343 self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
344 self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
346 def test_25_get_roadmconnection_ROADMC(self):
347 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
348 self.assertEqual(response.status_code, requests.codes.ok)
349 res = response.json()
350 self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
351 self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
353 def test_26_service_power_turndown_XPDRA_XPDRC(self):
354 url = "{}/operations/transportpce-olm:service-power-turndown"
357 "service-name": "test",
361 "dest-tp": "XPDR1-NETWORK1",
362 "src-tp": "XPDR1-CLIENT1",
366 "dest-tp": "DEG1-TTP-TXRX",
367 "src-tp": "SRG1-PP1-TXRX",
368 "node-id": "ROADMA01"
371 "dest-tp": "SRG1-PP1-TXRX",
372 "src-tp": "DEG2-TTP-TXRX",
373 "node-id": "ROADMC01"
376 "dest-tp": "XPDR1-CLIENT1",
377 "src-tp": "XPDR1-NETWORK1",
383 response = test_utils.post_request(url, data)
384 self.assertEqual(response.status_code, requests.codes.ok)
385 res = response.json()
386 self.assertIn('Success', res["output"]["result"])
388 def test_27_get_roadmconnection_ROADMA(self):
389 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
390 self.assertEqual(response.status_code, requests.codes.ok)
391 res = response.json()
392 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
393 self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
395 def test_28_get_roadmconnection_ROADMC(self):
396 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
397 self.assertEqual(response.status_code, requests.codes.ok)
398 res = response.json()
399 self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
401 def test_29_servicePath_delete_AToZ(self):
402 response = test_utils.service_path_request("delete", "test", "1",
403 [{"node-id": "XPDRA01",
404 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
405 {"node-id": "ROADMA01",
406 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
407 {"node-id": "ROADMC01",
408 "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
409 {"node-id": "XPDRC01",
410 "dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}])
411 self.assertEqual(response.status_code, requests.codes.ok)
412 res = response.json()
413 self.assertIn('Request processed', res["output"]["result"])
416 def test_30_servicePath_delete_ZToA(self):
417 response = test_utils.service_path_request("delete", "test", "1",
418 [{"node-id": "XPDRC01",
419 "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
420 {"node-id": "ROADMC01",
421 "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
422 {"node-id": "ROADMA01",
423 "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
424 {"node-id": "XPDRA01",
425 "src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}])
426 self.assertEqual(response.status_code, requests.codes.ok)
427 res = response.json()
428 self.assertIn('Request processed', res["output"]["result"])
431 """to test case where SRG where the xpdr is connected to has no optical range data"""
433 def test_31_connect_xprdA_to_roadmA(self):
434 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
435 "ROADMA01", "1", "SRG1-PP2-TXRX")
436 self.assertEqual(response.status_code, requests.codes.ok)
437 res = response.json()
438 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
440 def test_32_connect_roadmA_to_xpdrA(self):
441 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
442 "ROADMA01", "1", "SRG1-PP2-TXRX")
443 self.assertEqual(response.status_code, requests.codes.ok)
444 res = response.json()
445 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
447 def test_33_servicePath_create_AToZ(self):
448 response = test_utils.service_path_request("create", "test2", "2",
449 [{"node-id": "XPDRA01",
450 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
451 {"node-id": "ROADMA01",
452 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
453 self.assertEqual(response.status_code, requests.codes.ok)
454 res = response.json()
455 self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
459 def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
460 response = test_utils.check_netconf_node_request(
462 "interface/XPDR1-NETWORK2-2/org-openroadm-optical-channel-interfaces:och")
463 self.assertEqual(response.status_code, requests.codes.ok)
464 res = response.json()
465 self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
466 self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
468 def test_35_servicePath_delete_AToZ(self):
469 response = test_utils.service_path_request("delete", "test", "1",
470 [{"node-id": "XPDRA01",
471 "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
472 {"node-id": "ROADMA01",
473 "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}])
474 self.assertEqual(response.status_code, requests.codes.ok)
475 res = response.json()
476 self.assertIn('Request processed', res["output"]["result"])
479 def test_36_xpdrA_device_disconnected(self):
480 response = test_utils.unmount_device("XPDRA01")
481 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
483 def test_37_xpdrC_device_disconnected(self):
484 response = test_utils.unmount_device("XPDRC01")
485 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
487 def test_38_calculate_span_loss_current(self):
488 url = "{}/operations/transportpce-olm:calculate-spanloss-current"
489 response = test_utils.post_request(url, None)
490 self.assertEqual(response.status_code, requests.codes.ok)
491 res = response.json()
492 self.assertIn('Success',
493 res["output"]["result"])
496 def test_39_rdmA_device_disconnected(self):
497 response = test_utils.unmount_device("ROADMA01")
498 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
500 def test_40_rdmC_device_disconnected(self):
501 response = test_utils.unmount_device("ROADMC01")
502 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
505 if __name__ == "__main__":
506 unittest.main(verbosity=2)