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