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