3 #############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
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 #############################################################################
21 from unittest.result import failfast
23 class TransportPCERendererTesting(unittest.TestCase):
25 honeynode_process1 = None
26 honeynode_process2 = None
28 restconf_baseurl = "http://localhost:8181/restconf"
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"],
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"],
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))
60 cls.__start_honeynode1()
62 cls.__start_honeynode2()
68 def tearDownClass(cls):
69 for child in psutil.Process(cls.odl_process.pid).children():
70 child.send_signal(signal.SIGINT)
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)
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)
82 cls.honeynode_process2.send_signal(signal.SIGINT)
83 cls.honeynode_process2.wait()
86 print ("execution of {}".format(self.id().split(".")[-1]))
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))
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)
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))
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)
127 def test_03_rdm_portmapping(self):
128 url = ("{}/config/portmapping:network/"
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()
137 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
138 'logical-connection-point': 'DEG1-TTP-TXRX'},
139 res['nodes'][0]['mapping'])
141 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
142 'logical-connection-point': 'SRG1-PP7-TXRX'},
143 res['nodes'][0]['mapping'])
145 def test_04_xpdr_portmapping(self):
146 url = ("{}/config/portmapping:network/"
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()
155 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
156 'logical-connection-point': 'XPDR1-NETWORK1'},
157 res['nodes'][0]['mapping'])
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'])
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",
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"])
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'])
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'},
221 self.assertDictEqual(
222 {'wavelength-number': 7},
223 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
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'])
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'},
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'])
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'},
287 self.assertDictEqual(
288 {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
290 res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
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'},
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'])
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'},
332 self.assertDictEqual(
335 'auto-negotiation': 'enabled',
338 res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
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"])
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",
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'}})
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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'])
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)
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)
507 if __name__ == "__main__":
508 unittest.main(verbosity=2, failfast=True)