fix some pylint issues
[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         # pylint: disable=not-an-iterable
33         for process in cls.processes:
34             test_utils.shutdown_process(process)
35         print("all processes killed")
36
37     def setUp(self):
38         print("execution of {}".format(self.id().split(".")[-1]))
39         time.sleep(1)
40
41     def test_01_xpdrA_device_connected(self):
42         response = test_utils.mount_device("XPDRA01", 'xpdra')
43         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
44
45     def test_02_xpdrC_device_connected(self):
46         response = test_utils.mount_device("XPDRC01", 'xpdrc')
47         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
48
49     def test_03_rdmA_device_connected(self):
50         response = test_utils.mount_device("ROADMA01", 'roadma-full')
51         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
52
53     def test_04_rdmC_device_connected(self):
54         response = test_utils.mount_device("ROADMC01", 'roadmc-full')
55         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
56
57     def test_05_connect_xprdA_to_roadmA(self):
58         response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
59                                                           "ROADMA01", "1", "SRG1-PP1-TXRX")
60         self.assertEqual(response.status_code, requests.codes.ok)
61         res = response.json()
62         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
63
64     def test_06_connect_roadmA_to_xpdrA(self):
65         response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
66                                                           "ROADMA01", "1", "SRG1-PP1-TXRX")
67         self.assertEqual(response.status_code, requests.codes.ok)
68         res = response.json()
69         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
70
71     def test_07_connect_xprdC_to_roadmC(self):
72         response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
73                                                           "ROADMC01", "1", "SRG1-PP1-TXRX")
74         self.assertEqual(response.status_code, requests.codes.ok)
75         res = response.json()
76         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
77
78     def test_08_connect_roadmC_to_xpdrC(self):
79         response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
80                                                           "ROADMC01", "1", "SRG1-PP1-TXRX")
81         self.assertEqual(response.status_code, requests.codes.ok)
82         res = response.json()
83         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
84
85     def test_09_create_OTS_ROADMA(self):
86         response = test_utils.create_ots_oms_request("ROADMA01", "DEG1-TTP-TXRX")
87         self.assertEqual(response.status_code, requests.codes.ok)
88         res = response.json()
89         self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
90                       res["output"]["result"])
91
92     def test_10_create_OTS_ROADMC(self):
93         response = test_utils.create_ots_oms_request("ROADMC01", "DEG2-TTP-TXRX")
94         self.assertEqual(response.status_code, requests.codes.ok)
95         res = response.json()
96         self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
97                       res["output"]["result"])
98
99     def test_11_get_PM_ROADMA(self):
100         url = "{}/operations/transportpce-olm:get-pm"
101         data = {
102             "input": {
103                 "node-id": "ROADMA01",
104                 "resource-type": "interface",
105                 "granularity": "15min",
106                 "resource-identifier": {
107                     "resource-name": "OTS-DEG1-TTP-TXRX"
108                 }
109             }
110         }
111         response = test_utils.post_request(url, data)
112         self.assertEqual(response.status_code, requests.codes.ok)
113         res = response.json()
114         self.assertIn({
115             "pmparameter-name": "OpticalPowerOutput",
116             "pmparameter-value": "2.5"
117         }, res["output"]["measurements"])
118         self.assertIn({
119             "pmparameter-name": "OpticalReturnLoss",
120             "pmparameter-value": "49.9"
121         }, res["output"]["measurements"])
122         self.assertIn({
123             "pmparameter-name": "OpticalPowerInput",
124             "pmparameter-value": "3"
125         }, res["output"]["measurements"])
126
127     def test_12_get_PM_ROADMC(self):
128         url = "{}/operations/transportpce-olm:get-pm"
129         data = {
130             "input": {
131                 "node-id": "ROADMC01",
132                 "resource-type": "interface",
133                 "granularity": "15min",
134                 "resource-identifier": {
135                     "resource-name": "OTS-DEG2-TTP-TXRX"
136                 }
137             }
138         }
139         response = test_utils.post_request(url, data)
140         self.assertEqual(response.status_code, requests.codes.ok)
141         res = response.json()
142         self.assertIn({
143             "pmparameter-name": "OpticalPowerOutput",
144             "pmparameter-value": "18.1"
145         }, res["output"]["measurements"])
146         self.assertIn({
147             "pmparameter-name": "OpticalReturnLoss",
148             "pmparameter-value": "48.8"
149         }, res["output"]["measurements"])
150         self.assertIn({
151             "pmparameter-name": "OpticalPowerInput",
152             "pmparameter-value": "-3.2"
153         }, res["output"]["measurements"])
154
155     def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
156         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
157         data = {
158             "input": {
159                 "src-type": "link",
160                 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
161             }
162         }
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"])
168         self.assertIn({
169             "spanloss": "5.7",
170             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
171         }, res["output"]["spans"])
172         time.sleep(5)
173
174     def test_14_calculate_span_loss_base_all(self):
175         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
176         data = {
177             "input": {
178                 "src-type": "all"
179             }
180         }
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"])
186         self.assertIn({
187             "spanloss": "15.1",
188             "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
189         }, res["output"]["spans"])
190         self.assertIn({
191             "spanloss": "5.7",
192             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
193         }, res["output"]["spans"])
194         time.sleep(5)
195
196     def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
197         response = test_utils.check_netconf_node_request(
198             "ROADMA01",
199             "interface/OTS-DEG1-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
200         self.assertEqual(response.status_code, requests.codes.ok)
201         res = response.json()
202         self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
203         self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
204
205     def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
206         response = test_utils.check_netconf_node_request(
207             "ROADMC01",
208             "interface/OTS-DEG2-TTP-TXRX/org-openroadm-optical-transport-interfaces:ots")
209         self.assertEqual(response.status_code, requests.codes.ok)
210         res = response.json()
211         self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
212         self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
213
214     def test_17_servicePath_create_AToZ(self):
215         response = test_utils.service_path_request("create", "test", "1",
216                                                    [{"node-id": "XPDRA01",
217                                                      "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
218                                                     {"node-id": "ROADMA01",
219                                                      "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
220                                                     {"node-id": "ROADMC01",
221                                                      "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
222                                                     {"node-id": "XPDRC01",
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"])
227         # time.sleep(40)
228         time.sleep(10)
229
230     def test_18_servicePath_create_ZToA(self):
231         response = test_utils.service_path_request("create", "test", "1",
232                                                    [{"node-id": "XPDRC01",
233                                                      "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
234                                                     {"node-id": "ROADMC01",
235                                                      "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
236                                                     {"node-id": "ROADMA01",
237                                                      "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
238                                                     {"node-id": "XPDRA01",
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"])
243         # time.sleep(40)
244         time.sleep(10)
245
246     def test_19_service_power_setup_XPDRA_XPDRC(self):
247         url = "{}/operations/transportpce-olm:service-power-setup"
248         data = {
249             "input": {
250                 "service-name": "test",
251                 "wave-number": 1,
252                 "nodes": [
253                     {
254                         "dest-tp": "XPDR1-NETWORK1",
255                         "src-tp": "XPDR1-CLIENT1",
256                         "node-id": "XPDRA01"
257                     },
258                     {
259                         "dest-tp": "DEG1-TTP-TXRX",
260                         "src-tp": "SRG1-PP1-TXRX",
261                         "node-id": "ROADMA01"
262                     },
263                     {
264                         "dest-tp": "SRG1-PP1-TXRX",
265                         "src-tp": "DEG2-TTP-TXRX",
266                         "node-id": "ROADMC01"
267                     },
268                     {
269                         "dest-tp": "XPDR1-CLIENT1",
270                         "src-tp": "XPDR1-NETWORK1",
271                         "node-id": "XPDRC01"
272                     }
273                 ]
274             }
275         }
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"])
280
281     def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
282         response = test_utils.check_netconf_node_request(
283             "XPDRA01",
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(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
288         self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
289
290     def test_21_get_roadmconnection_ROADMA(self):
291         response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-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(-3.3, res['roadm-connections'][0]['target-output-power'])
296
297     def test_22_get_roadmconnection_ROADMC(self):
298         response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-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'])
302
303     def test_23_service_power_setup_XPDRC_XPDRA(self):
304         url = "{}/operations/transportpce-olm:service-power-setup"
305         data = {
306             "input": {
307                 "service-name": "test",
308                 "wave-number": 1,
309                 "nodes": [
310                     {
311                         "dest-tp": "XPDR1-NETWORK1",
312                         "src-tp": "XPDR1-CLIENT1",
313                         "node-id": "XPDRC01"
314                     },
315                     {
316                         "dest-tp": "DEG2-TTP-TXRX",
317                         "src-tp": "SRG1-PP1-TXRX",
318                         "node-id": "ROADMC01"
319                     },
320                     {
321                         "src-tp": "DEG1-TTP-TXRX",
322                         "dest-tp": "SRG1-PP1-TXRX",
323                         "node-id": "ROADMA01"
324                     },
325                     {
326                         "src-tp": "XPDR1-NETWORK1",
327                         "dest-tp": "XPDR1-CLIENT1",
328                         "node-id": "XPDRA01"
329                     }
330                 ]
331             }
332         }
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"])
337
338     def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
339         response = test_utils.check_netconf_node_request(
340             "XPDRC01",
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(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
345         self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
346
347     def test_25_get_roadmconnection_ROADMC(self):
348         response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-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, res['roadm-connections'][0]['target-output-power'])
353
354     def test_26_service_power_turndown_XPDRA_XPDRC(self):
355         url = "{}/operations/transportpce-olm:service-power-turndown"
356         data = {
357             "input": {
358                 "service-name": "test",
359                 "wave-number": 1,
360                 "nodes": [
361                     {
362                         "dest-tp": "XPDR1-NETWORK1",
363                         "src-tp": "XPDR1-CLIENT1",
364                         "node-id": "XPDRA01"
365                     },
366                     {
367                         "dest-tp": "DEG1-TTP-TXRX",
368                         "src-tp": "SRG1-PP1-TXRX",
369                         "node-id": "ROADMA01"
370                     },
371                     {
372                         "dest-tp": "SRG1-PP1-TXRX",
373                         "src-tp": "DEG2-TTP-TXRX",
374                         "node-id": "ROADMC01"
375                     },
376                     {
377                         "dest-tp": "XPDR1-CLIENT1",
378                         "src-tp": "XPDR1-NETWORK1",
379                         "node-id": "XPDRC01"
380                     }
381                 ]
382             }
383         }
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"])
388
389     def test_27_get_roadmconnection_ROADMA(self):
390         response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-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'])
395
396     def test_28_get_roadmconnection_ROADMC(self):
397         response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-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'])
401
402     def test_29_servicePath_delete_AToZ(self):
403         response = test_utils.service_path_request("delete", "test", "1",
404                                                    [{"node-id": "XPDRA01",
405                                                      "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
406                                                     {"node-id": "ROADMA01",
407                                                      "dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
408                                                     {"node-id": "ROADMC01",
409                                                      "dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
410                                                     {"node-id": "XPDRC01",
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"])
415         time.sleep(10)
416
417     def test_30_servicePath_delete_ZToA(self):
418         response = test_utils.service_path_request("delete", "test", "1",
419                                                    [{"node-id": "XPDRC01",
420                                                      "dest-tp": "XPDR1-NETWORK1", "src-tp": "XPDR1-CLIENT1"},
421                                                     {"node-id": "ROADMC01",
422                                                      "dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP1-TXRX"},
423                                                     {"node-id": "ROADMA01",
424                                                      "src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
425                                                     {"node-id": "XPDRA01",
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"])
430         time.sleep(10)
431
432     #"""to test case where SRG where the xpdr is connected to has no optical range data"""
433
434     def test_31_connect_xprdA_to_roadmA(self):
435         response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
436                                                           "ROADMA01", "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"])
440
441     def test_32_connect_roadmA_to_xpdrA(self):
442         response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
443                                                           "ROADMA01", "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"])
447
448     def test_33_servicePath_create_AToZ(self):
449         response = test_utils.service_path_request("create", "test2", "2",
450                                                    [{"node-id": "XPDRA01",
451                                                      "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
452                                                     {"node-id": "ROADMA01",
453                                                      "dest-tp": "DEG1-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"])
457         # time.sleep(40)
458         time.sleep(10)
459
460     def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
461         response = test_utils.check_netconf_node_request(
462             "XPDRA01",
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'])
468
469     def test_35_servicePath_delete_AToZ(self):
470         response = test_utils.service_path_request("delete", "test", "1",
471                                                    [{"node-id": "XPDRA01",
472                                                      "dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
473                                                     {"node-id": "ROADMA01",
474                                                      "dest-tp": "DEG1-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"])
478         time.sleep(10)
479
480     def test_36_xpdrA_device_disconnected(self):
481         response = test_utils.unmount_device("XPDRA01")
482         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
483
484     def test_37_xpdrC_device_disconnected(self):
485         response = test_utils.unmount_device("XPDRC01")
486         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
487
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"])
495         time.sleep(5)
496
497     def test_39_rdmA_device_disconnected(self):
498         response = test_utils.unmount_device("ROADMA01")
499         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
500
501     def test_40_rdmC_device_disconnected(self):
502         response = test_utils.unmount_device("ROADMC01")
503         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
504
505
506 if __name__ == "__main__":
507     unittest.main(verbosity=2)