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
25 class TransportPCERendererTesting(unittest.TestCase):
27 honeynode_process1 = None
28 honeynode_process2 = None
30 restconf_baseurl = "http://localhost:8181/restconf"
32 # START_IGNORE_XTESTING
36 print("starting honeynode1...")
37 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
40 print("starting honeynode2...")
41 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
44 print("starting opendaylight...")
45 cls.odl_process = test_utils.start_tpce()
47 print("opendaylight started")
50 def tearDownClass(cls):
51 for child in psutil.Process(cls.odl_process.pid).children():
52 child.send_signal(signal.SIGINT)
54 cls.odl_process.send_signal(signal.SIGINT)
55 cls.odl_process.wait()
56 for child in psutil.Process(cls.honeynode_process1.pid).children():
57 child.send_signal(signal.SIGINT)
59 cls.honeynode_process1.send_signal(signal.SIGINT)
60 cls.honeynode_process1.wait()
61 for child in psutil.Process(cls.honeynode_process2.pid).children():
62 child.send_signal(signal.SIGINT)
64 cls.honeynode_process2.send_signal(signal.SIGINT)
65 cls.honeynode_process2.wait()
68 print("execution of {}".format(self.id().split(".")[-1]))
73 def test_01_rdm_device_connected(self):
74 url = ("{}/config/network-topology:"
75 "network-topology/topology/topology-netconf/node/ROADMA01"
76 .format(self.restconf_baseurl))
78 "node-id": "ROADMA01",
79 "netconf-node-topology:username": "admin",
80 "netconf-node-topology:password": "admin",
81 "netconf-node-topology:host": "127.0.0.1",
82 "netconf-node-topology:port": "17831",
83 "netconf-node-topology:tcp-only": "false",
84 "netconf-node-topology:pass-through": {}}]}
85 headers = {'content-type': 'application/json'}
86 response = requests.request(
87 "PUT", url, data=json.dumps(data), headers=headers,
88 auth=('admin', 'admin'))
89 self.assertEqual(response.status_code, requests.codes.created)
92 def test_02_xpdr_device_connected(self):
93 url = ("{}/config/network-topology:"
94 "network-topology/topology/topology-netconf/node/XPDRA01"
95 .format(self.restconf_baseurl))
98 "netconf-node-topology:username": "admin",
99 "netconf-node-topology:password": "admin",
100 "netconf-node-topology:host": "127.0.0.1",
101 "netconf-node-topology:port": "17830",
102 "netconf-node-topology:tcp-only": "false",
103 "netconf-node-topology:pass-through": {}}]}
104 headers = {'content-type': 'application/json'}
105 response = requests.request(
106 "PUT", url, data=json.dumps(data), headers=headers,
107 auth=('admin', 'admin'))
108 self.assertEqual(response.status_code, requests.codes.created)
111 def test_03_rdm_portmapping(self):
112 url = ("{}/config/transportpce-portmapping:network/"
114 .format(self.restconf_baseurl))
115 headers = {'content-type': 'application/json'}
116 response = requests.request(
117 "GET", url, headers=headers, auth=('admin', 'admin'))
118 self.assertEqual(response.status_code, requests.codes.ok)
119 res = response.json()
121 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
122 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
123 res['nodes'][0]['mapping'])
125 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
126 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
127 res['nodes'][0]['mapping'])
129 def test_04_xpdr_portmapping(self):
130 url = ("{}/config/transportpce-portmapping:network/"
132 .format(self.restconf_baseurl))
133 headers = {'content-type': 'application/json'}
134 response = requests.request(
135 "GET", url, headers=headers, auth=('admin', 'admin'))
136 self.assertEqual(response.status_code, requests.codes.ok)
137 res = response.json()
139 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
140 'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
141 'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
142 'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7aa'},
143 res['nodes'][0]['mapping'])
145 {'supporting-port': 'C1',
146 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
147 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
148 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
149 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae5'},
150 res['nodes'][0]['mapping'])
152 def test_05_service_path_create(self):
153 url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
154 data = {"renderer:input": {
155 "renderer:service-name": "service_test",
156 "renderer:wave-number": "7",
157 "renderer:modulation-format": "qpsk",
158 "renderer:operation": "create",
160 {"renderer:node-id": "ROADMA01",
161 "renderer:src-tp": "SRG1-PP7-TXRX",
162 "renderer:dest-tp": "DEG1-TTP-TXRX"},
163 {"renderer:node-id": "XPDRA01",
164 "renderer:src-tp": "XPDR1-CLIENT1",
165 "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
166 headers = {'content-type': 'application/json'}
167 response = requests.request(
168 "POST", url, data=json.dumps(data),
169 headers=headers, auth=('admin', 'admin'))
170 self.assertEqual(response.status_code, requests.codes.ok)
171 res = response.json()
172 self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
174 def test_06_service_path_create_rdm_check(self):
175 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
176 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
177 "interface/DEG1-TTP-TXRX-7"
178 .format(self.restconf_baseurl))
179 headers = {'content-type': 'application/json'}
180 response = requests.request(
181 "GET", url, headers=headers, auth=('admin', 'admin'))
182 self.assertEqual(response.status_code, requests.codes.ok)
183 res = response.json()
184 self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-7', 'administrative-state': 'inService',
185 'supporting-circuit-pack-name': '2/0',
186 'type': 'org-openroadm-interfaces:opticalChannel',
187 'supporting-port': 'L1'}, res['interface'][0])
188 self.assertDictEqual(
189 {'wavelength-number': 7},
190 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
192 def test_07_service_path_create_rdm_check(self):
193 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
194 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
195 "interface/SRG1-PP7-TXRX-7"
196 .format(self.restconf_baseurl))
197 headers = {'content-type': 'application/json'}
198 response = requests.request(
199 "GET", url, headers=headers, auth=('admin', 'admin'))
200 self.assertEqual(response.status_code, requests.codes.ok)
201 res = response.json()
202 self.assertDictContainsSubset(
203 {'name': 'SRG1-PP7-TXRX-7', 'administrative-state': 'inService',
204 'supporting-circuit-pack-name': '4/0',
205 'type': 'org-openroadm-interfaces:opticalChannel',
206 'supporting-port': 'C7'},
208 self.assertDictEqual(
209 {'wavelength-number': 7},
210 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
212 def test_08_service_path_create_rdm_check(self):
213 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
214 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
215 "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
216 .format(self.restconf_baseurl))
217 headers = {'content-type': 'application/json'}
218 response = requests.request(
219 "GET", url, headers=headers, auth=('admin', 'admin'))
220 self.assertEqual(response.status_code, requests.codes.ok)
221 res = response.json()
222 self.assertDictContainsSubset(
223 {'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
224 'wavelength-number': 7,
225 'opticalControlMode': 'off'},
226 res['roadm-connections'][0])
227 self.assertDictEqual(
228 {'src-if': 'SRG1-PP7-TXRX-7'},
229 res['roadm-connections'][0]['source'])
230 self.assertDictEqual(
231 {'dst-if': 'DEG1-TTP-TXRX-7'},
232 res['roadm-connections'][0]['destination'])
234 def test_09_service_path_create_xpdr_check(self):
235 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
236 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
237 "interface/XPDR1-NETWORK1-7"
238 .format(self.restconf_baseurl))
239 headers = {'content-type': 'application/json'}
240 response = requests.request(
241 "GET", url, headers=headers, auth=('admin', 'admin'))
242 self.assertEqual(response.status_code, requests.codes.ok)
243 res = response.json()
244 self.assertDictContainsSubset(
245 {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
246 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
247 'type': 'org-openroadm-interfaces:opticalChannel',
248 'supporting-port': '1'},
250 self.assertDictEqual(
251 {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
252 u'transmit-power': -5,
253 u'wavelength-number': 7,
254 u'modulation-format': u'dp-qpsk'},
255 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
257 def test_10_service_path_create_xpdr_check(self):
258 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
259 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
260 "interface/XPDR1-NETWORK1-OTU"
261 .format(self.restconf_baseurl))
262 headers = {'content-type': 'application/json'}
263 response = requests.request(
264 "GET", url, headers=headers, auth=('admin', 'admin'))
265 self.assertEqual(response.status_code, requests.codes.ok)
266 res = response.json()
267 self.assertDictContainsSubset(
268 {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
269 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
270 'type': 'org-openroadm-interfaces:otnOtu',
271 'supporting-port': '1',
272 'supporting-interface': 'XPDR1-NETWORK1-7'},
274 self.assertDictEqual(
275 {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
277 res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
279 def test_11_service_path_create_xpdr_check(self):
280 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
281 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
282 "interface/XPDR1-NETWORK1-ODU"
283 .format(self.restconf_baseurl))
284 headers = {'content-type': 'application/json'}
285 response = requests.request(
286 "GET", url, headers=headers, auth=('admin', 'admin'))
287 self.assertEqual(response.status_code, requests.codes.ok)
288 res = response.json()
289 self.assertDictContainsSubset(
290 {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
291 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
292 'type': 'org-openroadm-interfaces:otnOdu',
293 'supporting-port': '1',
294 'supporting-interface': 'XPDR1-NETWORK1-OTU'},
296 self.assertDictContainsSubset(
297 {'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
298 u'monitoring-mode': u'terminated'},
299 res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
300 self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
301 res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
303 def test_12_service_path_create_xpdr_check(self):
304 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
305 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
306 "interface/XPDR1-CLIENT1-ETHERNET"
307 .format(self.restconf_baseurl))
308 headers = {'content-type': 'application/json'}
309 response = requests.request(
310 "GET", url, headers=headers, auth=('admin', 'admin'))
311 self.assertEqual(response.status_code, requests.codes.ok)
312 res = response.json()
313 self.assertDictContainsSubset(
314 {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
315 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
316 'type': 'org-openroadm-interfaces:ethernetCsmacd',
317 'supporting-port': 'C1'},
319 self.assertDictEqual(
322 'auto-negotiation': 'enabled',
325 res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
327 def test_13_service_path_create_xpdr_check(self):
328 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
329 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
330 "circuit-packs/1%2F0%2F1-PLUG-NET"
331 .format(self.restconf_baseurl))
332 headers = {'content-type': 'application/json'}
333 response = requests.request(
334 "GET", url, headers=headers, auth=('admin', 'admin'))
335 self.assertEqual(response.status_code, requests.codes.ok)
336 res = response.json()
337 self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
339 def test_14_service_path_delete(self):
340 url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
341 data = {"renderer:input": {
342 "renderer:service-name": "service_test",
343 "renderer:wave-number": "7",
344 "renderer:operation": "delete",
346 {"renderer:node-id": "ROADMA01",
347 "renderer:src-tp": "SRG1-PP7-TXRX",
348 "renderer:dest-tp": "DEG1-TTP-TXRX"},
349 {"renderer:node-id": "XPDRA01",
350 "renderer:src-tp": "XPDR1-CLIENT1",
351 "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
352 headers = {'content-type': 'application/json'}
353 response = requests.request(
354 "POST", url, data=json.dumps(data),
355 headers=headers, auth=('admin', 'admin'))
356 self.assertEqual(response.status_code, requests.codes.ok)
357 self.assertEqual(response.json(), {
358 'output': {'result': 'Request processed', 'success': True}})
360 def test_15_service_path_delete_rdm_check(self):
361 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
362 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
363 "interface/DEG1-TTP-TXRX-7"
364 .format(self.restconf_baseurl))
365 headers = {'content-type': 'application/json'}
366 response = requests.request(
367 "GET", url, headers=headers, auth=('admin', 'admin'))
368 self.assertEqual(response.status_code, requests.codes.not_found)
369 res = response.json()
371 {"error-type": "application", "error-tag": "data-missing",
372 "error-message": "Request could not be completed because the relevant data model content does not exist"},
373 res['errors']['error'])
375 def test_16_service_path_delete_rdm_check(self):
376 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
377 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
378 "interface/SRG1-PP7-TXRX-7"
379 .format(self.restconf_baseurl))
380 headers = {'content-type': 'application/json'}
381 response = requests.request(
382 "GET", url, headers=headers, auth=('admin', 'admin'))
383 self.assertEqual(response.status_code, requests.codes.not_found)
384 res = response.json()
386 {"error-type": "application", "error-tag": "data-missing",
387 "error-message": "Request could not be completed because the relevant data model content does not exist"},
388 res['errors']['error'])
390 def test_17_service_path_delete_rdm_check(self):
391 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
392 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
393 "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
394 .format(self.restconf_baseurl))
395 headers = {'content-type': 'application/json'}
396 response = requests.request(
397 "GET", url, headers=headers, auth=('admin', 'admin'))
398 self.assertEqual(response.status_code, requests.codes.not_found)
399 res = response.json()
401 {"error-type": "application", "error-tag": "data-missing",
402 "error-message": "Request could not be completed because the relevant data model content does not exist"},
403 res['errors']['error'])
405 def test_18_service_path_delete_xpdr_check(self):
406 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
407 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
408 "interface/XPDR1-NETWORK1-7"
409 .format(self.restconf_baseurl))
410 headers = {'content-type': 'application/json'}
411 response = requests.request(
412 "GET", url, headers=headers, auth=('admin', 'admin'))
413 self.assertEqual(response.status_code, requests.codes.not_found)
414 res = response.json()
416 {"error-type": "application", "error-tag": "data-missing",
417 "error-message": "Request could not be completed because the relevant data model content does not exist"},
418 res['errors']['error'])
420 def test_19_service_path_delete_xpdr_check(self):
421 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
422 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
423 "interface/XPDR1-NETWORK1-OTU"
424 .format(self.restconf_baseurl))
425 headers = {'content-type': 'application/json'}
426 response = requests.request(
427 "GET", url, headers=headers, auth=('admin', 'admin'))
428 self.assertEqual(response.status_code, requests.codes.not_found)
429 res = response.json()
431 {"error-type": "application", "error-tag": "data-missing",
432 "error-message": "Request could not be completed because the relevant data model content does not exist"},
433 res['errors']['error'])
435 def test_20_service_path_delete_xpdr_check(self):
436 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
437 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
438 "interface/XPDR1-NETWORK1-ODU"
439 .format(self.restconf_baseurl))
440 headers = {'content-type': 'application/json'}
441 response = requests.request(
442 "GET", url, headers=headers, auth=('admin', 'admin'))
443 self.assertEqual(response.status_code, requests.codes.not_found)
444 res = response.json()
446 {"error-type": "application", "error-tag": "data-missing",
447 "error-message": "Request could not be completed because the relevant data model content does not exist"},
448 res['errors']['error'])
450 def test_21_service_path_delete_xpdr_check(self):
451 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
452 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
453 "interface/XPDR1-CLIENT1-ETHERNET"
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",
462 "error-message": "Request could not be completed because the relevant data model content does not exist"},
463 res['errors']['error'])
465 def test_22_service_path_delete_xpdr_check(self):
466 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
467 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
468 "circuit-packs/1%2F0%2F1-PLUG-NET"
469 .format(self.restconf_baseurl))
470 headers = {'content-type': 'application/json'}
471 response = requests.request(
472 "GET", url, headers=headers, auth=('admin', 'admin'))
473 self.assertEqual(response.status_code, requests.codes.ok)
474 res = response.json()
475 self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
477 def test_23_rdm_device_disconnected(self):
478 url = ("{}/config/network-topology:"
479 "network-topology/topology/topology-netconf/node/ROADMA01"
480 .format(self.restconf_baseurl))
481 headers = {'content-type': 'application/json'}
482 response = requests.request(
483 "DELETE", url, headers=headers,
484 auth=('admin', 'admin'))
485 self.assertEqual(response.status_code, requests.codes.ok)
488 def test_24_xpdr_device_disconnected(self):
489 url = ("{}/config/network-topology:"
490 "network-topology/topology/topology-netconf/node/XPDRA01"
491 .format(self.restconf_baseurl))
492 headers = {'content-type': 'application/json'}
493 response = requests.request(
494 "DELETE", url, headers=headers,
495 auth=('admin', 'admin'))
496 self.assertEqual(response.status_code, requests.codes.ok)
500 if __name__ == "__main__":
501 unittest.main(verbosity=2, failfast=True)