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 response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data)
74 self.assertEqual(response.status_code, requests.codes.ok)
78 def test_02_get_nodeId(self):
79 response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
80 self.assertEqual(response.status_code, requests.codes.ok)
83 res['node'][0]['node-id'], 'ROADMA01-SRG1')
87 def test_03_get_linkId(self):
88 response = test_utils.get_ordm_topo_request("link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX")
89 self.assertEqual(response.status_code, requests.codes.ok)
92 res['ietf-network-topology:link'][0]['link-id'],
93 'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
96 # Path Computation success
97 def test_04_path_computation_xpdr_bi(self):
98 url = "{}/operations/transportpce-pce:path-computation-request"
100 "service-name": "service-1",
101 "resource-reserve": "true",
102 "pce-metric": "hop-count",
103 "service-handler-header": {
104 "request-id": "request-1"
107 "node-id": "XPDRA01",
108 "service-rate": "100",
109 "service-format": "Ethernet",
113 "node-id": "XPDRC01",
114 "service-rate": "100",
115 "service-format": "Ethernet",
120 response = test_utils.post_request(url, data)
121 self.assertEqual(response.status_code, requests.codes.ok)
122 res = response.json()
123 self.assertIn('Path is calculated',
124 res['output']['configuration-response-common']['response-message'])
127 # Path Computation success
128 def test_05_path_computation_rdm_bi(self):
129 url = "{}/operations/transportpce-pce:path-computation-request"
131 "service-name": "service-1",
132 "resource-reserve": "true",
133 "pce-metric": "hop-count",
134 "service-handler-header": {
135 "request-id": "request-1"
138 "node-id": "ROADMA01",
139 "service-rate": "100",
140 "service-format": "Ethernet",
144 "node-id": "ROADMC01",
145 "service-rate": "100",
146 "service-format": "Ethernet",
151 response = test_utils.post_request(url, data)
152 self.assertEqual(response.status_code, requests.codes.ok)
153 res = response.json()
154 self.assertIn('Path is calculated',
155 res['output']['configuration-response-common']['response-message'])
159 def test_06_delete_simple_topology_bi(self):
160 response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
161 self.assertEqual(response.status_code, requests.codes.ok)
164 # Test deleted topology
165 def test_07_test_topology_simple_bi_deleted(self):
166 response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
167 self.assertEqual(response.status_code, 404)
170 # Load simple bidirectional topology
171 def test_08_load_simple_topology_uni(self):
172 response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_uni_dir_data)
173 self.assertEqual(response.status_code, 201)
176 # Get existing nodeId
177 def test_09_get_nodeId(self):
178 response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
179 self.assertEqual(response.status_code, requests.codes.ok)
180 res = response.json()
182 res['node'][0]['node-id'],
186 # Get existing linkId
187 def test_10_get_linkId(self):
188 response = test_utils.get_ordm_topo_request("link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX")
189 self.assertEqual(response.status_code, requests.codes.ok)
190 res = response.json()
192 res['ietf-network-topology:link'][0]['link-id'],
193 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
196 # Path Computation success
197 def test_11_path_computation_xpdr_uni(self):
198 url = "{}/operations/transportpce-pce:path-computation-request"
200 "service-name": "service-1",
201 "resource-reserve": "true",
202 "pce-metric": "hop-count",
203 "service-handler-header": {
204 "request-id": "request-1"
207 "node-id": "XPONDER-1-2",
208 "service-rate": "100",
209 "service-format": "Ethernet",
213 "node-id": "XPONDER-3-2",
214 "service-rate": "100",
215 "service-format": "Ethernet",
220 response = test_utils.post_request(url, data)
221 self.assertEqual(response.status_code, requests.codes.ok)
222 res = response.json()
223 self.assertIn('Path is calculated',
224 res['output']['configuration-response-common']['response-message'])
227 # Path Computation success
228 def test_12_path_computation_rdm_uni(self):
229 url = "{}/operations/transportpce-pce:path-computation-request"
231 "service-name": "service1",
232 "resource-reserve": "true",
233 "service-handler-header": {
234 "request-id": "request1"
237 "service-rate": "100",
238 "service-format": "Ethernet",
240 "node-id": "OpenROADM-2-1"
243 "service-rate": "100",
244 "service-format": "Ethernet",
246 "node-id": "OpenROADM-2-2"
248 "pce-metric": "hop-count"
251 response = test_utils.post_request(url, data)
252 self.assertEqual(response.status_code, requests.codes.ok)
253 res = response.json()
254 self.assertIn('Path is calculated',
255 res['output']['configuration-response-common']['response-message'])
257 atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
258 ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
259 self.assertEqual(atozList, 15)
260 self.assertEqual(ztoaList, 15)
261 for i in range(0, 15):
262 atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
263 ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
264 if (atoz['id'] == '14'):
265 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
266 if (ztoa['id'] == '0'):
267 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
271 def test_13_delete_simple_topology(self):
272 response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
273 self.assertEqual(response.status_code, requests.codes.ok)
276 # Test deleted topology
277 def test_14_test_topology_simple_deleted(self):
278 response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
279 self.assertEqual(response.status_code, 404)
282 # Load complex topology
283 def test_15_load_complex_topology(self):
284 response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.complex_topo_uni_dir_data)
285 self.assertEqual(response.status_code, 201)
288 # Get existing nodeId
289 def test_16_get_nodeId(self):
290 response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
291 self.assertEqual(response.status_code, requests.codes.ok)
292 res = response.json()
294 res['node'][0]['node-id'],
298 # Test failed path computation
299 def test_17_fail_path_computation(self):
300 url = "{}/operations/transportpce-pce:path-computation-request"
302 "service-handler-header": {
303 "request-id": "request-1"
307 response = test_utils.post_request(url, data)
308 self.assertEqual(response.status_code, requests.codes.ok)
309 res = response.json()
310 self.assertIn('Service Name is not set',
311 res['output']['configuration-response-common']['response-message'])
314 # Test1 success path computation
315 def test_18_success1_path_computation(self):
316 url = "{}/operations/transportpce-pce:path-computation-request"
318 "service-name": "service1",
319 "resource-reserve": "true",
320 "service-handler-header": {
321 "request-id": "request1"
324 "service-format": "Ethernet",
325 "service-rate": "100",
327 "node-id": "XPONDER-2-2",
330 "port-device-name": "Some port-device-name",
331 "port-type": "Some port-type",
332 "port-name": "Some port-name",
333 "port-rack": "Some port-rack",
334 "port-shelf": "Some port-shelf",
335 "port-slot": "Some port-slot",
336 "port-sub-slot": "Some port-sub-slot"
341 "port-device-name": "Some port-device-name",
342 "port-type": "Some port-type",
343 "port-name": "Some port-name",
344 "port-rack": "Some port-rack",
345 "port-shelf": "Some port-shelf",
346 "port-slot": "Some port-slot",
347 "port-sub-slot": "Some port-sub-slot"
352 "service-format": "Ethernet",
353 "service-rate": "100",
355 "node-id": "XPONDER-1-2",
358 "port-device-name": "Some port-device-name",
359 "port-type": "Some port-type",
360 "port-name": "Some port-name",
361 "port-rack": "Some port-rack",
362 "port-shelf": "Some port-shelf",
363 "port-slot": "Some port-slot",
364 "port-sub-slot": "Some port-sub-slot"
369 "port-device-name": "Some port-device-name",
370 "port-type": "Some port-type",
371 "port-name": "Some port-name",
372 "port-rack": "Some port-rack",
373 "port-shelf": "Some port-shelf",
374 "port-slot": "Some port-slot",
375 "port-sub-slot": "Some port-sub-slot"
379 "hard-constraints": {
384 "existing-service": [
385 "Some existing-service"
389 "soft-constraints": {
394 "existing-service": [
395 "Some existing-service"
399 "pce-metric": "hop-count",
400 "locally-protected-links": "true"
403 response = test_utils.post_request(url, data)
404 self.assertEqual(response.status_code, requests.codes.ok)
405 res = response.json()
406 self.assertIn('Path is calculated',
407 res['output']['configuration-response-common']['response-message'])
410 # Test2 success path computation with path description
411 def test_19_success2_path_computation(self):
412 url = "{}/operations/transportpce-pce:path-computation-request"
414 "service-name": "service 1",
415 "resource-reserve": "true",
416 "service-handler-header": {
417 "request-id": "request 1"
420 "service-rate": "100",
421 "service-format": "Ethernet",
422 "node-id": "XPONDER-1-2",
426 "service-rate": "100",
427 "service-format": "Ethernet",
428 "node-id": "XPONDER-3-2",
431 "pce-metric": "hop-count"
434 response = test_utils.post_request(url, data)
435 self.assertEqual(response.status_code, requests.codes.ok)
436 res = response.json()
437 self.assertIn('Path is calculated',
438 res['output']['configuration-response-common']['response-message'])
439 self.assertEqual(5, res['output']['response-parameters']['path-description']
440 ['aToZ-direction']['aToZ-wavelength-number'])
441 self.assertEqual(5, res['output']['response-parameters']['path-description']
442 ['zToA-direction']['zToA-wavelength-number'])
445 # Test3 success path computation with hard-constraints exclude
446 def test_20_success3_path_computation(self):
447 url = "{}/operations/transportpce-pce:path-computation-request"
449 "service-name": "service 1",
450 "resource-reserve": "true",
451 "service-handler-header": {
452 "request-id": "request 1"
455 "service-rate": "100",
456 "service-format": "Ethernet",
457 "node-id": "XPONDER-1-2",
461 "service-rate": "100",
462 "service-format": "Ethernet",
463 "node-id": "XPONDER-3-2",
466 "hard-constraints": {
468 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
471 "pce-metric": "hop-count"
474 response = test_utils.post_request(url, data)
475 self.assertEqual(response.status_code, requests.codes.ok)
476 res = response.json()
477 self.assertIn('Path is calculated',
478 res['output']['configuration-response-common']['response-message'])
479 self.assertEqual(9, res['output']['response-parameters']['path-description']
480 ['aToZ-direction']['aToZ-wavelength-number'])
481 self.assertEqual(9, res['output']['response-parameters']['path-description']
482 ['zToA-direction']['zToA-wavelength-number'])
485 # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
486 def test_21_path_computation_before_oms_attribute_deletion(self):
487 url = "{}/operations/transportpce-pce:path-computation-request"
489 "service-name": "service 1",
490 "resource-reserve": "true",
491 "service-handler-header": {
492 "request-id": "request 1"
495 "service-rate": "100",
496 "service-format": "Ethernet",
497 "node-id": "XPONDER-2-2",
501 "service-rate": "100",
502 "service-format": "Ethernet",
503 "node-id": "XPONDER-1-2",
506 "pce-metric": "hop-count"
509 response = test_utils.post_request(url, data)
510 self.assertEqual(response.status_code, requests.codes.ok)
511 res = response.json()
512 self.assertIn('Path is calculated',
513 res['output']['configuration-response-common']['response-message'])
514 nbElmPath = len(res['output']['response-parameters']['path-description']
515 ['aToZ-direction']['aToZ'])
516 self.assertEqual(31, nbElmPath)
517 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
519 for i in range(0, nbElmPath):
520 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
521 if(resource_i == link):
523 self.assertEqual(find, True)
526 # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
527 def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
528 response = test_utils.del_oms_attr_request("OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2")
529 self.assertEqual(response.status_code, requests.codes.ok)
532 # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
533 def test_23_path_computation_after_oms_attribute_deletion(self):
534 url = "{}/operations/transportpce-pce:path-computation-request"
536 "service-name": "service 1",
537 "resource-reserve": "true",
538 "service-handler-header": {
539 "request-id": "request 1"
542 "service-rate": "100",
543 "service-format": "Ethernet",
544 "node-id": "XPONDER-2-2",
548 "service-rate": "100",
549 "service-format": "Ethernet",
550 "node-id": "XPONDER-1-2",
553 "pce-metric": "hop-count"
556 response = test_utils.post_request(url, data)
557 self.assertEqual(response.status_code, requests.codes.ok)
558 res = response.json()
559 self.assertIn('Path is calculated',
560 res['output']['configuration-response-common']['response-message'])
561 nbElmPath = len(res['output']['response-parameters']['path-description']
562 ['aToZ-direction']['aToZ'])
563 self.assertEqual(47, nbElmPath)
564 link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
566 for i in range(0, nbElmPath):
567 resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
568 if (resource_i == link):
570 self.assertNotEqual(find, True)
573 # Delete complex topology
574 def test_24_delete_complex_topology(self):
575 response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
576 self.assertEqual(response.status_code, requests.codes.ok)
579 # Test deleted complex topology
580 def test_25_test_topology_complex_deleted(self):
581 response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
582 self.assertEqual(response.status_code, 404)
586 if __name__ == "__main__":
587 unittest.main(verbosity=2)