Fix some typos in functional tests naming
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test05_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 # pylint: disable=wrong-import-order
19 import sys
20 sys.path.append('transportpce_tests/common/')
21 # pylint: disable=wrong-import-position
22 # pylint: disable=import-error
23 import test_utils  # nopep8
24
25
26 class TransportOlmTesting(unittest.TestCase):
27
28     processes = None
29     NODE_VERSION = '1.2.1'
30
31     @classmethod
32     def setUpClass(cls):
33         cls.processes = test_utils.start_tpce()
34         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
35                                                ('roadma-full', cls.NODE_VERSION),
36                                                ('roadmc-full', cls.NODE_VERSION),
37                                                ('xpdrc', cls.NODE_VERSION)])
38
39     @classmethod
40     def tearDownClass(cls):
41         # pylint: disable=not-an-iterable
42         for process in cls.processes:
43             test_utils.shutdown_process(process)
44         print("all processes killed")
45
46     def setUp(self):
47         # pylint: disable=consider-using-f-string
48         print("execution of {}".format(self.id().split(".")[-1]))
49         time.sleep(1)
50
51     def test_01_xpdrA_device_connected(self):
52         response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION))
53         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
54
55     def test_02_xpdrC_device_connected(self):
56         response = test_utils.mount_device("XPDRC01", ('xpdrc', self.NODE_VERSION))
57         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
58
59     def test_03_rdmA_device_connected(self):
60         response = test_utils.mount_device("ROADMA01", ('roadma-full', self.NODE_VERSION))
61         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
62
63     def test_04_rdmC_device_connected(self):
64         response = test_utils.mount_device("ROADMC01", ('roadmc-full', self.NODE_VERSION))
65         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
66
67     def test_05_connect_xpdrA_to_roadmA(self):
68         response = test_utils.transportpce_api_rpc_request(
69             'transportpce-networkutils', 'init-xpdr-rdm-links',
70             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1',
71                              'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
72         self.assertEqual(response['status_code'], requests.codes.ok)
73
74     def test_06_connect_roadmA_to_xpdrA(self):
75         response = test_utils.transportpce_api_rpc_request(
76             'transportpce-networkutils', 'init-rdm-xpdr-links',
77             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1',
78                              'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
79         self.assertEqual(response['status_code'], requests.codes.ok)
80
81     def test_07_connect_xpdrC_to_roadmC(self):
82         response = test_utils.transportpce_api_rpc_request(
83             'transportpce-networkutils', 'init-xpdr-rdm-links',
84             {'links-input': {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1',
85                              'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
86         self.assertEqual(response['status_code'], requests.codes.ok)
87
88     def test_08_connect_roadmC_to_xpdrC(self):
89         response = test_utils.transportpce_api_rpc_request(
90             'transportpce-networkutils', 'init-rdm-xpdr-links',
91             {'links-input': {'xpdr-node': 'XPDRC01', 'xpdr-num': '1', 'network-num': '1',
92                              'rdm-node': 'ROADMC01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
93         self.assertEqual(response['status_code'], requests.codes.ok)
94
95     def test_09_create_OTS_ROADMA(self):
96         response = test_utils.transportpce_api_rpc_request(
97             'transportpce-device-renderer', 'create-ots-oms',
98             {
99                 'node-id': 'ROADMA01',
100                 'logical-connection-point': 'DEG1-TTP-TXRX'
101             })
102         self.assertEqual(response['status_code'], requests.codes.ok)
103
104     def test_10_create_OTS_ROADMC(self):
105         response = test_utils.transportpce_api_rpc_request(
106             'transportpce-device-renderer', 'create-ots-oms',
107             {
108                 'node-id': 'ROADMC01',
109                 'logical-connection-point': 'DEG2-TTP-TXRX'
110             })
111         self.assertEqual(response['status_code'], requests.codes.ok)
112
113     def test_11_get_PM_ROADMA(self):
114         response = test_utils.transportpce_api_rpc_request(
115             'transportpce-olm', 'get-pm',
116             {
117                 'node-id': 'ROADMA01',
118                 'resource-type': 'interface',
119                 'granularity': '15min',
120                 'resource-identifier': {
121                     'resource-name': 'OTS-DEG1-TTP-TXRX'
122                 }
123             })
124         self.assertEqual(response['status_code'], requests.codes.ok)
125         measures = {}
126         for measure in response['output']['measurements']:
127             if bool(measure['pmparameter-name']) and bool(measure['pmparameter-value']):
128                 measures[measure['pmparameter-name']] = float(measure['pmparameter-value'])
129         self.assertIn('OpticalPowerOutput', list(measures.keys()))
130         self.assertIn('OpticalReturnLoss', list(measures.keys()))
131         self.assertIn('OpticalPowerInput', list(measures.keys()))
132         self.assertEqual(measures['OpticalPowerOutput'], 2.5)
133         self.assertEqual(measures['OpticalReturnLoss'], 49.9)
134         self.assertEqual(measures['OpticalPowerInput'], 3.0)
135
136     def test_12_get_PM_ROADMC(self):
137         response = test_utils.transportpce_api_rpc_request(
138             'transportpce-olm', 'get-pm',
139             {
140                 'node-id': 'ROADMC01',
141                 'resource-type': 'interface',
142                 'granularity': '15min',
143                 'resource-identifier': {
144                     'resource-name': 'OTS-DEG2-TTP-TXRX'
145                 }
146             })
147         self.assertEqual(response['status_code'], requests.codes.ok)
148         measures = {}
149         for measure in response['output']['measurements']:
150             if bool(measure['pmparameter-name']) and bool(measure['pmparameter-value']):
151                 measures[measure['pmparameter-name']] = float(measure['pmparameter-value'])
152         self.assertIn('OpticalPowerOutput', list(measures.keys()))
153         self.assertIn('OpticalReturnLoss', list(measures.keys()))
154         self.assertIn('OpticalPowerInput', list(measures.keys()))
155         self.assertEqual(measures['OpticalPowerOutput'], 18.1)
156         self.assertEqual(measures['OpticalReturnLoss'], 48.8)
157         self.assertEqual(measures['OpticalPowerInput'], -3.2)
158
159     def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
160         response = test_utils.transportpce_api_rpc_request(
161             'transportpce-olm', 'calculate-spanloss-base',
162             {
163                 'src-type': 'link',
164                 'link-id': 'ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX'
165             })
166         self.assertEqual(response['status_code'], requests.codes.ok)
167         self.assertIn('Success',
168                       response['output']['result'])
169         self.assertIn({
170             'spanloss': '5.7',
171             'link-id': 'ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX'
172         }, response['output']['spans'])
173         time.sleep(5)
174
175     def test_14_calculate_span_loss_base_all(self):
176         response = test_utils.transportpce_api_rpc_request(
177             'transportpce-olm', 'calculate-spanloss-base',
178             {
179                 'src-type': 'all'
180             })
181         self.assertEqual(response['status_code'], requests.codes.ok)
182         self.assertIn('Success',
183                       response['output']['result'])
184         self.assertIn({
185             'spanloss': '15.1',
186             'link-id': 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX'
187         }, response['output']['spans'])
188         self.assertIn({
189             'spanloss': '5.7',
190             'link-id': 'ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX'
191         }, response['output']['spans'])
192         time.sleep(5)
193
194     def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
195         response = test_utils.check_node_attribute2_request(
196             'ROADMA01', 'interface', 'OTS-DEG1-TTP-TXRX', 'org-openroadm-optical-transport-interfaces:ots')
197         self.assertEqual(response['status_code'], requests.codes.ok)
198         self.assertEqual(float(response['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit']), 5.7)
199         self.assertEqual(float(response['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive']), 15.1)
200         # FIXME test_utils.check_node_attribute2_request retrives values in config DS
201         #      but values are different (6 and 15) in nonconfig/operational DS and should not
202         #      Honeynode simulator bug ?
203         # Reopen similar issues found in RSPN tests at TRNSPRTPCE-591
204
205     def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
206         response = test_utils.check_node_attribute2_request(
207             'ROADMC01', 'interface', 'OTS-DEG2-TTP-TXRX', 'org-openroadm-optical-transport-interfaces:ots')
208         self.assertEqual(response['status_code'], requests.codes.ok)
209         self.assertEqual(float(response['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit']), 15.1)
210         self.assertEqual(float(response['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive']), 5.7)
211
212     def test_17_servicePath_create_AToZ(self):
213         response = test_utils.transportpce_api_rpc_request(
214             'transportpce-device-renderer', 'service-path',
215             {
216                 'service-name': 'test',
217                 'wave-number': '1',
218                 'modulation-format': 'dp-qpsk',
219                 'operation': 'create',
220                 'nodes':
221                 [{'node-id': 'XPDRA01',
222                   'dest-tp': 'XPDR1-NETWORK1', 'src-tp': 'XPDR1-CLIENT1'},
223                  {'node-id': 'ROADMA01',
224                     'dest-tp': 'DEG1-TTP-TXRX', 'src-tp': 'SRG1-PP1-TXRX'},
225                     {'node-id': 'ROADMC01',
226                      'dest-tp': 'SRG1-PP1-TXRX', 'src-tp': 'DEG2-TTP-TXRX'},
227                     {'node-id': 'XPDRC01',
228                      'dest-tp': 'XPDR1-CLIENT1', 'src-tp': 'XPDR1-NETWORK1'}],
229                 'center-freq': 196.1,
230                 'nmc-width': 40,
231                 'min-freq': 196.075,
232                 'max-freq': 196.125,
233                 'lower-spectral-slot-number': 761,
234                 'higher-spectral-slot-number': 768
235             })
236         self.assertEqual(response['status_code'], requests.codes.ok)
237         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
238         time.sleep(10)
239
240     def test_18_servicePath_create_ZToA(self):
241         response = test_utils.transportpce_api_rpc_request(
242             'transportpce-device-renderer', 'service-path',
243             {
244                 'service-name': 'test',
245                 'wave-number': '1',
246                 'modulation-format': 'dp-qpsk',
247                 'operation': 'create',
248                 'nodes':
249                 [{'node-id': 'XPDRC01',
250                   'dest-tp': 'XPDR1-NETWORK1', 'src-tp': 'XPDR1-CLIENT1'},
251                  {'node-id': 'ROADMC01',
252                     'dest-tp': 'DEG2-TTP-TXRX', 'src-tp': 'SRG1-PP1-TXRX'},
253                     {'node-id': 'ROADMA01',
254                      'src-tp': 'DEG1-TTP-TXRX', 'dest-tp': 'SRG1-PP1-TXRX'},
255                     {'node-id': 'XPDRA01',
256                      'src-tp': 'XPDR1-NETWORK1', 'dest-tp': 'XPDR1-CLIENT1'}],
257                 'center-freq': 196.1,
258                 'nmc-width': 40,
259                 'min-freq': 196.075,
260                 'max-freq': 196.125,
261                 'lower-spectral-slot-number': 761,
262                 'higher-spectral-slot-number': 768
263             })
264         self.assertEqual(response['status_code'], requests.codes.ok)
265         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
266         time.sleep(10)
267
268     def test_19_service_power_setup_XPDRA_XPDRC(self):
269         response = test_utils.transportpce_api_rpc_request(
270             'transportpce-olm', 'service-power-setup',
271             {
272                 'service-name': 'test',
273                 'wave-number': 1,
274                 'nodes': [
275                     {
276                         'dest-tp': 'XPDR1-NETWORK1',
277                         'src-tp': 'XPDR1-CLIENT1',
278                         'node-id': 'XPDRA01'
279                     },
280                     {
281                         'dest-tp': 'DEG1-TTP-TXRX',
282                         'src-tp': 'SRG1-PP1-TXRX',
283                         'node-id': 'ROADMA01'
284                     },
285                     {
286                         'dest-tp': 'SRG1-PP1-TXRX',
287                         'src-tp': 'DEG2-TTP-TXRX',
288                         'node-id': 'ROADMC01'
289                     },
290                     {
291                         'dest-tp': 'XPDR1-CLIENT1',
292                         'src-tp': 'XPDR1-NETWORK1',
293                         'node-id': 'XPDRC01'
294                     }
295                 ],
296                 'center-freq': 196.1,
297                 'nmc-width': 40,
298                 'min-freq': 196.075,
299                 'max-freq': 196.125,
300                 'lower-spectral-slot-number': 761,
301                 'higher-spectral-slot-number': 768
302             })
303         self.assertEqual(response['status_code'], requests.codes.ok)
304         self.assertIn('Success', response['output']['result'])
305
306     def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
307         response = test_utils.check_node_attribute2_request(
308             'XPDRA01', 'interface', 'XPDR1-NETWORK1-761:768', 'org-openroadm-optical-channel-interfaces:och')
309         self.assertEqual(response['status_code'], requests.codes.ok)
310         self.assertEqual(float(response['org-openroadm-optical-channel-interfaces:och']['transmit-power']), 0)
311         self.assertEqual(int(response['org-openroadm-optical-channel-interfaces:och']['wavelength-number']), 1)
312
313     def test_21_get_roadmconnection_ROADMA(self):
314         response = test_utils.check_node_attribute_request(
315             'ROADMA01', 'roadm-connections', 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768')
316         self.assertEqual(response['status_code'], requests.codes.ok)
317         self.assertEqual("gainLoss", response['roadm-connections'][0]['opticalControlMode'])
318         self.assertEqual(float(response['roadm-connections'][0]['target-output-power']), -3.3)
319
320     def test_22_get_roadmconnection_ROADMC(self):
321         response = test_utils.check_node_attribute_request(
322             'ROADMC01', 'roadm-connections', 'DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768')
323         self.assertEqual(response['status_code'], requests.codes.ok)
324         self.assertEqual("power", response['roadm-connections'][0]['opticalControlMode'])
325
326     def test_23_service_power_setup_XPDRC_XPDRA(self):
327         response = test_utils.transportpce_api_rpc_request(
328             'transportpce-olm', 'service-power-setup',
329             {
330                 'service-name': 'test',
331                 'wave-number': 1,
332                 'nodes': [
333                     {
334                         'dest-tp': 'XPDR1-NETWORK1',
335                         'src-tp': 'XPDR1-CLIENT1',
336                         'node-id': 'XPDRC01'
337                     },
338                     {
339                         'dest-tp': 'DEG2-TTP-TXRX',
340                         'src-tp': 'SRG1-PP1-TXRX',
341                         'node-id': 'ROADMC01'
342                     },
343                     {
344                         'src-tp': 'DEG1-TTP-TXRX',
345                         'dest-tp': 'SRG1-PP1-TXRX',
346                         'node-id': 'ROADMA01'
347                     },
348                     {
349                         'src-tp': 'XPDR1-NETWORK1',
350                         'dest-tp': 'XPDR1-CLIENT1',
351                         'node-id': 'XPDRA01'
352                     }
353                 ],
354                 'center-freq': 196.1,
355                 'nmc-width': 40,
356                 'min-freq': 196.075,
357                 'max-freq': 196.125,
358                 'lower-spectral-slot-number': 761,
359                 'higher-spectral-slot-number': 768
360             })
361         self.assertEqual(response['status_code'], requests.codes.ok)
362         self.assertIn('Success', response['output']['result'])
363
364     def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
365         response = test_utils.check_node_attribute2_request(
366             'XPDRC01', 'interface', 'XPDR1-NETWORK1-761:768', 'org-openroadm-optical-channel-interfaces:och')
367         self.assertEqual(response['status_code'], requests.codes.ok)
368         self.assertEqual(float(response['org-openroadm-optical-channel-interfaces:och']['transmit-power']), 0)
369         self.assertEqual(int(response['org-openroadm-optical-channel-interfaces:och']['wavelength-number']), 1)
370
371     def test_25_get_roadmconnection_ROADMC(self):
372         response = test_utils.check_node_attribute_request(
373             'ROADMC01', 'roadm-connections', 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768')
374         self.assertEqual(response['status_code'], requests.codes.ok)
375         self.assertEqual("gainLoss", response['roadm-connections'][0]['opticalControlMode'])
376         self.assertEqual(float(response['roadm-connections'][0]['target-output-power']), -0.63)
377
378     def test_26_service_power_turndown_XPDRA_XPDRC(self):
379         response = test_utils.transportpce_api_rpc_request(
380             'transportpce-olm', 'service-power-turndown',
381             {
382                 'service-name': 'test',
383                 'wave-number': 1,
384                 'nodes': [
385                     {
386                         'dest-tp': 'XPDR1-NETWORK1',
387                         'src-tp': 'XPDR1-CLIENT1',
388                         'node-id': 'XPDRA01'
389                     },
390                     {
391                         'dest-tp': 'DEG1-TTP-TXRX',
392                         'src-tp': 'SRG1-PP1-TXRX',
393                         'node-id': 'ROADMA01'
394                     },
395                     {
396                         'dest-tp': 'SRG1-PP1-TXRX',
397                         'src-tp': 'DEG2-TTP-TXRX',
398                         'node-id': 'ROADMC01'
399                     },
400                     {
401                         'dest-tp': 'XPDR1-CLIENT1',
402                         'src-tp': 'XPDR1-NETWORK1',
403                         'node-id': 'XPDRC01'
404                     }
405                 ],
406                 'center-freq': 196.1,
407                 'nmc-width': 40,
408                 'min-freq': 196.075,
409                 'max-freq': 196.125,
410                 'lower-spectral-slot-number': 761,
411                 'higher-spectral-slot-number': 768
412             })
413         self.assertEqual(response['status_code'], requests.codes.ok)
414         self.assertIn('Success', response['output']['result'])
415
416     def test_27_get_roadmconnection_ROADMA(self):
417         response = test_utils.check_node_attribute_request(
418             'ROADMA01', 'roadm-connections', 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768')
419         self.assertEqual(response['status_code'], requests.codes.ok)
420         self.assertEqual("off", response['roadm-connections'][0]['opticalControlMode'])
421         self.assertEqual(float(response['roadm-connections'][0]['target-output-power']), -60)
422
423     def test_28_get_roadmconnection_ROADMC(self):
424         response = test_utils.check_node_attribute_request(
425             'ROADMC01', 'roadm-connections', 'DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768')
426         self.assertEqual(response['status_code'], requests.codes.ok)
427         self.assertEqual("off", response['roadm-connections'][0]['opticalControlMode'])
428
429     def test_29_servicePath_delete_AToZ(self):
430         response = test_utils.transportpce_api_rpc_request(
431             'transportpce-device-renderer', 'service-path',
432             {
433                 'service-name': 'test',
434                 'wave-number': '1',
435                 'modulation-format': 'dp-qpsk',
436                 'operation': 'delete',
437                 'nodes':
438                 [{'node-id': 'XPDRA01',
439                   'dest-tp': 'XPDR1-NETWORK1', 'src-tp': 'XPDR1-CLIENT1'},
440                  {'node-id': 'ROADMA01',
441                     'dest-tp': 'DEG1-TTP-TXRX', 'src-tp': 'SRG1-PP1-TXRX'},
442                     {'node-id': 'ROADMC01',
443                      'dest-tp': 'SRG1-PP1-TXRX', 'src-tp': 'DEG2-TTP-TXRX'},
444                     {'node-id': 'XPDRC01',
445                      'dest-tp': 'XPDR1-CLIENT1', 'src-tp': 'XPDR1-NETWORK1'}],
446                 'center-freq': 196.1,
447                 'nmc-width': 40,
448                 'min-freq': 196.075,
449                 'max-freq': 196.125,
450                 'lower-spectral-slot-number': 761,
451                 'higher-spectral-slot-number': 768
452             })
453         self.assertEqual(response['status_code'], requests.codes.ok)
454         self.assertIn('Request processed', response['output']['result'])
455         time.sleep(10)
456
457     def test_30_servicePath_delete_ZToA(self):
458         response = test_utils.transportpce_api_rpc_request(
459             'transportpce-device-renderer', 'service-path',
460             {
461                 'service-name': 'test',
462                 'wave-number': '1',
463                 'modulation-format': 'dp-qpsk',
464                 'operation': 'delete',
465                 'nodes':
466                 [{'node-id': 'XPDRC01',
467                              'dest-tp': 'XPDR1-NETWORK1', 'src-tp': 'XPDR1-CLIENT1'},
468                  {'node-id': 'ROADMC01',
469                              'dest-tp': 'DEG2-TTP-TXRX', 'src-tp': 'SRG1-PP1-TXRX'},
470                  {'node-id': 'ROADMA01',
471                              'src-tp': 'DEG1-TTP-TXRX', 'dest-tp': 'SRG1-PP1-TXRX'},
472                  {'node-id': 'XPDRA01',
473                              'src-tp': 'XPDR1-NETWORK1', 'dest-tp': 'XPDR1-CLIENT1'}],
474                 'center-freq': 196.1,
475                 'nmc-width': 40,
476                 'min-freq': 196.075,
477                 'max-freq': 196.125,
478                 'lower-spectral-slot-number': 761,
479                 'higher-spectral-slot-number': 768
480             })
481         self.assertEqual(response['status_code'], requests.codes.ok)
482         self.assertIn('Request processed', response['output']['result'])
483         time.sleep(10)
484
485     # """to test case where SRG where the xpdr is connected to has no optical range data"""
486
487     def test_31_connect_xpdrA_to_roadmA(self):
488         response = test_utils.transportpce_api_rpc_request(
489             'transportpce-networkutils', 'init-xpdr-rdm-links',
490             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2',
491                              'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
492         self.assertEqual(response['status_code'], requests.codes.ok)
493
494     def test_32_connect_roadmA_to_xpdrA(self):
495         response = test_utils.transportpce_api_rpc_request(
496             'transportpce-networkutils', 'init-rdm-xpdr-links',
497             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '2',
498                              'rdm-node': 'ROADMA01', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
499         self.assertEqual(response['status_code'], requests.codes.ok)
500
501     def test_33_servicePath_create_AToZ(self):
502         response = test_utils.transportpce_api_rpc_request(
503             'transportpce-device-renderer', 'service-path',
504             {
505                 'service-name': 'test2',
506                 'wave-number': '2',
507                 'modulation-format': 'dp-qpsk',
508                 'operation': 'create',
509                 'nodes':
510                 [{'node-id': 'XPDRA01',
511                   'dest-tp': 'XPDR1-NETWORK2', 'src-tp': 'XPDR1-CLIENT2'},
512                  {'node-id': 'ROADMA01',
513                   'dest-tp': 'DEG1-TTP-TXRX', 'src-tp': 'SRG1-PP2-TXRX'}],
514                 'center-freq': 196.05,
515                 'nmc-width': 40,
516                 'min-freq': 196.025,
517                 'max-freq': 196.075,
518                 'lower-spectral-slot-number': 753,
519                 'higher-spectral-slot-number': 760
520             })
521         self.assertEqual(response['status_code'], requests.codes.ok)
522         self.assertIn('Interfaces created successfully for nodes', response['output']['result'])
523         time.sleep(10)
524
525     def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
526         response = test_utils.check_node_attribute2_request(
527             'XPDRA01', 'interface', 'XPDR1-NETWORK2-753:760', 'org-openroadm-optical-channel-interfaces:och')
528         self.assertEqual(response['status_code'], requests.codes.ok)
529         self.assertEqual(float(response['org-openroadm-optical-channel-interfaces:och']['transmit-power']), -5)
530         self.assertEqual(int(response['org-openroadm-optical-channel-interfaces:och']['wavelength-number']), 2)
531
532     def test_35_servicePath_delete_AToZ(self):
533         response = test_utils.transportpce_api_rpc_request(
534             'transportpce-device-renderer', 'service-path',
535             {
536                 'service-name': 'test2',
537                 'wave-number': '2',
538                 'modulation-format': 'dp-qpsk',
539                 'operation': 'delete',
540                 'nodes':
541                 [{'node-id': 'XPDRA01',
542                   'dest-tp': 'XPDR1-NETWORK2', 'src-tp': 'XPDR1-CLIENT2'},
543                  {'node-id': 'ROADMA01',
544                   'dest-tp': 'DEG1-TTP-TXRX', 'src-tp': 'SRG1-PP2-TXRX'}],
545                 'center-freq': 196.05,
546                 'nmc-width': 40,
547                 'min-freq': 196.025,
548                 'max-freq': 196.075,
549                 'lower-spectral-slot-number': 753,
550                 'higher-spectral-slot-number': 760
551             })
552         self.assertEqual(response['status_code'], requests.codes.ok)
553         self.assertIn('Request processed', response['output']['result'])
554         time.sleep(10)
555
556     def test_36_xpdrA_device_disconnected(self):
557         response = test_utils.unmount_device("XPDRA01")
558         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
559
560     def test_37_xpdrC_device_disconnected(self):
561         response = test_utils.unmount_device("XPDRC01")
562         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
563
564     def test_38_calculate_span_loss_current(self):
565         response = test_utils.transportpce_api_rpc_request(
566             'transportpce-olm', 'calculate-spanloss-current',
567             None)
568         self.assertEqual(response['status_code'], requests.codes.ok)
569         self.assertIn('Success',
570                       response["output"]["result"])
571         time.sleep(5)
572
573     def test_39_rdmA_device_disconnected(self):
574         response = test_utils.unmount_device("ROADMA01")
575         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
576
577     def test_40_rdmC_device_disconnected(self):
578         response = test_utils.unmount_device("ROADMC01")
579         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
580
581
582 if __name__ == "__main__":
583     unittest.main(verbosity=2)