fix few bugs and unused issues in functional tests
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_olm.py
1 #!/usr/bin/env python
2
3 #############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
5 #
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 #############################################################################
11
12 # pylint: disable=no-member
13 # pylint: disable=too-many-public-methods
14
15 import unittest
16 import time
17 import requests
18 from common import test_utils
19
20
21 class TransportOlmTesting(unittest.TestCase):
22
23     processes = None
24
25     @classmethod
26     def setUpClass(cls):
27         cls.processes = test_utils.start_tpce()
28         cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc'])
29
30     @classmethod
31     def tearDownClass(cls):
32         for process in cls.processes:
33             test_utils.shutdown_process(process)
34         print("all processes killed")
35
36     def setUp(self):
37         print("execution of {}".format(self.id().split(".")[-1]))
38         time.sleep(1)
39
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)
43
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)
47
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)
51
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)
55
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)
60         res = response.json()
61         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
62
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)
67         res = response.json()
68         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
69
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)
74         res = response.json()
75         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
76
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)
81         res = response.json()
82         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
83
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)
87         res = response.json()
88         self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
89                       res["output"]["result"])
90
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)
94         res = response.json()
95         self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
96                       res["output"]["result"])
97
98     def test_11_get_PM_ROADMA(self):
99         url = "{}/operations/transportpce-olm:get-pm"
100         data = {
101             "input": {
102                 "node-id": "ROADMA01",
103                 "resource-type": "interface",
104                 "granularity": "15min",
105                 "resource-identifier": {
106                     "resource-name": "OTS-DEG1-TTP-TXRX"
107                 }
108             }
109         }
110         response = test_utils.post_request(url, data)
111         self.assertEqual(response.status_code, requests.codes.ok)
112         res = response.json()
113         self.assertIn({
114             "pmparameter-name": "OpticalPowerOutput",
115             "pmparameter-value": "2.5"
116         }, res["output"]["measurements"])
117         self.assertIn({
118             "pmparameter-name": "OpticalReturnLoss",
119             "pmparameter-value": "49.9"
120         }, res["output"]["measurements"])
121         self.assertIn({
122             "pmparameter-name": "OpticalPowerInput",
123             "pmparameter-value": "3"
124         }, res["output"]["measurements"])
125
126     def test_12_get_PM_ROADMC(self):
127         url = "{}/operations/transportpce-olm:get-pm"
128         data = {
129             "input": {
130                 "node-id": "ROADMC01",
131                 "resource-type": "interface",
132                 "granularity": "15min",
133                 "resource-identifier": {
134                     "resource-name": "OTS-DEG2-TTP-TXRX"
135                 }
136             }
137         }
138         response = test_utils.post_request(url, data)
139         self.assertEqual(response.status_code, requests.codes.ok)
140         res = response.json()
141         self.assertIn({
142             "pmparameter-name": "OpticalPowerOutput",
143             "pmparameter-value": "18.1"
144         }, res["output"]["measurements"])
145         self.assertIn({
146             "pmparameter-name": "OpticalReturnLoss",
147             "pmparameter-value": "48.8"
148         }, res["output"]["measurements"])
149         self.assertIn({
150             "pmparameter-name": "OpticalPowerInput",
151             "pmparameter-value": "-3.2"
152         }, res["output"]["measurements"])
153
154     def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
155         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
156         data = {
157             "input": {
158                 "src-type": "link",
159                 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
160             }
161         }
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"])
167         self.assertIn({
168             "spanloss": "5.7",
169             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
170         }, res["output"]["spans"])
171         time.sleep(5)
172
173     def test_14_calculate_span_loss_base_all(self):
174         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
175         data = {
176             "input": {
177                 "src-type": "all"
178             }
179         }
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"])
185         self.assertIn({
186             "spanloss": "15.1",
187             "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
188         }, res["output"]["spans"])
189         self.assertIn({
190             "spanloss": "5.7",
191             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
192         }, res["output"]["spans"])
193         time.sleep(5)
194
195     def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
196         response = test_utils.check_netconf_node_request(
197             "ROADMA01",
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'])
203
204     def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
205         response = test_utils.check_netconf_node_request(
206             "ROADMC01",
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'])
212
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"])
226         # time.sleep(40)
227         time.sleep(10)
228
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"])
242         # time.sleep(40)
243         time.sleep(10)
244
245     def test_19_service_power_setup_XPDRA_XPDRC(self):
246         url = "{}/operations/transportpce-olm:service-power-setup"
247         data = {
248             "input": {
249                 "service-name": "test",
250                 "wave-number": 1,
251                 "nodes": [
252                     {
253                         "dest-tp": "XPDR1-NETWORK1",
254                         "src-tp": "XPDR1-CLIENT1",
255                         "node-id": "XPDRA01"
256                     },
257                     {
258                         "dest-tp": "DEG1-TTP-TXRX",
259                         "src-tp": "SRG1-PP1-TXRX",
260                         "node-id": "ROADMA01"
261                     },
262                     {
263                         "dest-tp": "SRG1-PP1-TXRX",
264                         "src-tp": "DEG2-TTP-TXRX",
265                         "node-id": "ROADMC01"
266                     },
267                     {
268                         "dest-tp": "XPDR1-CLIENT1",
269                         "src-tp": "XPDR1-NETWORK1",
270                         "node-id": "XPDRC01"
271                     }
272                 ]
273             }
274         }
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"])
279
280     def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
281         response = test_utils.check_netconf_node_request(
282             "XPDRA01",
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'])
288
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'])
295
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'])
301
302     def test_23_service_power_setup_XPDRC_XPDRA(self):
303         url = "{}/operations/transportpce-olm:service-power-setup"
304         data = {
305             "input": {
306                 "service-name": "test",
307                 "wave-number": 1,
308                 "nodes": [
309                     {
310                         "dest-tp": "XPDR1-NETWORK1",
311                         "src-tp": "XPDR1-CLIENT1",
312                         "node-id": "XPDRC01"
313                     },
314                     {
315                         "dest-tp": "DEG2-TTP-TXRX",
316                         "src-tp": "SRG1-PP1-TXRX",
317                         "node-id": "ROADMC01"
318                     },
319                     {
320                         "src-tp": "DEG1-TTP-TXRX",
321                         "dest-tp": "SRG1-PP1-TXRX",
322                         "node-id": "ROADMA01"
323                     },
324                     {
325                         "src-tp": "XPDR1-NETWORK1",
326                         "dest-tp": "XPDR1-CLIENT1",
327                         "node-id": "XPDRA01"
328                     }
329                 ]
330             }
331         }
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"])
336
337     def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
338         response = test_utils.check_netconf_node_request(
339             "XPDRC01",
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'])
345
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'])
352
353     def test_26_service_power_turndown_XPDRA_XPDRC(self):
354         url = "{}/operations/transportpce-olm:service-power-turndown"
355         data = {
356             "input": {
357                 "service-name": "test",
358                 "wave-number": 1,
359                 "nodes": [
360                     {
361                         "dest-tp": "XPDR1-NETWORK1",
362                         "src-tp": "XPDR1-CLIENT1",
363                         "node-id": "XPDRA01"
364                     },
365                     {
366                         "dest-tp": "DEG1-TTP-TXRX",
367                         "src-tp": "SRG1-PP1-TXRX",
368                         "node-id": "ROADMA01"
369                     },
370                     {
371                         "dest-tp": "SRG1-PP1-TXRX",
372                         "src-tp": "DEG2-TTP-TXRX",
373                         "node-id": "ROADMC01"
374                     },
375                     {
376                         "dest-tp": "XPDR1-CLIENT1",
377                         "src-tp": "XPDR1-NETWORK1",
378                         "node-id": "XPDRC01"
379                     }
380                 ]
381             }
382         }
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"])
387
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'])
394
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'])
400
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"])
414         time.sleep(10)
415
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"])
429         time.sleep(10)
430
431     """to test case where SRG where the xpdr is connected to has no optical range data"""
432
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"])
439
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"])
446
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"])
456         # time.sleep(40)
457         time.sleep(10)
458
459     def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
460         response = test_utils.check_netconf_node_request(
461             "XPDRA01",
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'])
467
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"])
477         time.sleep(10)
478
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)
482
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)
486
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"])
494         time.sleep(5)
495
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)
499
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)
503
504
505 if __name__ == "__main__":
506     unittest.main(verbosity=2)