generalize RESTconf requests methods in func tests
[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_renderer_service_path_nominal.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 json
14 #from unittest.result import failfast
15 import requests
16 from common import test_utils
17
18
19 class TransportPCERendererTesting(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'])
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 test_01_rdm_device_connected(self):
35         response = test_utils.mount_device("ROADM-A1", 'roadma')
36         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
37
38     def test_02_xpdr_device_connected(self):
39         response = test_utils.mount_device("XPDR-A1", 'xpdra')
40         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
41
42     def test_03_rdm_portmapping(self):
43         url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
44         response = test_utils.get_request(url)
45         self.assertEqual(response.status_code, requests.codes.ok)
46         res = response.json()
47         self.assertIn(
48             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
49              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
50             res['nodes'][0]['mapping'])
51         self.assertIn(
52             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
53              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
54             res['nodes'][0]['mapping'])
55
56     def test_04_xpdr_portmapping(self):
57         url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
58         response = test_utils.get_request(url)
59         self.assertEqual(response.status_code, requests.codes.ok)
60         res = response.json()
61         self.assertIn(
62             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
63              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
64              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
65              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
66              'lcp-hash-val': 'AMkDwQ7xTmRI'},
67             res['nodes'][0]['mapping'])
68         self.assertIn(
69             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
70              'supporting-port': 'C1',
71              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
72              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
73              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
74              'lcp-hash-val': 'AJUUr6I5fALj'},
75             res['nodes'][0]['mapping'])
76
77     def test_05_service_path_create(self):
78         url = "{}/operations/transportpce-device-renderer:service-path"
79         data = {"renderer:input": {
80             "renderer:service-name": "service_test",
81             "renderer:wave-number": "7",
82             "renderer:modulation-format": "qpsk",
83             "renderer:operation": "create",
84             "renderer:nodes": [
85                 {"renderer:node-id": "ROADM-A1",
86                  "renderer:src-tp": "SRG1-PP3-TXRX",
87                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
88                 {"renderer:node-id": "XPDR-A1",
89                  "renderer:src-tp": "XPDR1-CLIENT1",
90                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
91         response = test_utils.post_request(url, data)
92         self.assertEqual(response.status_code, requests.codes.ok)
93         res = response.json()
94         self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
95
96     def test_06_service_path_create_rdm_check(self):
97         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
98                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
99                "interface/DEG1-TTP-TXRX-nmc-7"
100                )
101         response = test_utils.get_request(url)
102         self.assertEqual(response.status_code, requests.codes.ok)
103         res = response.json()
104         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
105         self.assertDictEqual(
106             dict({
107                  'name': 'DEG1-TTP-TXRX-nmc-7',
108                  'administrative-state': 'inService',
109                  'supporting-circuit-pack-name': '1/0',
110                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
111                  'supporting-port': 'L1'
112                  }, **res['interface'][0]),
113             res['interface'][0]
114         )
115         self.assertDictEqual(
116             {u'frequency': 195.8, u'width': 40},
117             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
118
119     def test_07_service_path_create_rdm_check(self):
120         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
121                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
122                "interface/DEG1-TTP-TXRX-mc-7"
123                )
124         response = test_utils.get_request(url)
125         self.assertEqual(response.status_code, requests.codes.ok)
126         res = response.json()
127         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
128         self.assertDictEqual(
129             dict({
130                  'name': 'DEG1-TTP-TXRX-mc-7',
131                  'administrative-state': 'inService',
132                  'supporting-circuit-pack-name': '1/0',
133                  'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
134                  'supporting-port': 'L1'
135                  }, **res['interface'][0]),
136             res['interface'][0]
137         )
138         self.assertDictEqual(
139             {u'min-freq': 195.775, u'max-freq': 195.825},
140             res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
141
142     def test_08_service_path_create_rdm_check(self):
143         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
144                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
145                "interface/SRG1-PP3-TXRX-nmc-7"
146                )
147         response = test_utils.get_request(url)
148         self.assertEqual(response.status_code, requests.codes.ok)
149         res = response.json()
150         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
151         self.assertDictEqual(
152             dict({
153                  'name': 'SRG1-PP3-TXRX-nmc-7',
154                  'administrative-state': 'inService',
155                  'supporting-circuit-pack-name': '3/0',
156                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
157                  'supporting-port': 'C3'
158                  }, **res['interface'][0]),
159             res['interface'][0]
160         )
161         self.assertDictEqual(
162             {u'frequency': 195.8, u'width': 40},
163             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
164
165     # -mc supporting interfaces must not be created for SRG, only degrees
166     def test_09_service_path_create_rdm_check(self):
167         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
168                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
169                "interface/SRG1-PP3-TXRX-mc-7"
170                )
171         response = test_utils.get_request(url)
172         self.assertEqual(response.status_code, requests.codes.not_found)
173         res = response.json()
174         self.assertIn(
175             {"error-type": "application", "error-tag": "data-missing",
176              "error-message": "Request could not be completed because the relevant data model content does not exist"},
177             res['errors']['error'])
178
179     def test_10_service_path_create_rdm_check(self):
180         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
181                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
182                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
183                )
184         response = test_utils.get_request(url)
185         self.assertEqual(response.status_code, requests.codes.ok)
186         res = response.json()
187         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
188         self.assertDictEqual(
189             dict({
190                  'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
191                  'opticalControlMode': 'off'
192                  }, **res['roadm-connections'][0]),
193             res['roadm-connections'][0]
194         )
195         self.assertDictEqual(
196             {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
197             res['roadm-connections'][0]['source'])
198         self.assertDictEqual(
199             {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
200             res['roadm-connections'][0]['destination'])
201
202     def test_11_service_path_create_xpdr_check(self):
203         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
204                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
205                "interface/XPDR1-NETWORK1-7"
206                )
207         response = test_utils.get_request(url)
208         self.assertEqual(response.status_code, requests.codes.ok)
209         res = response.json()
210         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
211         self.assertDictEqual(
212             dict({
213                  'name': 'XPDR1-NETWORK1-7',
214                  'administrative-state': 'inService',
215                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
216                  'type': 'org-openroadm-interfaces:opticalChannel',
217                  'supporting-port': '1'
218                  }, **res['interface'][0]),
219             res['interface'][0]
220         )
221         self.assertDictEqual(
222             {u'rate': u'org-openroadm-common-types:R100G',
223              u'transmit-power': -5,
224              u'frequency': 195.8},
225             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
226
227     def test_12_service_path_create_xpdr_check(self):
228         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
229                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
230                "interface/XPDR1-NETWORK1-OTU"
231                )
232         response = test_utils.get_request(url)
233         self.assertEqual(response.status_code, requests.codes.ok)
234         res = response.json()
235         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
236         self.assertDictEqual(
237             dict({
238                  'name': 'XPDR1-NETWORK1-OTU',
239                  'administrative-state': 'inService',
240                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
241                  'type': 'org-openroadm-interfaces:otnOtu',
242                  'supporting-port': '1',
243                  'supporting-interface': 'XPDR1-NETWORK1-7'
244                  }, **res['interface'][0]),
245             res['interface'][0]
246         )
247         input_dict_2 = {'tx-sapi': 'AMkDwQ7xTmRI',
248                         'expected-dapi': 'AMkDwQ7xTmRI',
249                         'rate': 'org-openroadm-otn-common-types:OTU4',
250                         'fec': 'scfec'}
251         self.assertDictEqual(input_dict_2,
252                              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
253
254     def test_13_service_path_create_xpdr_check(self):
255         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
256                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
257                "interface/XPDR1-NETWORK1-ODU"
258                )
259         response = test_utils.get_request(url)
260         self.assertEqual(response.status_code, requests.codes.ok)
261         res = response.json()
262         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
263         self.assertDictEqual(
264             dict({
265                  'name': 'XPDR1-NETWORK1-ODU',
266                  'administrative-state': 'inService',
267                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
268                  'type': 'org-openroadm-interfaces:otnOdu',
269                  'supporting-port': '1',
270                  'supporting-interface': 'XPDR1-NETWORK1-OTU'
271                  }, **res['interface'][0]),
272             res['interface'][0]
273         )
274         self.assertDictEqual(
275             dict({
276                  'rate': 'org-openroadm-otn-common-types:ODU4',
277                  u'monitoring-mode': u'terminated'
278                  }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
279             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
280         )
281         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
282                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
283
284     def test_14_service_path_create_xpdr_check(self):
285         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
286                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
287                "interface/XPDR1-CLIENT1-ETHERNET"
288                )
289         response = test_utils.get_request(url)
290         self.assertEqual(response.status_code, requests.codes.ok)
291         res = response.json()
292         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
293         self.assertDictEqual(
294             dict({
295                  'name': 'XPDR1-CLIENT1-ETHERNET',
296                  'administrative-state': 'inService',
297                  'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
298                  'type': 'org-openroadm-interfaces:ethernetCsmacd',
299                  'supporting-port': 'C1'
300                  }, **res['interface'][0]),
301             res['interface'][0]
302         )
303         self.assertDictEqual(
304             {u'fec': u'off', u'speed': 100000},
305             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
306
307     def test_15_service_path_create_xpdr_check(self):
308         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
309                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
310                "circuit-packs/1%2F0%2F1-PLUG-NET"
311                )
312         response = test_utils.get_request(url)
313         self.assertEqual(response.status_code, requests.codes.ok)
314         res = response.json()
315         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
316
317     def test_16_service_path_create_xpdr_check(self):
318         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
319                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
320                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
321                )
322         response = test_utils.get_request(url)
323         self.assertEqual(response.status_code, requests.codes.ok)
324         res = response.json()
325         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
326
327     def test_17_service_path_delete(self):
328         url = "{}/operations/transportpce-device-renderer:service-path"
329         data = {"renderer:input": {
330             "renderer:service-name": "service_test",
331             "renderer:wave-number": "7",
332             "renderer:operation": "delete",
333             "renderer:nodes": [
334                 {"renderer:node-id": "ROADM-A1",
335                  "renderer:src-tp": "SRG1-PP3-TXRX",
336                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
337                 {"renderer:node-id": "XPDR-A1",
338                  "renderer:src-tp": "XPDR1-CLIENT1",
339                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
340         response = test_utils.post_request(url, data)
341         self.assertEqual(response.status_code, requests.codes.ok)
342         self.assertEqual(response.json(), {
343             'output': {'result': 'Request processed', 'success': True}})
344
345     def test_18_service_path_delete_rdm_check(self):
346         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
347                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
348                "interface/DEG1-TTP-TXRX-mc-7"
349                )
350         response = test_utils.get_request(url)
351         self.assertEqual(response.status_code, requests.codes.not_found)
352         res = response.json()
353         self.assertIn(
354             {"error-type": "application", "error-tag": "data-missing",
355              "error-message": "Request could not be completed because the relevant data model content does not exist"},
356             res['errors']['error'])
357
358     def test_19_service_path_delete_rdm_check(self):
359         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
360                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
361                "interface/DEG1-TTP-TXRX-nmc-7"
362                )
363         response = test_utils.get_request(url)
364         self.assertEqual(response.status_code, requests.codes.not_found)
365         res = response.json()
366         self.assertIn(
367             {"error-type": "application", "error-tag": "data-missing",
368              "error-message": "Request could not be completed because the relevant data model content does not exist"},
369             res['errors']['error'])
370
371     def test_20_service_path_delete_rdm_check(self):
372         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
373                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
374                "interface/SRG1-PP3-TXRX-mc-7"
375                )
376         response = test_utils.get_request(url)
377         self.assertEqual(response.status_code, requests.codes.not_found)
378         res = response.json()
379         self.assertIn({
380             "error-type": "application",
381             "error-tag": "data-missing",
382             "error-message": "Request could not be completed because the relevant data model content does not exist"},
383             res['errors']['error'])
384
385     def test_21_service_path_delete_rdm_check(self):
386         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
387                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
388                "interface/SRG1-PP3-TXRX-nmc-7"
389                )
390         response = test_utils.get_request(url)
391         self.assertEqual(response.status_code, requests.codes.not_found)
392         res = response.json()
393         self.assertIn({
394             "error-type": "application",
395             "error-tag": "data-missing",
396             "error-message": "Request could not be completed because the relevant data model content does not exist"},
397             res['errors']['error'])
398
399     def test_22_service_path_delete_rdm_check(self):
400         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
401                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
402                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
403                )
404         response = test_utils.get_request(url)
405         self.assertEqual(response.status_code, requests.codes.not_found)
406         res = response.json()
407         self.assertIn({
408             "error-type": "application",
409             "error-tag": "data-missing",
410             "error-message": "Request could not be completed because the relevant data model content does not exist"},
411             res['errors']['error'])
412
413     def test_23_service_path_delete_xpdr_check(self):
414         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
415                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
416                "interface/XPDR1-NETWORK1-7"
417                )
418         response = test_utils.get_request(url)
419         self.assertEqual(response.status_code, requests.codes.not_found)
420         res = response.json()
421         self.assertIn({
422             "error-type": "application",
423             "error-tag": "data-missing",
424             "error-message": "Request could not be completed because the relevant data model content does not exist"},
425             res['errors']['error'])
426
427     def test_24_service_path_delete_xpdr_check(self):
428         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
429                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
430                "interface/XPDR1-NETWORK1-OTU"
431                )
432         response = test_utils.get_request(url)
433         self.assertEqual(response.status_code, requests.codes.not_found)
434         res = response.json()
435         self.assertIn({
436             "error-type": "application",
437             "error-tag": "data-missing",
438             "error-message": "Request could not be completed because the relevant data model content does not exist"},
439             res['errors']['error'])
440
441     def test_25_service_path_delete_xpdr_check(self):
442         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
443                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
444                "interface/XPDR1-NETWORK1-ODU"
445                )
446         response = test_utils.get_request(url)
447         self.assertEqual(response.status_code, requests.codes.not_found)
448         res = response.json()
449         self.assertIn({
450             "error-type": "application",
451             "error-tag": "data-missing",
452             "error-message": "Request could not be completed because the relevant data model content does not exist"},
453             res['errors']['error'])
454
455     def test_26_service_path_delete_xpdr_check(self):
456         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
457                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
458                "interface/XPDR1-CLIENT1-ETHERNET"
459                )
460         response = test_utils.get_request(url)
461         self.assertEqual(response.status_code, requests.codes.not_found)
462         res = response.json()
463         self.assertIn({
464             "error-type": "application",
465             "error-tag": "data-missing",
466             "error-message": "Request could not be completed because the relevant data model content does not exist"},
467             res['errors']['error'])
468
469     def test_27_service_path_delete_xpdr_check(self):
470         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
471                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
472                "circuit-packs/1%2F0%2F1-PLUG-NET"
473                )
474         response = test_utils.get_request(url)
475         self.assertEqual(response.status_code, requests.codes.ok)
476         res = response.json()
477         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
478
479     def test_28_service_path_delete_xpdr_check(self):
480         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
481                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
482                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
483                )
484         response = test_utils.get_request(url)
485         self.assertEqual(response.status_code, requests.codes.ok)
486         res = response.json()
487         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
488
489     def test_29_rdm_device_disconnected(self):
490         response = test_utils.unmount_device("ROADM-A1")
491         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
492
493     def test_30_xpdr_device_disconnected(self):
494         response = test_utils.unmount_device("XPDR-A1")
495         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
496
497
498 if __name__ == "__main__":
499     unittest.main(verbosity=2, failfast=False)