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 ##############################################################################
17 from common import test_utils
20 class TransportPCEtesting(unittest.TestCase):
22 simple_topo_bi_dir_data = None
23 simple_topo_uni_dir_data = None
24 complex_topo_uni_dir_data = None
30 sample_files_parsed = False
31 TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
32 "..", "..", "sample_configs", "honeynode-topo.xml")
33 with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir:
34 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
36 TOPO_UNI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
37 "..", "..", "sample_configs", "NW-simple-topology.xml")
39 with open(TOPO_UNI_DIR_FILE, 'r') as topo_uni_dir:
40 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
42 TOPO_UNI_DIR_COMPLEX_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
43 "..", "..", "sample_configs", "NW-for-test-5-4.xml")
44 with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex:
45 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
46 sample_files_parsed = True
47 except PermissionError as err:
48 print("Permission Error when trying to read sample files\n", err)
50 except FileNotFoundError as err:
51 print("File Not found Error when trying to read sample files\n", err)
54 print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
57 if sample_files_parsed:
58 print("sample files content loaded")
60 cls.processes = test_utils.start_tpce()
63 def tearDownClass(cls):
64 for process in cls.processes:
65 test_utils.shutdown_process(process)
66 print("all processes killed")
68 def setUp(self): # instruction executed before each test method
71 # Load simple bidirectional topology
72 def test_01_load_simple_topology_bi(self):
73 url = "{}/config/ietf-network:networks/network/openroadm-topology"
74 data = self.simple_topo_bi_dir_data
75 response = test_utils.put_xmlrequest(url, data)
76 self.assertEqual(response.status_code, requests.codes.ok)
80 def test_02_get_nodeId(self):
81 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
82 response = test_utils.get_request(url)
83 self.assertEqual(response.status_code, requests.codes.ok)
86 res['node'][0]['node-id'], 'ROADMA01-SRG1')
90 def test_03_get_linkId(self):
91 url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
92 response = test_utils.get_request(url)
93 self.assertEqual(response.status_code, requests.codes.ok)
96 res['ietf-network-topology:link'][0]['link-id'],
97 'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
100 # Path Computation success
101 def test_04_path_computation_xpdr_bi(self):
102 url = "{}/operations/transportpce-pce:path-computation-request"
104 "service-name": "service-1",
105 "resource-reserve": "true",
106 "pce-metric": "hop-count",
107 "service-handler-header": {
108 "request-id": "request-1"
111 "node-id": "XPDRA01",
112 "service-rate": "100",
113 "service-format": "Ethernet",
117 "node-id": "XPDRC01",
118 "service-rate": "100",
119 "service-format": "Ethernet",
124 response = test_utils.post_request(url, data)
125 self.assertEqual(response.status_code, requests.codes.ok)
126 res = response.json()
127 self.assertIn('Path is calculated',
128 res['output']['configuration-response-common']['response-message'])
131 # Path Computation success
132 def test_05_path_computation_rdm_bi(self):
133 url = "{}/operations/transportpce-pce:path-computation-request"
135 "service-name": "service-1",
136 "resource-reserve": "true",
137 "pce-metric": "hop-count",
138 "service-handler-header": {
139 "request-id": "request-1"
142 "node-id": "ROADMA01",
143 "service-rate": "100",
144 "service-format": "Ethernet",
148 "node-id": "ROADMC01",
149 "service-rate": "100",
150 "service-format": "Ethernet",
155 response = test_utils.post_request(url, data)
156 self.assertEqual(response.status_code, requests.codes.ok)
157 res = response.json()
158 self.assertIn('Path is calculated',
159 res['output']['configuration-response-common']['response-message'])
163 def test_06_delete_simple_topology_bi(self):
164 url = "{}/config/ietf-network:networks/network/openroadm-topology"
165 response = test_utils.delete_request(url)
166 self.assertEqual(response.status_code, requests.codes.ok)
169 # Test deleted topology
170 def test_07_test_topology_simple_bi_deleted(self):
171 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
172 response = test_utils.get_request(url)
173 self.assertEqual(response.status_code, 404)
176 # Load simple bidirectional topology
177 def test_08_load_simple_topology_uni(self):
178 url = "{}/config/ietf-network:networks/network/openroadm-topology"
179 data = self.simple_topo_uni_dir_data
180 response = test_utils.put_xmlrequest(url, data)
181 self.assertEqual(response.status_code, 201)
184 # Get existing nodeId
185 def test_09_get_nodeId(self):
186 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
187 response = test_utils.get_request(url)
188 self.assertEqual(response.status_code, requests.codes.ok)
189 res = response.json()
191 res['node'][0]['node-id'],
195 # Get existing linkId
196 def test_10_get_linkId(self):
197 url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
198 response = test_utils.get_request(url)
199 self.assertEqual(response.status_code, requests.codes.ok)
200 res = response.json()
202 res['ietf-network-topology:link'][0]['link-id'],
203 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
206 # Path Computation success
207 def test_11_path_computation_xpdr_uni(self):
208 url = "{}/operations/transportpce-pce:path-computation-request"
210 "service-name": "service-1",
211 "resource-reserve": "true",
212 "pce-metric": "hop-count",
213 "service-handler-header": {
214 "request-id": "request-1"
217 "node-id": "XPONDER-1-2",
218 "service-rate": "100",
219 "service-format": "Ethernet",
223 "node-id": "XPONDER-3-2",
224 "service-rate": "100",
225 "service-format": "Ethernet",
230 response = test_utils.post_request(url, data)
231 self.assertEqual(response.status_code, requests.codes.ok)
232 res = response.json()
233 self.assertIn('Path is calculated',
234 res['output']['configuration-response-common']['response-message'])
237 # Path Computation success
238 def test_12_path_computation_rdm_uni(self):
239 url = "{}/operations/transportpce-pce:path-computation-request"
241 "service-name": "service1",
242 "resource-reserve": "true",
243 "service-handler-header": {
244 "request-id": "request1"
247 "service-rate": "100",
248 "service-format": "Ethernet",
250 "node-id": "OpenROADM-2-1"
253 "service-rate": "100",
254 "service-format": "Ethernet",
256 "node-id": "OpenROADM-2-2"
258 "pce-metric": "hop-count"
261 response = test_utils.post_request(url, data)
262 self.assertEqual(response.status_code, requests.codes.ok)
263 res = response.json()
264 self.assertIn('Path is calculated',
265 res['output']['configuration-response-common']['response-message'])
267 atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
268 ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
269 self.assertEqual(atozList, 15)
270 self.assertEqual(ztoaList, 15)
271 for i in range(0, 15):
272 atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
273 ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
274 if (atoz['id'] == '14'):
275 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
276 if (ztoa['id'] == '0'):
277 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
281 def test_13_delete_simple_topology(self):
282 url = "{}/config/ietf-network:networks/network/openroadm-topology"
283 response = test_utils.delete_request(url)
284 self.assertEqual(response.status_code, requests.codes.ok)
287 # Test deleted topology
288 def test_14_test_topology_simple_deleted(self):
289 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
290 response = test_utils.get_request(url)
291 self.assertEqual(response.status_code, 404)
294 # Load complex topology
295 def test_15_load_complex_topology(self):
296 url = "{}/config/ietf-network:networks/network/openroadm-topology"
297 data = self.complex_topo_uni_dir_data
298 response = test_utils.put_xmlrequest(url, data)
299 self.assertEqual(response.status_code, 201)
302 # Get existing nodeId
303 def test_16_get_nodeId(self):
304 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
305 response = test_utils.get_request(url)
306 self.assertEqual(response.status_code, requests.codes.ok)
307 res = response.json()
309 res['node'][0]['node-id'],
313 # Test failed path computation
314 def test_17_fail_path_computation(self):
315 url = "{}/operations/transportpce-pce:path-computation-request"
317 "service-handler-header": {
318 "request-id": "request-1"
322 response = test_utils.post_request(url, data)
323 self.assertEqual(response.status_code, requests.codes.ok)
324 res = response.json()
325 self.assertIn('Service Name is not set',
326 res['output']['configuration-response-common']['response-message'])
329 # Test1 success path computation
330 def test_18_success1_path_computation(self):
331 url = "{}/operations/transportpce-pce:path-computation-request"
333 "service-name": "service1",
334 "resource-reserve": "true",
335 "service-handler-header": {
336 "request-id": "request1"
339 "service-format": "Ethernet",
340 "service-rate": "100",
342 "node-id": "XPONDER-2-2",
345 "port-device-name": "Some port-device-name",
346 "port-type": "Some port-type",
347 "port-name": "Some port-name",
348 "port-rack": "Some port-rack",
349 "port-shelf": "Some port-shelf",
350 "port-slot": "Some port-slot",
351 "port-sub-slot": "Some port-sub-slot"
356 "port-device-name": "Some port-device-name",
357 "port-type": "Some port-type",
358 "port-name": "Some port-name",
359 "port-rack": "Some port-rack",
360 "port-shelf": "Some port-shelf",
361 "port-slot": "Some port-slot",
362 "port-sub-slot": "Some port-sub-slot"
367 "service-format": "Ethernet",
368 "service-rate": "100",
370 "node-id": "XPONDER-1-2",
373 "port-device-name": "Some port-device-name",
374 "port-type": "Some port-type",
375 "port-name": "Some port-name",
376 "port-rack": "Some port-rack",
377 "port-shelf": "Some port-shelf",
378 "port-slot": "Some port-slot",
379 "port-sub-slot": "Some port-sub-slot"
384 "port-device-name": "Some port-device-name",
385 "port-type": "Some port-type",
386 "port-name": "Some port-name",
387 "port-rack": "Some port-rack",
388 "port-shelf": "Some port-shelf",
389 "port-slot": "Some port-slot",
390 "port-sub-slot": "Some port-sub-slot"
394 "hard-constraints": {
399 "existing-service": [
400 "Some existing-service"
404 "soft-constraints": {
409 "existing-service": [
410 "Some existing-service"
414 "pce-metric": "hop-count",
415 "locally-protected-links": "true"
418 response = test_utils.post_request(url, data)
419 self.assertEqual(response.status_code, requests.codes.ok)
420 res = response.json()
421 self.assertIn('Path is calculated',
422 res['output']['configuration-response-common']['response-message'])
425 # Test2 success path computation with path description
426 def test_19_success2_path_computation(self):
427 url = "{}/operations/transportpce-pce:path-computation-request"
429 "service-name": "service 1",
430 "resource-reserve": "true",
431 "service-handler-header": {
432 "request-id": "request 1"
435 "service-rate": "100",
436 "service-format": "Ethernet",
437 "node-id": "XPONDER-1-2",
441 "service-rate": "100",
442 "service-format": "Ethernet",
443 "node-id": "XPONDER-3-2",
446 "pce-metric": "hop-count"
449 response = test_utils.post_request(url, data)
450 self.assertEqual(response.status_code, requests.codes.ok)
451 res = response.json()
452 self.assertIn('Path is calculated',
453 res['output']['configuration-response-common']['response-message'])
454 self.assertEqual(5, res['output']['response-parameters']['path-description']
455 ['aToZ-direction']['aToZ-wavelength-number'])
456 self.assertEqual(5, res['output']['response-parameters']['path-description']
457 ['zToA-direction']['zToA-wavelength-number'])
460 # Test3 success path computation with hard-constraints exclude
461 def test_20_success3_path_computation(self):
462 url = "{}/operations/transportpce-pce:path-computation-request"
464 "service-name": "service 1",
465 "resource-reserve": "true",
466 "service-handler-header": {
467 "request-id": "request 1"
470 "service-rate": "100",
471 "service-format": "Ethernet",
472 "node-id": "XPONDER-1-2",
476 "service-rate": "100",
477 "service-format": "Ethernet",
478 "node-id": "XPONDER-3-2",
481 "hard-constraints": {
483 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
486 "pce-metric": "hop-count"
489 response = test_utils.post_request(url, data)
490 self.assertEqual(response.status_code, requests.codes.ok)
491 res = response.json()
492 self.assertIn('Path is calculated',
493 res['output']['configuration-response-common']['response-message'])
494 self.assertEqual(9, res['output']['response-parameters']['path-description']
495 ['aToZ-direction']['aToZ-wavelength-number'])
496 self.assertEqual(9, res['output']['response-parameters']['path-description']
497 ['zToA-direction']['zToA-wavelength-number'])
500 # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
501 def test_21_path_computation_before_oms_attribute_deletion(self):
502 url = "{}/operations/transportpce-pce:path-computation-request"
504 "service-name": "service 1",
505 "resource-reserve": "true",
506 "service-handler-header": {
507 "request-id": "request 1"
510 "service-rate": "100",
511 "service-format": "Ethernet",
512 "node-id": "XPONDER-2-2",
516 "service-rate": "100",
517 "service-format": "Ethernet",
518 "node-id": "XPONDER-1-2",
521 "pce-metric": "hop-count"
524 response = test_utils.post_request(url, data)
525 self.assertEqual(response.status_code, requests.codes.ok)
526 res = response.json()
527 self.assertIn('Path is calculated',
528 res['output']['configuration-response-common']['response-message'])
529 nbElmPath = len(res['output']['response-parameters']['path-description']
530 ['aToZ-direction']['aToZ'])
531 self.assertEqual(31, nbElmPath)
532 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
534 for i in range(0, nbElmPath):
535 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
536 if(resource_i == link):
538 self.assertEqual(find, True)
541 # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
542 def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
543 url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
544 "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2/org-openroadm-network-topology:OMS-attributes/span"
546 response = test_utils.delete_request(url)
547 self.assertEqual(response.status_code, requests.codes.ok)
550 # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
551 def test_23_path_computation_after_oms_attribute_deletion(self):
552 url = "{}/operations/transportpce-pce:path-computation-request"
554 "service-name": "service 1",
555 "resource-reserve": "true",
556 "service-handler-header": {
557 "request-id": "request 1"
560 "service-rate": "100",
561 "service-format": "Ethernet",
562 "node-id": "XPONDER-2-2",
566 "service-rate": "100",
567 "service-format": "Ethernet",
568 "node-id": "XPONDER-1-2",
571 "pce-metric": "hop-count"
574 response = test_utils.post_request(url, data)
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 nbElmPath = len(res['output']['response-parameters']['path-description']
580 ['aToZ-direction']['aToZ'])
581 self.assertEqual(47, nbElmPath)
582 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
584 for i in range(0, nbElmPath):
585 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
586 if (resource_i == link):
588 self.assertNotEqual(find, True)
591 # Delete complex topology
592 def test_24_delete_complex_topology(self):
593 url = "{}/config/ietf-network:networks/network/openroadm-topology"
594 response = test_utils.delete_request(url)
595 self.assertEqual(response.status_code, requests.codes.ok)
598 # Test deleted complex topology
599 def test_25_test_topology_complex_deleted(self):
600 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
601 response = test_utils.get_request(url)
602 self.assertEqual(response.status_code, 404)
606 if __name__ == "__main__":
607 unittest.main(verbosity=2)