46b957419591c339e1cad6f5b0a1b14a58603357
[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               'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7aa'},
143              res['nodes'][0]['mapping'])
144         self.assertIn(
145              {'supporting-port': 'C1',
146               'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
147               'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
148               'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
149               'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae5'},
150              res['nodes'][0]['mapping'])
151
152     def test_05_service_path_create(self):
153         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
154         data = {"renderer:input": {
155             "renderer:service-name": "service_test",
156             "renderer:wave-number": "7",
157             "renderer:modulation-format": "qpsk",
158             "renderer:operation": "create",
159             "renderer:nodes": [
160                 {"renderer:node-id": "ROADMA01",
161                  "renderer:src-tp": "SRG1-PP7-TXRX",
162                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
163                 {"renderer:node-id": "XPDRA01",
164                  "renderer:src-tp": "XPDR1-CLIENT1",
165                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
166         headers = {'content-type': 'application/json'}
167         response = requests.request(
168             "POST", url, data=json.dumps(data),
169             headers=headers, auth=('admin', 'admin'))
170         self.assertEqual(response.status_code, requests.codes.ok)
171         res = response.json()
172         self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
173
174     def test_06_service_path_create_rdm_check(self):
175         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
176                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
177                "interface/DEG1-TTP-TXRX-7"
178                .format(self.restconf_baseurl))
179         headers = {'content-type': 'application/json'}
180         response = requests.request(
181             "GET", url, headers=headers, auth=('admin', 'admin'))
182         self.assertEqual(response.status_code, requests.codes.ok)
183         res = response.json()
184         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-7', 'administrative-state': 'inService',
185                                        'supporting-circuit-pack-name': '2/0',
186                                        'type': 'org-openroadm-interfaces:opticalChannel',
187                                        'supporting-port': 'L1'}, res['interface'][0])
188         self.assertDictEqual(
189             {'wavelength-number': 7},
190             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
191
192     def test_07_service_path_create_rdm_check(self):
193         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
194                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
195                "interface/SRG1-PP7-TXRX-7"
196                .format(self.restconf_baseurl))
197         headers = {'content-type': 'application/json'}
198         response = requests.request(
199             "GET", url, headers=headers, auth=('admin', 'admin'))
200         self.assertEqual(response.status_code, requests.codes.ok)
201         res = response.json()
202         self.assertDictContainsSubset(
203             {'name': 'SRG1-PP7-TXRX-7', 'administrative-state': 'inService',
204              'supporting-circuit-pack-name': '4/0',
205              'type': 'org-openroadm-interfaces:opticalChannel',
206              'supporting-port': 'C7'},
207             res['interface'][0])
208         self.assertDictEqual(
209             {'wavelength-number': 7},
210             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
211
212     def test_08_service_path_create_rdm_check(self):
213         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
214                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
215                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
216                .format(self.restconf_baseurl))
217         headers = {'content-type': 'application/json'}
218         response = requests.request(
219             "GET", url, headers=headers, auth=('admin', 'admin'))
220         self.assertEqual(response.status_code, requests.codes.ok)
221         res = response.json()
222         self.assertDictContainsSubset(
223             {'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
224              'wavelength-number': 7,
225              'opticalControlMode': 'off'},
226             res['roadm-connections'][0])
227         self.assertDictEqual(
228             {'src-if': 'SRG1-PP7-TXRX-7'},
229             res['roadm-connections'][0]['source'])
230         self.assertDictEqual(
231             {'dst-if': 'DEG1-TTP-TXRX-7'},
232             res['roadm-connections'][0]['destination'])
233
234     def test_09_service_path_create_xpdr_check(self):
235         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
236                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
237                "interface/XPDR1-NETWORK1-7"
238                .format(self.restconf_baseurl))
239         headers = {'content-type': 'application/json'}
240         response = requests.request(
241             "GET", url, headers=headers, auth=('admin', 'admin'))
242         self.assertEqual(response.status_code, requests.codes.ok)
243         res = response.json()
244         self.assertDictContainsSubset(
245             {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
246              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
247              'type': 'org-openroadm-interfaces:opticalChannel',
248              'supporting-port': '1'},
249             res['interface'][0])
250         self.assertDictEqual(
251             {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
252              u'transmit-power': -5,
253              u'wavelength-number': 7,
254              u'modulation-format': u'dp-qpsk'},
255             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
256
257     def test_10_service_path_create_xpdr_check(self):
258         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
259                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
260                "interface/XPDR1-NETWORK1-OTU"
261                .format(self.restconf_baseurl))
262         headers = {'content-type': 'application/json'}
263         response = requests.request(
264             "GET", url, headers=headers, auth=('admin', 'admin'))
265         self.assertEqual(response.status_code, requests.codes.ok)
266         res = response.json()
267         self.assertDictContainsSubset(
268             {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
269              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
270              'type': 'org-openroadm-interfaces:otnOtu',
271              'supporting-port': '1',
272              'supporting-interface': 'XPDR1-NETWORK1-7'},
273             res['interface'][0])
274         self.assertDictEqual(
275             {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
276              u'fec': u'scfec'},
277             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
278
279     def test_11_service_path_create_xpdr_check(self):
280         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
281                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
282                "interface/XPDR1-NETWORK1-ODU"
283                .format(self.restconf_baseurl))
284         headers = {'content-type': 'application/json'}
285         response = requests.request(
286             "GET", url, headers=headers, auth=('admin', 'admin'))
287         self.assertEqual(response.status_code, requests.codes.ok)
288         res = response.json()
289         self.assertDictContainsSubset(
290             {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
291              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
292              'type': 'org-openroadm-interfaces:otnOdu',
293              'supporting-port': '1',
294              'supporting-interface': 'XPDR1-NETWORK1-OTU'},
295             res['interface'][0])
296         self.assertDictContainsSubset(
297             {'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
298              u'monitoring-mode': u'terminated'},
299             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
300         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
301                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
302
303     def test_12_service_path_create_xpdr_check(self):
304         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
305                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
306                "interface/XPDR1-CLIENT1-ETHERNET"
307                .format(self.restconf_baseurl))
308         headers = {'content-type': 'application/json'}
309         response = requests.request(
310             "GET", url, headers=headers, auth=('admin', 'admin'))
311         self.assertEqual(response.status_code, requests.codes.ok)
312         res = response.json()
313         self.assertDictContainsSubset(
314             {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
315              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
316              'type': 'org-openroadm-interfaces:ethernetCsmacd',
317              'supporting-port': 'C1'},
318             res['interface'][0])
319         self.assertDictEqual(
320             {'speed': 100000,
321              'mtu': 9000,
322              'auto-negotiation': 'enabled',
323              'duplex': 'full',
324              'fec': 'off'},
325             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
326
327     def test_13_service_path_create_xpdr_check(self):
328         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
329                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
330                "circuit-packs/1%2F0%2F1-PLUG-NET"
331                .format(self.restconf_baseurl))
332         headers = {'content-type': 'application/json'}
333         response = requests.request(
334             "GET", url, headers=headers, auth=('admin', 'admin'))
335         self.assertEqual(response.status_code, requests.codes.ok)
336         res = response.json()
337         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
338
339     def test_14_service_path_delete(self):
340         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
341         data = {"renderer:input": {
342             "renderer:service-name": "service_test",
343             "renderer:wave-number": "7",
344             "renderer:operation": "delete",
345             "renderer:nodes": [
346                 {"renderer:node-id": "ROADMA01",
347                  "renderer:src-tp": "SRG1-PP7-TXRX",
348                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
349                 {"renderer:node-id": "XPDRA01",
350                  "renderer:src-tp": "XPDR1-CLIENT1",
351                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
352         headers = {'content-type': 'application/json'}
353         response = requests.request(
354             "POST", url, data=json.dumps(data),
355             headers=headers, auth=('admin', 'admin'))
356         self.assertEqual(response.status_code, requests.codes.ok)
357         self.assertEqual(response.json(), {
358             'output': {'result': 'Request processed', 'success': True}})
359
360     def test_15_service_path_delete_rdm_check(self):
361         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
362                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
363                "interface/DEG1-TTP-TXRX-7"
364                .format(self.restconf_baseurl))
365         headers = {'content-type': 'application/json'}
366         response = requests.request(
367             "GET", url, headers=headers, auth=('admin', 'admin'))
368         self.assertEqual(response.status_code, requests.codes.not_found)
369         res = response.json()
370         self.assertIn(
371             {"error-type": "application", "error-tag": "data-missing",
372              "error-message": "Request could not be completed because the relevant data model content does not exist"},
373             res['errors']['error'])
374
375     def test_16_service_path_delete_rdm_check(self):
376         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
377                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
378                "interface/SRG1-PP7-TXRX-7"
379                .format(self.restconf_baseurl))
380         headers = {'content-type': 'application/json'}
381         response = requests.request(
382             "GET", url, headers=headers, auth=('admin', 'admin'))
383         self.assertEqual(response.status_code, requests.codes.not_found)
384         res = response.json()
385         self.assertIn(
386             {"error-type": "application", "error-tag": "data-missing",
387                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
388             res['errors']['error'])
389
390     def test_17_service_path_delete_rdm_check(self):
391         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
392                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
393                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
394                .format(self.restconf_baseurl))
395         headers = {'content-type': 'application/json'}
396         response = requests.request(
397             "GET", url, headers=headers, auth=('admin', 'admin'))
398         self.assertEqual(response.status_code, requests.codes.not_found)
399         res = response.json()
400         self.assertIn(
401             {"error-type": "application", "error-tag": "data-missing",
402                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
403             res['errors']['error'])
404
405     def test_18_service_path_delete_xpdr_check(self):
406         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
407                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
408                "interface/XPDR1-NETWORK1-7"
409                .format(self.restconf_baseurl))
410         headers = {'content-type': 'application/json'}
411         response = requests.request(
412             "GET", url, headers=headers, auth=('admin', 'admin'))
413         self.assertEqual(response.status_code, requests.codes.not_found)
414         res = response.json()
415         self.assertIn(
416             {"error-type": "application", "error-tag": "data-missing",
417                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
418             res['errors']['error'])
419
420     def test_19_service_path_delete_xpdr_check(self):
421         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
422                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
423                "interface/XPDR1-NETWORK1-OTU"
424                .format(self.restconf_baseurl))
425         headers = {'content-type': 'application/json'}
426         response = requests.request(
427             "GET", url, headers=headers, auth=('admin', 'admin'))
428         self.assertEqual(response.status_code, requests.codes.not_found)
429         res = response.json()
430         self.assertIn(
431             {"error-type": "application", "error-tag": "data-missing",
432                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
433             res['errors']['error'])
434
435     def test_20_service_path_delete_xpdr_check(self):
436         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
437                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
438                "interface/XPDR1-NETWORK1-ODU"
439                .format(self.restconf_baseurl))
440         headers = {'content-type': 'application/json'}
441         response = requests.request(
442             "GET", url, headers=headers, auth=('admin', 'admin'))
443         self.assertEqual(response.status_code, requests.codes.not_found)
444         res = response.json()
445         self.assertIn(
446             {"error-type": "application", "error-tag": "data-missing",
447                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
448             res['errors']['error'])
449
450     def test_21_service_path_delete_xpdr_check(self):
451         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
452                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
453                "interface/XPDR1-CLIENT1-ETHERNET"
454                .format(self.restconf_baseurl))
455         headers = {'content-type': 'application/json'}
456         response = requests.request(
457             "GET", url, headers=headers, auth=('admin', 'admin'))
458         self.assertEqual(response.status_code, requests.codes.not_found)
459         res = response.json()
460         self.assertIn(
461             {"error-type": "application", "error-tag": "data-missing",
462                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
463             res['errors']['error'])
464
465     def test_22_service_path_delete_xpdr_check(self):
466         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
467                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
468                "circuit-packs/1%2F0%2F1-PLUG-NET"
469                .format(self.restconf_baseurl))
470         headers = {'content-type': 'application/json'}
471         response = requests.request(
472             "GET", url, headers=headers, auth=('admin', 'admin'))
473         self.assertEqual(response.status_code, requests.codes.ok)
474         res = response.json()
475         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
476
477     def test_23_rdm_device_disconnected(self):
478         url = ("{}/config/network-topology:"
479                "network-topology/topology/topology-netconf/node/ROADMA01"
480                .format(self.restconf_baseurl))
481         headers = {'content-type': 'application/json'}
482         response = requests.request(
483             "DELETE", url, headers=headers,
484             auth=('admin', 'admin'))
485         self.assertEqual(response.status_code, requests.codes.ok)
486         time.sleep(20)
487
488     def test_24_xpdr_device_disconnected(self):
489         url = ("{}/config/network-topology:"
490                "network-topology/topology/topology-netconf/node/XPDRA01"
491                .format(self.restconf_baseurl))
492         headers = {'content-type': 'application/json'}
493         response = requests.request(
494             "DELETE", url, headers=headers,
495             auth=('admin', 'admin'))
496         self.assertEqual(response.status_code, requests.codes.ok)
497         time.sleep(20)
498
499
500 if __name__ == "__main__":
501     unittest.main(verbosity=2, failfast=True)