rationalize functional tests calls to service-path
[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 import unittest
13 import time
14 import json
15 import requests
16 from common import test_utils
17
18
19 class TransportOlmTesting(unittest.TestCase):
20
21     processes = None
22
23     @classmethod
24     def setUpClass(cls):
25         cls.processes = test_utils.start_tpce()
26         cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc'])
27
28     @classmethod
29     def tearDownClass(cls):
30         for process in cls.processes:
31             test_utils.shutdown_process(process)
32         print("all processes killed")
33
34     def setUp(self):
35         print("execution of {}".format(self.id().split(".")[-1]))
36         time.sleep(1)
37
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)
41
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)
45
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)
49
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)
53
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)
58         res = response.json()
59         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
60
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)
65         res = response.json()
66         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
67
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)
72         res = response.json()
73         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
74
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)
79         res = response.json()
80         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
81
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)
85         res = response.json()
86         self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
87                       res["output"]["result"])
88
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)
92         res = response.json()
93         self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
94                       res["output"]["result"])
95
96     def test_11_get_PM_ROADMA(self):
97         url = "{}/operations/transportpce-olm:get-pm"
98         data = {
99             "input": {
100                 "node-id": "ROADMA01",
101                 "resource-type": "interface",
102                 "granularity": "15min",
103                 "resource-identifier": {
104                     "resource-name": "OTS-DEG1-TTP-TXRX"
105                 }
106             }
107         }
108         response = test_utils.post_request(url, data)
109         self.assertEqual(response.status_code, requests.codes.ok)
110         res = response.json()
111         self.assertIn({
112             "pmparameter-name": "OpticalPowerOutput",
113             "pmparameter-value": "2.5"
114         }, res["output"]["measurements"])
115         self.assertIn({
116             "pmparameter-name": "OpticalReturnLoss",
117             "pmparameter-value": "49.9"
118         }, res["output"]["measurements"])
119         self.assertIn({
120             "pmparameter-name": "OpticalPowerInput",
121             "pmparameter-value": "3"
122         }, res["output"]["measurements"])
123
124     def test_12_get_PM_ROADMC(self):
125         url = "{}/operations/transportpce-olm:get-pm"
126         data = {
127             "input": {
128                 "node-id": "ROADMC01",
129                 "resource-type": "interface",
130                 "granularity": "15min",
131                 "resource-identifier": {
132                     "resource-name": "OTS-DEG2-TTP-TXRX"
133                 }
134             }
135         }
136         response = test_utils.post_request(url, data)
137         self.assertEqual(response.status_code, requests.codes.ok)
138         res = response.json()
139         self.assertIn({
140             "pmparameter-name": "OpticalPowerOutput",
141             "pmparameter-value": "18.1"
142         }, res["output"]["measurements"])
143         self.assertIn({
144             "pmparameter-name": "OpticalReturnLoss",
145             "pmparameter-value": "48.8"
146         }, res["output"]["measurements"])
147         self.assertIn({
148             "pmparameter-name": "OpticalPowerInput",
149             "pmparameter-value": "-3.2"
150         }, res["output"]["measurements"])
151
152     def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
153         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
154         data = {
155             "input": {
156                 "src-type": "link",
157                 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
158             }
159         }
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"])
165         self.assertIn({
166             "spanloss": "5.7",
167             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
168         }, res["output"]["spans"])
169         time.sleep(5)
170
171     def test_14_calculate_span_loss_base_all(self):
172         url = "{}/operations/transportpce-olm:calculate-spanloss-base"
173         data = {
174             "input": {
175                 "src-type": "all"
176             }
177         }
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"])
183         self.assertIn({
184             "spanloss": "15.1",
185             "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
186         }, res["output"]["spans"])
187         self.assertIn({
188             "spanloss": "5.7",
189             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
190         }, res["output"]["spans"])
191         time.sleep(5)
192
193     def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
194         response = test_utils.check_netconf_node_request(
195             "ROADMA01",
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'])
201
202     def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
203         response = test_utils.check_netconf_node_request(
204             "ROADMC01",
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'])
210
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"])
224         # time.sleep(40)
225         time.sleep(10)
226
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"])
240         # time.sleep(40)
241         time.sleep(10)
242
243     def test_19_service_power_setup_XPDRA_XPDRC(self):
244         url = "{}/operations/transportpce-olm:service-power-setup"
245         data = {
246             "input": {
247                 "service-name": "test",
248                 "wave-number": 1,
249                 "nodes": [
250                     {
251                         "dest-tp": "XPDR1-NETWORK1",
252                         "src-tp": "XPDR1-CLIENT1",
253                         "node-id": "XPDRA01"
254                     },
255                     {
256                         "dest-tp": "DEG1-TTP-TXRX",
257                         "src-tp": "SRG1-PP1-TXRX",
258                         "node-id": "ROADMA01"
259                     },
260                     {
261                         "dest-tp": "SRG1-PP1-TXRX",
262                         "src-tp": "DEG2-TTP-TXRX",
263                         "node-id": "ROADMC01"
264                     },
265                     {
266                         "dest-tp": "XPDR1-CLIENT1",
267                         "src-tp": "XPDR1-NETWORK1",
268                         "node-id": "XPDRC01"
269                     }
270                 ]
271             }
272         }
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"])
277
278     def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
279         response = test_utils.check_netconf_node_request(
280             "XPDRA01",
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'])
286
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'])
293
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'])
299
300     def test_23_service_power_setup_XPDRC_XPDRA(self):
301         url = "{}/operations/transportpce-olm:service-power-setup"
302         data = {
303             "input": {
304                 "service-name": "test",
305                 "wave-number": 1,
306                 "nodes": [
307                     {
308                         "dest-tp": "XPDR1-NETWORK1",
309                         "src-tp": "XPDR1-CLIENT1",
310                         "node-id": "XPDRC01"
311                     },
312                     {
313                         "dest-tp": "DEG2-TTP-TXRX",
314                         "src-tp": "SRG1-PP1-TXRX",
315                         "node-id": "ROADMC01"
316                     },
317                     {
318                         "src-tp": "DEG1-TTP-TXRX",
319                         "dest-tp": "SRG1-PP1-TXRX",
320                         "node-id": "ROADMA01"
321                     },
322                     {
323                         "src-tp": "XPDR1-NETWORK1",
324                         "dest-tp": "XPDR1-CLIENT1",
325                         "node-id": "XPDRA01"
326                     }
327                 ]
328             }
329         }
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"])
334
335     def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
336         response = test_utils.check_netconf_node_request(
337             "XPDRC01",
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'])
343
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'])
350
351     def test_26_service_power_turndown_XPDRA_XPDRC(self):
352         url = "{}/operations/transportpce-olm:service-power-turndown"
353         data = {
354             "input": {
355                 "service-name": "test",
356                 "wave-number": 1,
357                 "nodes": [
358                     {
359                         "dest-tp": "XPDR1-NETWORK1",
360                         "src-tp": "XPDR1-CLIENT1",
361                         "node-id": "XPDRA01"
362                     },
363                     {
364                         "dest-tp": "DEG1-TTP-TXRX",
365                         "src-tp": "SRG1-PP1-TXRX",
366                         "node-id": "ROADMA01"
367                     },
368                     {
369                         "dest-tp": "SRG1-PP1-TXRX",
370                         "src-tp": "DEG2-TTP-TXRX",
371                         "node-id": "ROADMC01"
372                     },
373                     {
374                         "dest-tp": "XPDR1-CLIENT1",
375                         "src-tp": "XPDR1-NETWORK1",
376                         "node-id": "XPDRC01"
377                     }
378                 ]
379             }
380         }
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"])
385
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'])
392
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'])
398
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"])
412         time.sleep(10)
413
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"])
427         time.sleep(10)
428
429     """to test case where SRG where the xpdr is connected to has no optical range data"""
430
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"])
437
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"])
444
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"])
454         # time.sleep(40)
455         time.sleep(10)
456
457     def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
458         response = test_utils.check_netconf_node_request(
459             "XPDRA01",
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'])
465
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"])
475         time.sleep(10)
476
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)
480
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)
484
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"])
492         time.sleep(5)
493
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)
497
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)
501
502
503 if __name__ == "__main__":
504     unittest.main(verbosity=2)