2 ##############################################################################
3 # Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 ##############################################################################
23 class TransportPCEtesting(unittest.TestCase):
26 simple_topo_bi_dir_data = None
27 simple_topo_uni_dir_data = None
28 complex_topo_uni_dir_data = None
29 restconf_baseurl = "http://localhost:8181/restconf"
33 topo_bi_dir_file = "sample_configs/honeynode-topo.xml"
34 if os.path.isfile(topo_bi_dir_file):
35 with open(topo_bi_dir_file, 'r') as topo_bi_dir:
36 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
37 topo_uni_dir_file = "sample_configs/NW-simple-topology.xml"
38 if os.path.isfile(topo_uni_dir_file):
39 with open(topo_uni_dir_file, 'r') as topo_uni_dir:
40 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
41 topo_uni_dir_complex_file = "sample_configs/NW-for-test-5-4.xml"
42 if os.path.isfile(topo_uni_dir_complex_file):
43 with open(topo_uni_dir_complex_file, 'r') as topo_uni_dir_complex:
44 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
47 def setUpClass(cls): # a class method called before tests in an individual class run.
49 print("starting opendaylight...")
50 cls.odl_process = test_utils.start_tpce()
52 print("opendaylight started")
55 def tearDownClass(cls):
56 for child in psutil.Process(cls.odl_process.pid).children():
57 child.send_signal(signal.SIGINT)
59 cls.odl_process.send_signal(signal.SIGINT)
60 cls.odl_process.wait()
62 def setUp(self): # instruction executed before each test method
65 # Load simple bidirectional topology
66 def test_01_load_simple_topology_bi(self):
67 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
68 .format(self.restconf_baseurl))
69 body = self.simple_topo_bi_dir_data
70 headers = {'content-type': 'application/xml',
71 "Accept": "application/xml"}
72 response = requests.request(
73 "PUT", url, data=body, headers=headers,
74 auth=('admin', 'admin'))
75 self.assertEqual(response.status_code, requests.codes.ok)
79 def test_02_get_nodeId(self):
80 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
81 .format(self.restconf_baseurl))
82 headers = {'content-type': 'application/json',
83 "Accept": "application/json"}
84 response = requests.request(
85 "GET", url, headers=headers, auth=('admin', 'admin'))
86 self.assertEqual(response.status_code, requests.codes.ok)
89 res['node'][0]['node-id'], 'ROADMA01-SRG1')
93 def test_03_get_linkId(self):
94 url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
95 .format(self.restconf_baseurl))
96 headers = {'content-type': 'application/json',
97 "Accept": "application/json"}
98 response = requests.request(
99 "GET", url, headers=headers, auth=('admin', 'admin'))
100 self.assertEqual(response.status_code, requests.codes.ok)
101 res = response.json()
103 res['ietf-network-topology:link'][0]['link-id'],
104 'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
107 # Path Computation success
108 def test_04_path_computation_xpdr_bi(self):
109 url = ("{}/operations/transportpce-pce:path-computation-request"
110 .format(self.restconf_baseurl))
112 "service-name": "service-1",
113 "resource-reserve": "true",
114 "pce-metric": "hop-count",
115 "service-handler-header": {
116 "request-id": "request-1"
119 "node-id": "XPDRA01",
120 "service-rate": "100",
121 "service-format": "Ethernet",
125 "node-id": "XPDRC01",
126 "service-rate": "100",
127 "service-format": "Ethernet",
132 headers = {'content-type': 'application/json',
133 "Accept": "application/json"}
134 response = requests.request(
135 "POST", url, data=json.dumps(body), headers=headers,
136 auth=('admin', 'admin'))
137 self.assertEqual(response.status_code, requests.codes.ok)
138 res = response.json()
139 self.assertIn('Path is calculated',
140 res['output']['configuration-response-common']['response-message'])
143 # Path Computation success
144 def test_05_path_computation_rdm_bi(self):
145 url = ("{}/operations/transportpce-pce:path-computation-request"
146 .format(self.restconf_baseurl))
148 "service-name": "service-1",
149 "resource-reserve": "true",
150 "pce-metric": "hop-count",
151 "service-handler-header": {
152 "request-id": "request-1"
155 "node-id": "ROADMA01",
156 "service-rate": "100",
157 "service-format": "Ethernet",
161 "node-id": "ROADMC01",
162 "service-rate": "100",
163 "service-format": "Ethernet",
168 headers = {'content-type': 'application/json',
169 "Accept": "application/json"}
170 response = requests.request(
171 "POST", url, data=json.dumps(body), headers=headers,
172 auth=('admin', 'admin'))
173 self.assertEqual(response.status_code, requests.codes.ok)
174 res = response.json()
175 self.assertIn('Path is calculated',
176 res['output']['configuration-response-common']['response-message'])
180 def test_06_delete_simple_topology_bi(self):
181 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
182 .format(self.restconf_baseurl))
183 headers = {'content-type': 'application/xml',
184 "Accept": "application/json"}
185 response = requests.request(
186 "DELETE", url, headers=headers, auth=('admin', 'admin'))
187 self.assertEqual(response.status_code, requests.codes.ok)
190 # Test deleted topology
191 def test_07_test_topology_simple_bi_deleted(self):
192 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
193 .format(self.restconf_baseurl))
194 headers = {'content-type': 'application/json',
195 "Accept": "application/json"}
196 response = requests.request(
197 "GET", url, headers=headers, auth=('admin', 'admin'))
198 self.assertEqual(response.status_code, 404)
201 # Load simple bidirectional topology
202 def test_08_load_simple_topology_uni(self):
203 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
204 .format(self.restconf_baseurl))
205 body = self.simple_topo_uni_dir_data
206 headers = {'content-type': 'application/xml',
207 "Accept": "application/xml"}
208 response = requests.request(
209 "PUT", url, data=body, headers=headers,
210 auth=('admin', 'admin'))
211 self.assertEqual(response.status_code, 201)
214 # Get existing nodeId
215 def test_09_get_nodeId(self):
216 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
217 .format(self.restconf_baseurl))
218 headers = {'content-type': 'application/json',
219 "Accept": "application/json"}
220 response = requests.request(
221 "GET", url, headers=headers, auth=('admin', 'admin'))
222 self.assertEqual(response.status_code, requests.codes.ok)
223 res = response.json()
225 res['node'][0]['node-id'],
229 # Get existing linkId
230 def test_10_get_linkId(self):
231 url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
232 .format(self.restconf_baseurl))
233 headers = {'content-type': 'application/json',
234 "Accept": "application/json"}
235 response = requests.request(
236 "GET", url, headers=headers, auth=('admin', 'admin'))
237 self.assertEqual(response.status_code, requests.codes.ok)
238 res = response.json()
240 res['ietf-network-topology:link'][0]['link-id'],
241 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
244 # Path Computation success
245 def test_11_path_computation_xpdr_uni(self):
246 url = ("{}/operations/transportpce-pce:path-computation-request"
247 .format(self.restconf_baseurl))
249 "service-name": "service-1",
250 "resource-reserve": "true",
251 "pce-metric": "hop-count",
252 "service-handler-header": {
253 "request-id": "request-1"
256 "node-id": "XPONDER-1-2",
257 "service-rate": "100",
258 "service-format": "Ethernet",
262 "node-id": "XPONDER-3-2",
263 "service-rate": "100",
264 "service-format": "Ethernet",
269 headers = {'content-type': 'application/json',
270 "Accept": "application/json"}
271 response = requests.request(
272 "POST", url, data=json.dumps(body), headers=headers,
273 auth=('admin', 'admin'))
274 self.assertEqual(response.status_code, requests.codes.ok)
275 res = response.json()
276 self.assertIn('Path is calculated',
277 res['output']['configuration-response-common']['response-message'])
280 # Path Computation success
281 def test_12_path_computation_rdm_uni(self):
282 url = ("{}/operations/transportpce-pce:path-computation-request"
283 .format(self.restconf_baseurl))
285 "service-name": "service1",
286 "resource-reserve": "true",
287 "service-handler-header": {
288 "request-id": "request1"
291 "service-rate": "100",
292 "service-format": "Ethernet",
294 "node-id": "OpenROADM-2-1"
297 "service-rate": "100",
298 "service-format": "Ethernet",
300 "node-id": "OpenROADM-2-2"
302 "pce-metric": "hop-count"
305 headers = {'content-type': 'application/json',
306 "Accept": "application/json"}
307 response = requests.request(
308 "POST", url, data=json.dumps(body), headers=headers,
309 auth=('admin', 'admin'))
310 self.assertEqual(response.status_code, requests.codes.ok)
311 res = response.json()
312 self.assertIn('Path is calculated',
313 res['output']['configuration-response-common']['response-message'])
315 atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
316 ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
317 self.assertEqual(atozList, 15)
318 self.assertEqual(ztoaList, 15)
319 for i in range(0, 15):
320 atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
321 ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
322 if (atoz['id'] == '14'):
323 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
324 if (ztoa['id'] == '0'):
325 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
329 def test_13_delete_simple_topology(self):
330 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
331 .format(self.restconf_baseurl))
332 headers = {'content-type': 'application/xml',
333 "Accept": "application/json"}
334 response = requests.request(
335 "DELETE", url, headers=headers, auth=('admin', 'admin'))
336 self.assertEqual(response.status_code, requests.codes.ok)
339 # Test deleted topology
340 def test_14_test_topology_simple_deleted(self):
341 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
342 .format(self.restconf_baseurl))
343 headers = {'content-type': 'application/json',
344 "Accept": "application/json"}
345 response = requests.request(
346 "GET", url, headers=headers, auth=('admin', 'admin'))
347 self.assertEqual(response.status_code, 404)
350 # Load complex topology
351 def test_15_load_complex_topology(self):
352 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
353 .format(self.restconf_baseurl))
354 body = self.complex_topo_uni_dir_data
355 headers = {'content-type': 'application/xml',
356 "Accept": "application/json"}
357 response = requests.request(
358 "PUT", url, data=body, headers=headers,
359 auth=('admin', 'admin'))
360 self.assertEqual(response.status_code, 201)
363 # Get existing nodeId
364 def test_16_get_nodeId(self):
365 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
366 .format(self.restconf_baseurl))
367 headers = {'content-type': 'application/json',
368 "Accept": "application/json"}
369 response = requests.request(
370 "GET", url, headers=headers, auth=('admin', 'admin'))
371 self.assertEqual(response.status_code, requests.codes.ok)
372 res = response.json()
374 res['node'][0]['node-id'],
378 # Test failed path computation
379 def test_17_fail_path_computation(self):
380 url = ("{}/operations/transportpce-pce:path-computation-request"
381 .format(self.restconf_baseurl))
383 "service-handler-header": {
384 "request-id": "request-1"
388 headers = {'content-type': 'application/json',
389 "Accept": "application/json"}
390 response = requests.request(
391 "POST", url, data=json.dumps(body), headers=headers,
392 auth=('admin', 'admin'))
393 self.assertEqual(response.status_code, requests.codes.ok)
394 res = response.json()
395 self.assertIn('Service Name is not set',
396 res['output']['configuration-response-common']['response-message'])
399 # Test1 success path computation
400 def test_18_success1_path_computation(self):
401 url = ("{}/operations/transportpce-pce:path-computation-request"
402 .format(self.restconf_baseurl))
404 "service-name": "service1",
405 "resource-reserve": "true",
406 "service-handler-header": {
407 "request-id": "request1"
410 "service-format": "Ethernet",
411 "service-rate": "100",
413 "node-id": "XPONDER-2-2",
416 "port-device-name": "Some port-device-name",
417 "port-type": "Some port-type",
418 "port-name": "Some port-name",
419 "port-rack": "Some port-rack",
420 "port-shelf": "Some port-shelf",
421 "port-slot": "Some port-slot",
422 "port-sub-slot": "Some port-sub-slot"
427 "port-device-name": "Some port-device-name",
428 "port-type": "Some port-type",
429 "port-name": "Some port-name",
430 "port-rack": "Some port-rack",
431 "port-shelf": "Some port-shelf",
432 "port-slot": "Some port-slot",
433 "port-sub-slot": "Some port-sub-slot"
438 "service-format": "Ethernet",
439 "service-rate": "100",
441 "node-id": "XPONDER-1-2",
444 "port-device-name": "Some port-device-name",
445 "port-type": "Some port-type",
446 "port-name": "Some port-name",
447 "port-rack": "Some port-rack",
448 "port-shelf": "Some port-shelf",
449 "port-slot": "Some port-slot",
450 "port-sub-slot": "Some port-sub-slot"
455 "port-device-name": "Some port-device-name",
456 "port-type": "Some port-type",
457 "port-name": "Some port-name",
458 "port-rack": "Some port-rack",
459 "port-shelf": "Some port-shelf",
460 "port-slot": "Some port-slot",
461 "port-sub-slot": "Some port-sub-slot"
465 "hard-constraints": {
470 "existing-service": [
471 "Some existing-service"
475 "soft-constraints": {
480 "existing-service": [
481 "Some existing-service"
485 "pce-metric": "hop-count",
486 "locally-protected-links": "true"
489 headers = {'content-type': 'application/json',
490 "Accept": "application/json"}
491 response = requests.request(
492 "POST", url, data=json.dumps(body), headers=headers,
493 auth=('admin', 'admin'))
494 self.assertEqual(response.status_code, requests.codes.ok)
495 res = response.json()
496 self.assertIn('Path is calculated',
497 res['output']['configuration-response-common']['response-message'])
500 # Test2 success path computation with path description
501 def test_19_success2_path_computation(self):
502 url = ("{}/operations/transportpce-pce:path-computation-request"
503 .format(self.restconf_baseurl))
505 "service-name": "service 1",
506 "resource-reserve": "true",
507 "service-handler-header": {
508 "request-id": "request 1"
511 "service-rate": "100",
512 "service-format": "Ethernet",
513 "node-id": "XPONDER-1-2",
517 "service-rate": "100",
518 "service-format": "Ethernet",
519 "node-id": "XPONDER-3-2",
522 "pce-metric": "hop-count"
525 headers = {'content-type': 'application/json',
526 "Accept": "application/json"}
527 response = requests.request(
528 "POST", url, data=json.dumps(body), headers=headers,
529 auth=('admin', 'admin'))
530 self.assertEqual(response.status_code, requests.codes.ok)
531 res = response.json()
532 self.assertIn('Path is calculated',
533 res['output']['configuration-response-common']['response-message'])
534 self.assertEqual(5, res['output']['response-parameters']['path-description']
535 ['aToZ-direction']['aToZ-wavelength-number'])
536 self.assertEqual(5, res['output']['response-parameters']['path-description']
537 ['zToA-direction']['zToA-wavelength-number'])
540 # Test3 success path computation with hard-constraints exclude
541 def test_20_success3_path_computation(self):
542 url = ("{}/operations/transportpce-pce:path-computation-request"
543 .format(self.restconf_baseurl))
545 "service-name": "service 1",
546 "resource-reserve": "true",
547 "service-handler-header": {
548 "request-id": "request 1"
551 "service-rate": "100",
552 "service-format": "Ethernet",
553 "node-id": "XPONDER-1-2",
557 "service-rate": "100",
558 "service-format": "Ethernet",
559 "node-id": "XPONDER-3-2",
562 "hard-constraints": {
564 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
567 "pce-metric": "hop-count"
570 headers = {'content-type': 'application/json',
571 "Accept": "application/json"}
572 response = requests.request(
573 "POST", url, data=json.dumps(body), headers=headers,
574 auth=('admin', 'admin'))
575 self.assertEqual(response.status_code, requests.codes.ok)
576 res = response.json()
577 self.assertIn('Path is calculated',
578 res['output']['configuration-response-common']['response-message'])
579 self.assertEqual(9, res['output']['response-parameters']['path-description']
580 ['aToZ-direction']['aToZ-wavelength-number'])
581 self.assertEqual(9, res['output']['response-parameters']['path-description']
582 ['zToA-direction']['zToA-wavelength-number'])
585 # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
586 def test_21_path_computation_before_oms_attribute_deletion(self):
587 url = ("{}/operations/transportpce-pce:path-computation-request"
588 .format(self.restconf_baseurl))
590 "service-name": "service 1",
591 "resource-reserve": "true",
592 "service-handler-header": {
593 "request-id": "request 1"
596 "service-rate": "100",
597 "service-format": "Ethernet",
598 "node-id": "XPONDER-2-2",
602 "service-rate": "100",
603 "service-format": "Ethernet",
604 "node-id": "XPONDER-1-2",
607 "pce-metric": "hop-count"
610 headers = {'content-type': 'application/json',
611 "Accept": "application/json"}
612 response = requests.request(
613 "POST", url, data=json.dumps(body), headers=headers,
614 auth=('admin', 'admin'))
615 self.assertEqual(response.status_code, requests.codes.ok)
616 res = response.json()
617 self.assertIn('Path is calculated',
618 res['output']['configuration-response-common']['response-message'])
619 nbElmPath = len(res['output']['response-parameters']['path-description']
620 ['aToZ-direction']['aToZ'])
621 self.assertEqual(31, nbElmPath)
622 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
624 for i in range(0, nbElmPath):
625 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
626 if(resource_i == link):
628 self.assertEqual(find, True)
631 # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
632 def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
633 url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
634 "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2/org-openroadm-network-topology:OMS-attributes/span"
635 .format(self.restconf_baseurl))
636 headers = {'content-type': 'application/xml',
637 "Accept": "application/json"}
638 response = requests.request(
639 "DELETE", url, headers=headers, auth=('admin', 'admin'))
640 self.assertEqual(response.status_code, requests.codes.ok)
643 # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
644 def test_23_path_computation_after_oms_attribute_deletion(self):
645 url = ("{}/operations/transportpce-pce:path-computation-request"
646 .format(self.restconf_baseurl))
648 "service-name": "service 1",
649 "resource-reserve": "true",
650 "service-handler-header": {
651 "request-id": "request 1"
654 "service-rate": "100",
655 "service-format": "Ethernet",
656 "node-id": "XPONDER-2-2",
660 "service-rate": "100",
661 "service-format": "Ethernet",
662 "node-id": "XPONDER-1-2",
665 "pce-metric": "hop-count"
668 headers = {'content-type': 'application/json',
669 "Accept": "application/json"}
670 response = requests.request(
671 "POST", url, data=json.dumps(body), headers=headers,
672 auth=('admin', 'admin'))
673 self.assertEqual(response.status_code, requests.codes.ok)
674 res = response.json()
675 self.assertIn('Path is calculated',
676 res['output']['configuration-response-common']['response-message'])
677 nbElmPath = len(res['output']['response-parameters']['path-description']
678 ['aToZ-direction']['aToZ'])
679 self.assertEqual(47, nbElmPath)
680 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
682 for i in range(0, nbElmPath):
683 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
684 if (resource_i == link):
686 self.assertNotEqual(find, True)
689 # Delete complex topology
690 def test_24_delete_complex_topology(self):
691 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
692 .format(self.restconf_baseurl))
693 headers = {'content-type': 'application/xml',
694 "Accept": "application/json"}
695 response = requests.request(
696 "DELETE", url, headers=headers, auth=('admin', 'admin'))
697 self.assertEqual(response.status_code, requests.codes.ok)
700 # Test deleted complex topology
701 def test_25_test_topology_complex_deleted(self):
702 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
703 .format(self.restconf_baseurl))
704 headers = {'content-type': 'application/json',
705 "Accept": "application/json"}
706 response = requests.request(
707 "GET", url, headers=headers, auth=('admin', 'admin'))
708 self.assertEqual(response.status_code, 404)
712 if __name__ == "__main__":
713 unittest.main(verbosity=2)