fix Honeynode issues with fluorine
[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
24 class TransportPCERendererTesting(unittest.TestCase):
25
26     honeynode_process1 = None
27     honeynode_process2 = None
28     odl_process = None
29     restconf_baseurl = "http://localhost:8181/restconf"
30
31     @classmethod
32     def __start_honeynode1(cls):
33         executable = ("./honeynode/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
34                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
35         if os.path.isfile(executable):
36             with open('honeynode1.log', 'w') as outfile:
37                 cls.honeynode_process1 = subprocess.Popen(
38                     [executable, "17830", "sample_configs/ord_2.1/oper-ROADMA.xml"],
39                     stdout=outfile)
40
41     @classmethod
42     def __start_honeynode2(cls):
43         executable = ("./honeynode/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
44                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
45         if os.path.isfile(executable):
46             with open('honeynode2.log', 'w') as outfile:
47                 cls.honeynode_process2 = subprocess.Popen(
48                     [executable, "17831", "sample_configs/ord_2.1/oper-XPDRA.xml"],
49                     stdout=outfile)
50
51     @classmethod
52     def __start_odl(cls):
53         executable = "../karaf/target/assembly/bin/karaf"
54         with open('odl.log', 'w') as outfile:
55             cls.odl_process = subprocess.Popen(
56                 ["bash", executable, "server"], stdout=outfile,
57                 stdin=open(os.devnull))
58
59     @classmethod
60     def setUpClass(cls):
61         cls.__start_honeynode1()
62         time.sleep(40)
63         cls.__start_honeynode2()
64         time.sleep(40)
65         cls.__start_odl()
66         time.sleep(60)
67
68     @classmethod
69     def tearDownClass(cls):
70         for child in psutil.Process(cls.odl_process.pid).children():
71             child.send_signal(signal.SIGINT)
72             child.wait()
73         cls.odl_process.send_signal(signal.SIGINT)
74         cls.odl_process.wait()
75         for child in psutil.Process(cls.honeynode_process1.pid).children():
76             child.send_signal(signal.SIGINT)
77             child.wait()
78         cls.honeynode_process1.send_signal(signal.SIGINT)
79         cls.honeynode_process1.wait()
80         for child in psutil.Process(cls.honeynode_process2.pid).children():
81             child.send_signal(signal.SIGINT)
82             child.wait()
83         cls.honeynode_process2.send_signal(signal.SIGINT)
84         cls.honeynode_process2.wait()
85
86     def setUp(self):
87         print ("execution of {}".format(self.id().split(".")[-1]))
88         time.sleep(10)
89
90     def test_01_rdm_device_connected(self):
91         url = ("{}/config/network-topology:"
92                "network-topology/topology/topology-netconf/node/ROADMA"
93                .format(self.restconf_baseurl))
94         data = {"node": [{
95              "node-id": "ROADMA",
96              "netconf-node-topology:username": "admin",
97              "netconf-node-topology:password": "admin",
98              "netconf-node-topology:host": "127.0.0.1",
99              "netconf-node-topology:port": "17830",
100              "netconf-node-topology:tcp-only": "false",
101              "netconf-node-topology:pass-through": {}}]}
102         headers = {'content-type': 'application/json'}
103         response = requests.request(
104              "PUT", url, data=json.dumps(data), headers=headers,
105               auth=('admin', 'admin'))
106         self.assertEqual(response.status_code, requests.codes.created)
107         time.sleep(20)
108
109     def test_02_xpdr_device_connected(self):
110         url = ("{}/config/network-topology:"
111                "network-topology/topology/topology-netconf/node/XPDRA"
112               .format(self.restconf_baseurl))
113         data = {"node": [{
114             "node-id": "XPDRA",
115             "netconf-node-topology:username": "admin",
116             "netconf-node-topology:password": "admin",
117             "netconf-node-topology:host": "127.0.0.1",
118             "netconf-node-topology:port": "17831",
119             "netconf-node-topology:tcp-only": "false",
120             "netconf-node-topology:pass-through": {}}]}
121         headers = {'content-type': 'application/json'}
122         response = requests.request(
123             "PUT", url, data=json.dumps(data), headers=headers,
124             auth=('admin', 'admin'))
125         self.assertEqual(response.status_code, requests.codes.created)
126         time.sleep(20)
127
128     def test_03_rdm_portmapping(self):
129         url = ("{}/config/portmapping:network/"
130                "nodes/ROADMA"
131                .format(self.restconf_baseurl))
132         headers = {'content-type': 'application/json'}
133         response = requests.request(
134              "GET", url, headers=headers, auth=('admin', 'admin'))
135         self.assertEqual(response.status_code, requests.codes.ok)
136         res = response.json()
137         self.assertIn(
138              {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
139               'logical-connection-point': 'DEG1-TTP-TXRX'},
140              res['nodes'][0]['mapping'])
141         self.assertIn(
142              {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
143               'logical-connection-point': 'SRG1-PP7-TXRX'},
144              res['nodes'][0]['mapping'])
145
146     def test_04_xpdr_portmapping(self):
147         url = ("{}/config/portmapping:network/"
148                "nodes/XPDRA"
149                .format(self.restconf_baseurl))
150         headers = {'content-type': 'application/json'}
151         response = requests.request(
152              "GET", url, headers=headers, auth=('admin', 'admin'))
153         self.assertEqual(response.status_code, requests.codes.ok)
154         res = response.json()
155         self.assertIn(
156              {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
157               'logical-connection-point': 'XPDR1-NETWORK1'},
158              res['nodes'][0]['mapping'])
159         self.assertIn(
160              {'supporting-port': 'C1',
161               'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
162               'logical-connection-point': 'XPDR1-CLIENT1'},
163              res['nodes'][0]['mapping'])
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', 'success': True}})
372
373     def test_15_service_path_delete_rdm_check(self):
374         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
375                 "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
376                 "interface/DEG1-TTP-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_16_service_path_delete_rdm_check(self):
389         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
390                "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
391                "interface/SRG1-PP7-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", "error-message":"Request could not be completed because the relevant data model content does not exist "},
400              res['errors']['error'])
401
402     def test_17_service_path_delete_rdm_check(self):
403         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
404                "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
405                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
406                .format(self.restconf_baseurl))
407         headers = {'content-type': 'application/json'}
408         response = requests.request(
409              "GET", url, headers=headers, auth=('admin', 'admin'))
410         self.assertEqual(response.status_code, requests.codes.not_found)
411         res = response.json()
412         self.assertIn(
413              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist "},
414              res['errors']['error'])
415
416     def test_18_service_path_delete_xpdr_check(self):
417         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
418                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
419                "interface/XPDR1-NETWORK1-7"
420                .format(self.restconf_baseurl))
421         headers = {'content-type': 'application/json'}
422         response = requests.request(
423              "GET", url, headers=headers, auth=('admin', 'admin'))
424         self.assertEqual(response.status_code, requests.codes.not_found)
425         res = response.json()
426         self.assertIn(
427              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist "},
428              res['errors']['error'])
429
430     def test_19_service_path_delete_xpdr_check(self):
431         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
432                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
433                "interface/XPDR1-NETWORK1-OTU"
434                .format(self.restconf_baseurl))
435         headers = {'content-type': 'application/json'}
436         response = requests.request(
437              "GET", url, headers=headers, auth=('admin', 'admin'))
438         self.assertEqual(response.status_code, requests.codes.not_found)
439         res = response.json()
440         self.assertIn(
441              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist "},
442              res['errors']['error'])
443
444     def test_20_service_path_delete_xpdr_check(self):
445         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
446                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
447                "interface/XPDR1-NETWORK1-ODU"
448                .format(self.restconf_baseurl))
449         headers = {'content-type': 'application/json'}
450         response = requests.request(
451              "GET", url, headers=headers, auth=('admin', 'admin'))
452         self.assertEqual(response.status_code, requests.codes.not_found)
453         res = response.json()
454         self.assertIn(
455              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist "},
456              res['errors']['error'])
457
458     def test_21_service_path_delete_xpdr_check(self):
459         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
460                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
461                "interface/XPDR1-CLIENT1-ETHERNET"
462                .format(self.restconf_baseurl))
463         headers = {'content-type': 'application/json'}
464         response = requests.request(
465              "GET", url, headers=headers, auth=('admin', 'admin'))
466         self.assertEqual(response.status_code, requests.codes.not_found)
467         res = response.json()
468         self.assertIn(
469              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist "},
470              res['errors']['error'])
471
472     def test_22_service_path_delete_xpdr_check(self):
473         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
474                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
475                "circuit-packs/1%2F0%2F1-PLUG-NET"
476                .format(self.restconf_baseurl))
477         headers = {'content-type': 'application/json'}
478         response = requests.request(
479             "GET", url, headers=headers, auth=('admin', 'admin'))
480         self.assertEqual(response.status_code, requests.codes.ok)
481         res = response.json()
482         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
483
484     def test_23_rdm_device_disconnected(self):
485         url = ("{}/config/network-topology:"
486                "network-topology/topology/topology-netconf/node/ROADMA"
487                .format(self.restconf_baseurl))
488         headers = {'content-type': 'application/json'}
489         response = requests.request(
490               "DELETE", url, headers=headers,
491               auth=('admin', 'admin'))
492         self.assertEqual(response.status_code, requests.codes.ok)
493         time.sleep(20)
494
495     def test_24_xpdr_device_disconnected(self):
496         url = ("{}/config/network-topology:"
497                 "network-topology/topology/topology-netconf/node/XPDRA"
498                .format(self.restconf_baseurl))
499         headers = {'content-type': 'application/json'}
500         response = requests.request(
501              "DELETE", url, headers=headers,
502              auth=('admin', 'admin'))
503         self.assertEqual(response.status_code, requests.codes.ok)
504         time.sleep(20)
505
506
507 if __name__ == "__main__":
508     unittest.main(verbosity=2, failfast=True)