9533a64f350737a90625d7524285b9b55b145bd1
[transportpce.git] / tests / transportpce_tests / 1.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 requests
14 import time
15 import subprocess
16 import signal
17 import json
18 import os
19 import psutil
20 import shutil
21 from unittest.result import failfast
22 import test_utils
23
24
25 class TransportPCERendererTesting(unittest.TestCase):
26
27     honeynode_process1 = None
28     honeynode_process2 = None
29     odl_process = None
30     restconf_baseurl = "http://localhost:8181/restconf"
31
32 # START_IGNORE_XTESTING
33
34     @classmethod
35     def setUpClass(cls):
36         print("starting honeynode1...")
37         cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
38         time.sleep(20)
39
40         print("starting honeynode2...")
41         cls.honeynode_process2 = test_utils.start_roadma_honeynode()
42         time.sleep(20)
43
44         print("starting opendaylight...")
45         cls.odl_process = test_utils.start_tpce()
46         time.sleep(60)
47         print("opendaylight started")
48
49     @classmethod
50     def tearDownClass(cls):
51         for child in psutil.Process(cls.odl_process.pid).children():
52             child.send_signal(signal.SIGINT)
53             child.wait()
54         cls.odl_process.send_signal(signal.SIGINT)
55         cls.odl_process.wait()
56         for child in psutil.Process(cls.honeynode_process1.pid).children():
57             child.send_signal(signal.SIGINT)
58             child.wait()
59         cls.honeynode_process1.send_signal(signal.SIGINT)
60         cls.honeynode_process1.wait()
61         for child in psutil.Process(cls.honeynode_process2.pid).children():
62             child.send_signal(signal.SIGINT)
63             child.wait()
64         cls.honeynode_process2.send_signal(signal.SIGINT)
65         cls.honeynode_process2.wait()
66
67     def setUp(self):
68         print("execution of {}".format(self.id().split(".")[-1]))
69         time.sleep(10)
70
71 # END_IGNORE_XTESTING
72
73     def test_01_rdm_device_connected(self):
74         url = ("{}/config/network-topology:"
75                "network-topology/topology/topology-netconf/node/ROADMA01"
76                .format(self.restconf_baseurl))
77         data = {"node": [{
78             "node-id": "ROADMA01",
79             "netconf-node-topology:username": "admin",
80             "netconf-node-topology:password": "admin",
81             "netconf-node-topology:host": "127.0.0.1",
82             "netconf-node-topology:port": "17831",
83             "netconf-node-topology:tcp-only": "false",
84             "netconf-node-topology:pass-through": {}}]}
85         headers = {'content-type': 'application/json'}
86         response = requests.request(
87             "PUT", url, data=json.dumps(data), headers=headers,
88             auth=('admin', 'admin'))
89         self.assertEqual(response.status_code, requests.codes.created)
90         time.sleep(20)
91
92     def test_02_xpdr_device_connected(self):
93         url = ("{}/config/network-topology:"
94                "network-topology/topology/topology-netconf/node/XPDRA01"
95                .format(self.restconf_baseurl))
96         data = {"node": [{
97             "node-id": "XPDRA01",
98             "netconf-node-topology:username": "admin",
99             "netconf-node-topology:password": "admin",
100             "netconf-node-topology:host": "127.0.0.1",
101             "netconf-node-topology:port": "17830",
102             "netconf-node-topology:tcp-only": "false",
103             "netconf-node-topology:pass-through": {}}]}
104         headers = {'content-type': 'application/json'}
105         response = requests.request(
106             "PUT", url, data=json.dumps(data), headers=headers,
107             auth=('admin', 'admin'))
108         self.assertEqual(response.status_code, requests.codes.created)
109         time.sleep(20)
110
111     def test_03_rdm_portmapping(self):
112         url = ("{}/config/transportpce-portmapping:network/"
113                "nodes/ROADMA01"
114                .format(self.restconf_baseurl))
115         headers = {'content-type': 'application/json'}
116         response = requests.request(
117             "GET", url, headers=headers, auth=('admin', 'admin'))
118         self.assertEqual(response.status_code, requests.codes.ok)
119         res = response.json()
120         self.assertIn(
121             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
122              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
123             res['nodes'][0]['mapping'])
124         self.assertIn(
125             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
126              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
127             res['nodes'][0]['mapping'])
128
129     def test_04_xpdr_portmapping(self):
130         url = ("{}/config/transportpce-portmapping:network/"
131                "nodes/XPDRA01"
132                .format(self.restconf_baseurl))
133         headers = {'content-type': 'application/json'}
134         response = requests.request(
135             "GET", url, headers=headers, auth=('admin', 'admin'))
136         self.assertEqual(response.status_code, requests.codes.ok)
137         res = response.json()
138         self.assertIn(
139             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
140              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
141              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network'},
142             res['nodes'][0]['mapping'])
143         self.assertIn(
144             {'supporting-port': 'C1',
145              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
146              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
147              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
148             res['nodes'][0]['mapping'])
149
150     def test_05_service_path_create(self):
151         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
152         data = {"renderer:input": {
153             "renderer:service-name": "service_test",
154             "renderer:wave-number": "7",
155             "renderer:modulation-format": "qpsk",
156             "renderer:operation": "create",
157             "renderer:nodes": [
158                 {"renderer:node-id": "ROADMA01",
159                  "renderer:src-tp": "SRG1-PP7-TXRX",
160                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
161                 {"renderer:node-id": "XPDRA01",
162                  "renderer:src-tp": "XPDR1-CLIENT1",
163                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
164         headers = {'content-type': 'application/json'}
165         response = requests.request(
166             "POST", url, data=json.dumps(data),
167             headers=headers, auth=('admin', 'admin'))
168         self.assertEqual(response.status_code, requests.codes.ok)
169         res = response.json()
170         self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
171
172     def test_06_service_path_create_rdm_check(self):
173         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
174                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
175                "interface/DEG1-TTP-TXRX-7"
176                .format(self.restconf_baseurl))
177         headers = {'content-type': 'application/json'}
178         response = requests.request(
179             "GET", url, headers=headers, auth=('admin', 'admin'))
180         self.assertEqual(response.status_code, requests.codes.ok)
181         res = response.json()
182         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-7', 'administrative-state': 'inService',
183                                        'supporting-circuit-pack-name': '2/0',
184                                        'type': 'org-openroadm-interfaces:opticalChannel',
185                                        'supporting-port': 'L1'}, res['interface'][0])
186         self.assertDictEqual(
187             {'wavelength-number': 7},
188             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
189
190     def test_07_service_path_create_rdm_check(self):
191         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
192                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
193                "interface/SRG1-PP7-TXRX-7"
194                .format(self.restconf_baseurl))
195         headers = {'content-type': 'application/json'}
196         response = requests.request(
197             "GET", url, headers=headers, auth=('admin', 'admin'))
198         self.assertEqual(response.status_code, requests.codes.ok)
199         res = response.json()
200         self.assertDictContainsSubset(
201             {'name': 'SRG1-PP7-TXRX-7', 'administrative-state': 'inService',
202              'supporting-circuit-pack-name': '4/0',
203              'type': 'org-openroadm-interfaces:opticalChannel',
204              'supporting-port': 'C7'},
205             res['interface'][0])
206         self.assertDictEqual(
207             {'wavelength-number': 7},
208             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
209
210     def test_08_service_path_create_rdm_check(self):
211         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
212                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
213                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
214                .format(self.restconf_baseurl))
215         headers = {'content-type': 'application/json'}
216         response = requests.request(
217             "GET", url, headers=headers, auth=('admin', 'admin'))
218         self.assertEqual(response.status_code, requests.codes.ok)
219         res = response.json()
220         self.assertDictContainsSubset(
221             {'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
222              'wavelength-number': 7,
223              'opticalControlMode': 'off'},
224             res['roadm-connections'][0])
225         self.assertDictEqual(
226             {'src-if': 'SRG1-PP7-TXRX-7'},
227             res['roadm-connections'][0]['source'])
228         self.assertDictEqual(
229             {'dst-if': 'DEG1-TTP-TXRX-7'},
230             res['roadm-connections'][0]['destination'])
231
232     def test_09_service_path_create_xpdr_check(self):
233         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
234                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
235                "interface/XPDR1-NETWORK1-7"
236                .format(self.restconf_baseurl))
237         headers = {'content-type': 'application/json'}
238         response = requests.request(
239             "GET", url, headers=headers, auth=('admin', 'admin'))
240         self.assertEqual(response.status_code, requests.codes.ok)
241         res = response.json()
242         self.assertDictContainsSubset(
243             {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
244              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
245              'type': 'org-openroadm-interfaces:opticalChannel',
246              'supporting-port': '1'},
247             res['interface'][0])
248         self.assertDictEqual(
249             {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
250              u'transmit-power': -5,
251              u'wavelength-number': 7,
252              u'modulation-format': u'dp-qpsk'},
253             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
254
255     def test_10_service_path_create_xpdr_check(self):
256         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
257                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
258                "interface/XPDR1-NETWORK1-OTU"
259                .format(self.restconf_baseurl))
260         headers = {'content-type': 'application/json'}
261         response = requests.request(
262             "GET", url, headers=headers, auth=('admin', 'admin'))
263         self.assertEqual(response.status_code, requests.codes.ok)
264         res = response.json()
265         self.assertDictContainsSubset(
266             {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
267              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
268              'type': 'org-openroadm-interfaces:otnOtu',
269              'supporting-port': '1',
270              'supporting-interface': 'XPDR1-NETWORK1-7'},
271             res['interface'][0])
272         self.assertDictEqual(
273             {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
274              u'fec': u'scfec'},
275             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
276
277     def test_11_service_path_create_xpdr_check(self):
278         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
279                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
280                "interface/XPDR1-NETWORK1-ODU"
281                .format(self.restconf_baseurl))
282         headers = {'content-type': 'application/json'}
283         response = requests.request(
284             "GET", url, headers=headers, auth=('admin', 'admin'))
285         self.assertEqual(response.status_code, requests.codes.ok)
286         res = response.json()
287         self.assertDictContainsSubset(
288             {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
289              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
290              'type': 'org-openroadm-interfaces:otnOdu',
291              'supporting-port': '1',
292              'supporting-interface': 'XPDR1-NETWORK1-OTU'},
293             res['interface'][0])
294         self.assertDictContainsSubset(
295             {'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
296              u'monitoring-mode': u'terminated'},
297             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
298         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
299                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
300
301     def test_12_service_path_create_xpdr_check(self):
302         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
303                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
304                "interface/XPDR1-CLIENT1-ETHERNET"
305                .format(self.restconf_baseurl))
306         headers = {'content-type': 'application/json'}
307         response = requests.request(
308             "GET", url, headers=headers, auth=('admin', 'admin'))
309         self.assertEqual(response.status_code, requests.codes.ok)
310         res = response.json()
311         self.assertDictContainsSubset(
312             {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
313              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
314              'type': 'org-openroadm-interfaces:ethernetCsmacd',
315              'supporting-port': 'C1'},
316             res['interface'][0])
317         self.assertDictEqual(
318             {'speed': 100000,
319              'mtu': 9000,
320              'auto-negotiation': 'enabled',
321              'duplex': 'full',
322              'fec': 'off'},
323             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
324
325     def test_13_service_path_create_xpdr_check(self):
326         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
327                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
328                "circuit-packs/1%2F0%2F1-PLUG-NET"
329                .format(self.restconf_baseurl))
330         headers = {'content-type': 'application/json'}
331         response = requests.request(
332             "GET", url, headers=headers, auth=('admin', 'admin'))
333         self.assertEqual(response.status_code, requests.codes.ok)
334         res = response.json()
335         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
336
337     def test_14_service_path_delete(self):
338         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
339         data = {"renderer:input": {
340             "renderer:service-name": "service_test",
341             "renderer:wave-number": "7",
342             "renderer:operation": "delete",
343             "renderer:nodes": [
344                 {"renderer:node-id": "ROADMA01",
345                  "renderer:src-tp": "SRG1-PP7-TXRX",
346                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
347                 {"renderer:node-id": "XPDRA01",
348                  "renderer:src-tp": "XPDR1-CLIENT1",
349                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
350         headers = {'content-type': 'application/json'}
351         response = requests.request(
352             "POST", url, data=json.dumps(data),
353             headers=headers, auth=('admin', 'admin'))
354         self.assertEqual(response.status_code, requests.codes.ok)
355         self.assertEqual(response.json(), {
356             'output': {'result': 'Request processed', 'success': True}})
357
358     def test_15_service_path_delete_rdm_check(self):
359         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
360                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
361                "interface/DEG1-TTP-TXRX-7"
362                .format(self.restconf_baseurl))
363         headers = {'content-type': 'application/json'}
364         response = requests.request(
365             "GET", url, headers=headers, auth=('admin', 'admin'))
366         self.assertEqual(response.status_code, requests.codes.not_found)
367         res = response.json()
368         self.assertIn(
369             {"error-type": "application", "error-tag": "data-missing",
370              "error-message": "Request could not be completed because the relevant data model content does not exist"},
371             res['errors']['error'])
372
373     def test_16_service_path_delete_rdm_check(self):
374         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
375                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
376                "interface/SRG1-PP7-TXRX-7"
377                .format(self.restconf_baseurl))
378         headers = {'content-type': 'application/json'}
379         response = requests.request(
380             "GET", url, headers=headers, auth=('admin', 'admin'))
381         self.assertEqual(response.status_code, requests.codes.not_found)
382         res = response.json()
383         self.assertIn(
384             {"error-type": "application", "error-tag": "data-missing",
385                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
386             res['errors']['error'])
387
388     def test_17_service_path_delete_rdm_check(self):
389         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
390                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
391                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
392                .format(self.restconf_baseurl))
393         headers = {'content-type': 'application/json'}
394         response = requests.request(
395             "GET", url, headers=headers, auth=('admin', 'admin'))
396         self.assertEqual(response.status_code, requests.codes.not_found)
397         res = response.json()
398         self.assertIn(
399             {"error-type": "application", "error-tag": "data-missing",
400                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
401             res['errors']['error'])
402
403     def test_18_service_path_delete_xpdr_check(self):
404         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
405                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
406                "interface/XPDR1-NETWORK1-7"
407                .format(self.restconf_baseurl))
408         headers = {'content-type': 'application/json'}
409         response = requests.request(
410             "GET", url, headers=headers, auth=('admin', 'admin'))
411         self.assertEqual(response.status_code, requests.codes.not_found)
412         res = response.json()
413         self.assertIn(
414             {"error-type": "application", "error-tag": "data-missing",
415                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
416             res['errors']['error'])
417
418     def test_19_service_path_delete_xpdr_check(self):
419         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
420                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
421                "interface/XPDR1-NETWORK1-OTU"
422                .format(self.restconf_baseurl))
423         headers = {'content-type': 'application/json'}
424         response = requests.request(
425             "GET", url, headers=headers, auth=('admin', 'admin'))
426         self.assertEqual(response.status_code, requests.codes.not_found)
427         res = response.json()
428         self.assertIn(
429             {"error-type": "application", "error-tag": "data-missing",
430                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
431             res['errors']['error'])
432
433     def test_20_service_path_delete_xpdr_check(self):
434         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
435                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
436                "interface/XPDR1-NETWORK1-ODU"
437                .format(self.restconf_baseurl))
438         headers = {'content-type': 'application/json'}
439         response = requests.request(
440             "GET", url, headers=headers, auth=('admin', 'admin'))
441         self.assertEqual(response.status_code, requests.codes.not_found)
442         res = response.json()
443         self.assertIn(
444             {"error-type": "application", "error-tag": "data-missing",
445                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
446             res['errors']['error'])
447
448     def test_21_service_path_delete_xpdr_check(self):
449         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
450                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
451                "interface/XPDR1-CLIENT1-ETHERNET"
452                .format(self.restconf_baseurl))
453         headers = {'content-type': 'application/json'}
454         response = requests.request(
455             "GET", url, headers=headers, auth=('admin', 'admin'))
456         self.assertEqual(response.status_code, requests.codes.not_found)
457         res = response.json()
458         self.assertIn(
459             {"error-type": "application", "error-tag": "data-missing",
460                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
461             res['errors']['error'])
462
463     def test_22_service_path_delete_xpdr_check(self):
464         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
465                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
466                "circuit-packs/1%2F0%2F1-PLUG-NET"
467                .format(self.restconf_baseurl))
468         headers = {'content-type': 'application/json'}
469         response = requests.request(
470             "GET", url, headers=headers, auth=('admin', 'admin'))
471         self.assertEqual(response.status_code, requests.codes.ok)
472         res = response.json()
473         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
474
475     def test_23_rdm_device_disconnected(self):
476         url = ("{}/config/network-topology:"
477                "network-topology/topology/topology-netconf/node/ROADMA01"
478                .format(self.restconf_baseurl))
479         headers = {'content-type': 'application/json'}
480         response = requests.request(
481             "DELETE", url, headers=headers,
482             auth=('admin', 'admin'))
483         self.assertEqual(response.status_code, requests.codes.ok)
484         time.sleep(20)
485
486     def test_24_xpdr_device_disconnected(self):
487         url = ("{}/config/network-topology:"
488                "network-topology/topology/topology-netconf/node/XPDRA01"
489                .format(self.restconf_baseurl))
490         headers = {'content-type': 'application/json'}
491         response = requests.request(
492             "DELETE", url, headers=headers,
493             auth=('admin', 'admin'))
494         self.assertEqual(response.status_code, requests.codes.ok)
495         time.sleep(20)
496
497
498 if __name__ == "__main__":
499     unittest.main(verbosity=2, failfast=True)