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