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
24 class TransportPCERendererTesting(unittest.TestCase):
26 honeynode_process1 = None
27 honeynode_process2 = None
29 restconf_baseurl = "http://localhost:8181/restconf"
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"],
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"],
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))
61 cls.__start_honeynode1()
63 cls.__start_honeynode2()
69 def tearDownClass(cls):
70 for child in psutil.Process(cls.odl_process.pid).children():
71 child.send_signal(signal.SIGINT)
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)
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)
83 cls.honeynode_process2.send_signal(signal.SIGINT)
84 cls.honeynode_process2.wait()
87 print ("execution of {}".format(self.id().split(".")[-1]))
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))
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)
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))
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)
128 def test_03_rdm_portmapping(self):
129 url = ("{}/config/transportpce-portmapping:network/"
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()
138 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
139 'logical-connection-point': 'DEG1-TTP-TXRX'},
140 res['nodes'][0]['mapping'])
142 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
143 'logical-connection-point': 'SRG1-PP7-TXRX'},
144 res['nodes'][0]['mapping'])
146 def test_04_xpdr_portmapping(self):
147 url = ("{}/config/transportpce-portmapping:network/"
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()
156 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
157 'logical-connection-point': 'XPDR1-NETWORK1'},
158 res['nodes'][0]['mapping'])
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'])
165 def test_05_service_path_create(self):
166 url = "{}/operations/transportpce-device-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/transportpce-device-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', 'success': True}})
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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()
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'])
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'])
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)
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)
507 if __name__ == "__main__":
508 unittest.main(verbosity=2, failfast=True)