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 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
185 self.assertDictEqual(
187 'name': 'DEG1-TTP-TXRX-7',
188 'administrative-state': 'inService',
189 'supporting-circuit-pack-name': '2/0',
190 'type': 'org-openroadm-interfaces:opticalChannel',
191 'supporting-port': 'L1'
192 }, **res['interface'][0]),
195 self.assertDictEqual(
196 {'wavelength-number': 7},
197 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
199 def test_07_service_path_create_rdm_check(self):
200 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
201 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
202 "interface/SRG1-PP7-TXRX-7"
203 .format(self.restconf_baseurl))
204 headers = {'content-type': 'application/json'}
205 response = requests.request(
206 "GET", url, headers=headers, auth=('admin', 'admin'))
207 self.assertEqual(response.status_code, requests.codes.ok)
208 res = response.json()
209 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
210 self.assertDictEqual(
212 'name': 'SRG1-PP7-TXRX-7',
213 'administrative-state': 'inService',
214 'supporting-circuit-pack-name': '4/0',
215 'type': 'org-openroadm-interfaces:opticalChannel',
216 'supporting-port': 'C7'
217 }, **res['interface'][0]),
220 self.assertDictEqual(
221 {'wavelength-number': 7},
222 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
224 def test_08_service_path_create_rdm_check(self):
225 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
226 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
227 "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
228 .format(self.restconf_baseurl))
229 headers = {'content-type': 'application/json'}
230 response = requests.request(
231 "GET", url, headers=headers, auth=('admin', 'admin'))
232 self.assertEqual(response.status_code, requests.codes.ok)
233 res = response.json()
234 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
235 self.assertDictEqual(
237 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
238 'wavelength-number': 7,
239 'opticalControlMode': 'off'
240 }, **res['roadm-connections'][0]),
241 res['roadm-connections'][0]
243 self.assertDictEqual(
244 {'src-if': 'SRG1-PP7-TXRX-7'},
245 res['roadm-connections'][0]['source'])
246 self.assertDictEqual(
247 {'dst-if': 'DEG1-TTP-TXRX-7'},
248 res['roadm-connections'][0]['destination'])
250 def test_09_service_path_create_xpdr_check(self):
251 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
252 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
253 "interface/XPDR1-NETWORK1-7"
254 .format(self.restconf_baseurl))
255 headers = {'content-type': 'application/json'}
256 response = requests.request(
257 "GET", url, headers=headers, auth=('admin', 'admin'))
258 self.assertEqual(response.status_code, requests.codes.ok)
259 res = response.json()
260 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
261 self.assertDictEqual(
263 'name': 'XPDR1-NETWORK1-7',
264 'administrative-state': 'inService',
265 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
266 'type': 'org-openroadm-interfaces:opticalChannel',
267 'supporting-port': '1'
268 }, **res['interface'][0]),
271 self.assertDictEqual(
272 {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
273 u'transmit-power': -5,
274 u'wavelength-number': 7,
275 u'modulation-format': u'dp-qpsk'},
276 res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
278 def test_10_service_path_create_xpdr_check(self):
279 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
280 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
281 "interface/XPDR1-NETWORK1-OTU"
282 .format(self.restconf_baseurl))
283 headers = {'content-type': 'application/json'}
284 response = requests.request(
285 "GET", url, headers=headers, auth=('admin', 'admin'))
286 self.assertEqual(response.status_code, requests.codes.ok)
287 res = response.json()
288 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
289 self.assertDictEqual(
291 'name': 'XPDR1-NETWORK1-OTU',
292 'administrative-state': 'inService',
293 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
294 'type': 'org-openroadm-interfaces:otnOtu',
295 'supporting-port': '1',
296 'supporting-interface': 'XPDR1-NETWORK1-7'
297 }, **res['interface'][0]),
300 self.assertDictEqual(
301 {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
303 res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
305 def test_11_service_path_create_xpdr_check(self):
306 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
307 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
308 "interface/XPDR1-NETWORK1-ODU"
309 .format(self.restconf_baseurl))
310 headers = {'content-type': 'application/json'}
311 response = requests.request(
312 "GET", url, headers=headers, auth=('admin', 'admin'))
313 self.assertEqual(response.status_code, requests.codes.ok)
314 res = response.json()
315 # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
316 self.assertDictEqual(
318 'name': 'XPDR1-NETWORK1-ODU',
319 'administrative-state': 'inService',
320 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
321 'type': 'org-openroadm-interfaces:otnOdu',
322 'supporting-port': '1',
323 'supporting-interface': 'XPDR1-NETWORK1-OTU'
324 }, **res['interface'][0]),
327 self.assertDictEqual(
329 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
330 u'monitoring-mode': u'terminated'
331 }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
332 res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
334 self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
335 res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
337 def test_12_service_path_create_xpdr_check(self):
338 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
339 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
340 "interface/XPDR1-CLIENT1-ETHERNET"
341 .format(self.restconf_baseurl))
342 headers = {'content-type': 'application/json'}
343 response = requests.request(
344 "GET", url, headers=headers, auth=('admin', 'admin'))
345 self.assertEqual(response.status_code, requests.codes.ok)
346 res = response.json()
347 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
348 self.assertDictEqual(
350 'name': 'XPDR1-CLIENT1-ETHERNET',
351 'administrative-state': 'inService',
352 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
353 'type': 'org-openroadm-interfaces:ethernetCsmacd',
354 'supporting-port': 'C1'
355 }, **res['interface'][0]),
358 self.assertDictEqual(
361 'auto-negotiation': 'enabled',
364 res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
366 def test_13_service_path_create_xpdr_check(self):
367 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
368 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
369 "circuit-packs/1%2F0%2F1-PLUG-NET"
370 .format(self.restconf_baseurl))
371 headers = {'content-type': 'application/json'}
372 response = requests.request(
373 "GET", url, headers=headers, auth=('admin', 'admin'))
374 self.assertEqual(response.status_code, requests.codes.ok)
375 res = response.json()
376 self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
378 def test_14_service_path_delete(self):
379 url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
380 data = {"renderer:input": {
381 "renderer:service-name": "service_test",
382 "renderer:wave-number": "7",
383 "renderer:operation": "delete",
385 {"renderer:node-id": "ROADMA01",
386 "renderer:src-tp": "SRG1-PP7-TXRX",
387 "renderer:dest-tp": "DEG1-TTP-TXRX"},
388 {"renderer:node-id": "XPDRA01",
389 "renderer:src-tp": "XPDR1-CLIENT1",
390 "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
391 headers = {'content-type': 'application/json'}
392 response = requests.request(
393 "POST", url, data=json.dumps(data),
394 headers=headers, auth=('admin', 'admin'))
395 self.assertEqual(response.status_code, requests.codes.ok)
396 self.assertEqual(response.json(), {
397 'output': {'result': 'Request processed', 'success': True}})
399 def test_15_service_path_delete_rdm_check(self):
400 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
401 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
402 "interface/DEG1-TTP-TXRX-7"
403 .format(self.restconf_baseurl))
404 headers = {'content-type': 'application/json'}
405 response = requests.request(
406 "GET", url, headers=headers, auth=('admin', 'admin'))
407 self.assertEqual(response.status_code, requests.codes.not_found)
408 res = response.json()
410 {"error-type": "application", "error-tag": "data-missing",
411 "error-message": "Request could not be completed because the relevant data model content does not exist"},
412 res['errors']['error'])
414 def test_16_service_path_delete_rdm_check(self):
415 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
416 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
417 "interface/SRG1-PP7-TXRX-7"
418 .format(self.restconf_baseurl))
419 headers = {'content-type': 'application/json'}
420 response = requests.request(
421 "GET", url, headers=headers, auth=('admin', 'admin'))
422 self.assertEqual(response.status_code, requests.codes.not_found)
423 res = response.json()
425 {"error-type": "application", "error-tag": "data-missing",
426 "error-message": "Request could not be completed because the relevant data model content does not exist"},
427 res['errors']['error'])
429 def test_17_service_path_delete_rdm_check(self):
430 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
431 "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
432 "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
433 .format(self.restconf_baseurl))
434 headers = {'content-type': 'application/json'}
435 response = requests.request(
436 "GET", url, headers=headers, auth=('admin', 'admin'))
437 self.assertEqual(response.status_code, requests.codes.not_found)
438 res = response.json()
440 {"error-type": "application", "error-tag": "data-missing",
441 "error-message": "Request could not be completed because the relevant data model content does not exist"},
442 res['errors']['error'])
444 def test_18_service_path_delete_xpdr_check(self):
445 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
446 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
447 "interface/XPDR1-NETWORK1-7"
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",
456 "error-message": "Request could not be completed because the relevant data model content does not exist"},
457 res['errors']['error'])
459 def test_19_service_path_delete_xpdr_check(self):
460 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
461 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
462 "interface/XPDR1-NETWORK1-OTU"
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",
471 "error-message": "Request could not be completed because the relevant data model content does not exist"},
472 res['errors']['error'])
474 def test_20_service_path_delete_xpdr_check(self):
475 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
476 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
477 "interface/XPDR1-NETWORK1-ODU"
478 .format(self.restconf_baseurl))
479 headers = {'content-type': 'application/json'}
480 response = requests.request(
481 "GET", url, headers=headers, auth=('admin', 'admin'))
482 self.assertEqual(response.status_code, requests.codes.not_found)
483 res = response.json()
485 {"error-type": "application", "error-tag": "data-missing",
486 "error-message": "Request could not be completed because the relevant data model content does not exist"},
487 res['errors']['error'])
489 def test_21_service_path_delete_xpdr_check(self):
490 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
491 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
492 "interface/XPDR1-CLIENT1-ETHERNET"
493 .format(self.restconf_baseurl))
494 headers = {'content-type': 'application/json'}
495 response = requests.request(
496 "GET", url, headers=headers, auth=('admin', 'admin'))
497 self.assertEqual(response.status_code, requests.codes.not_found)
498 res = response.json()
500 {"error-type": "application", "error-tag": "data-missing",
501 "error-message": "Request could not be completed because the relevant data model content does not exist"},
502 res['errors']['error'])
504 def test_22_service_path_delete_xpdr_check(self):
505 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
506 "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
507 "circuit-packs/1%2F0%2F1-PLUG-NET"
508 .format(self.restconf_baseurl))
509 headers = {'content-type': 'application/json'}
510 response = requests.request(
511 "GET", url, headers=headers, auth=('admin', 'admin'))
512 self.assertEqual(response.status_code, requests.codes.ok)
513 res = response.json()
514 self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
516 def test_23_rdm_device_disconnected(self):
517 url = ("{}/config/network-topology:"
518 "network-topology/topology/topology-netconf/node/ROADMA01"
519 .format(self.restconf_baseurl))
520 headers = {'content-type': 'application/json'}
521 response = requests.request(
522 "DELETE", url, headers=headers,
523 auth=('admin', 'admin'))
524 self.assertEqual(response.status_code, requests.codes.ok)
527 def test_24_xpdr_device_disconnected(self):
528 url = ("{}/config/network-topology:"
529 "network-topology/topology/topology-netconf/node/XPDRA01"
530 .format(self.restconf_baseurl))
531 headers = {'content-type': 'application/json'}
532 response = requests.request(
533 "DELETE", url, headers=headers,
534 auth=('admin', 'admin'))
535 self.assertEqual(response.status_code, requests.codes.ok)
539 if __name__ == "__main__":
540 unittest.main(verbosity=2, failfast=True)