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",
124 "node-id": "XPDRC01",
130 headers = {'content-type': 'application/json',
131 "Accept": "application/json"}
132 response = requests.request(
133 "POST", url, data=json.dumps(body), headers=headers,
134 auth=('admin', 'admin'))
135 self.assertEqual(response.status_code, requests.codes.ok)
136 res = response.json()
137 self.assertIn('Path is calculated',
138 res['output']['configuration-response-common']['response-message'])
141 # Path Computation success
142 def test_05_path_computation_rdm_bi(self):
143 url = ("{}/operations/transportpce-pce:path-computation-request"
144 .format(self.restconf_baseurl))
146 "service-name": "service-1",
147 "resource-reserve": "true",
148 "pce-metric": "hop-count",
149 "service-handler-header": {
150 "request-id": "request-1"
153 "node-id": "ROADMA01",
158 "node-id": "ROADMC01",
164 headers = {'content-type': 'application/json',
165 "Accept": "application/json"}
166 response = requests.request(
167 "POST", url, data=json.dumps(body), headers=headers,
168 auth=('admin', 'admin'))
169 self.assertEqual(response.status_code, requests.codes.ok)
170 res = response.json()
171 self.assertIn('Path is calculated',
172 res['output']['configuration-response-common']['response-message'])
176 def test_06_delete_simple_topology_bi(self):
177 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
178 .format(self.restconf_baseurl))
179 headers = {'content-type': 'application/xml',
180 "Accept": "application/json"}
181 response = requests.request(
182 "DELETE", url, headers=headers, auth=('admin', 'admin'))
183 self.assertEqual(response.status_code, requests.codes.ok)
186 # Test deleted topology
187 def test_07_test_topology_simple_bi_deleted(self):
188 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
189 .format(self.restconf_baseurl))
190 headers = {'content-type': 'application/json',
191 "Accept": "application/json"}
192 response = requests.request(
193 "GET", url, headers=headers, auth=('admin', 'admin'))
194 self.assertEqual(response.status_code, 404)
197 # Load simple bidirectional topology
198 def test_08_load_simple_topology_uni(self):
199 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
200 .format(self.restconf_baseurl))
201 body = self.simple_topo_uni_dir_data
202 headers = {'content-type': 'application/xml',
203 "Accept": "application/xml"}
204 response = requests.request(
205 "PUT", url, data=body, headers=headers,
206 auth=('admin', 'admin'))
207 self.assertEqual(response.status_code, 201)
210 # Get existing nodeId
211 def test_09_get_nodeId(self):
212 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
213 .format(self.restconf_baseurl))
214 headers = {'content-type': 'application/json',
215 "Accept": "application/json"}
216 response = requests.request(
217 "GET", url, headers=headers, auth=('admin', 'admin'))
218 self.assertEqual(response.status_code, requests.codes.ok)
219 res = response.json()
221 res['node'][0]['node-id'],
225 # Get existing linkId
226 def test_10_get_linkId(self):
227 url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
228 .format(self.restconf_baseurl))
229 headers = {'content-type': 'application/json',
230 "Accept": "application/json"}
231 response = requests.request(
232 "GET", url, headers=headers, auth=('admin', 'admin'))
233 self.assertEqual(response.status_code, requests.codes.ok)
234 res = response.json()
236 res['ietf-network-topology:link'][0]['link-id'],
237 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
240 # Path Computation success
241 def test_11_path_computation_xpdr_uni(self):
242 url = ("{}/operations/transportpce-pce:path-computation-request"
243 .format(self.restconf_baseurl))
245 "service-name": "service-1",
246 "resource-reserve": "true",
247 "pce-metric": "hop-count",
248 "service-handler-header": {
249 "request-id": "request-1"
252 "node-id": "XPONDER-1-2",
257 "node-id": "XPONDER-3-2",
263 headers = {'content-type': 'application/json',
264 "Accept": "application/json"}
265 response = requests.request(
266 "POST", url, data=json.dumps(body), headers=headers,
267 auth=('admin', 'admin'))
268 self.assertEqual(response.status_code, requests.codes.ok)
269 res = response.json()
270 self.assertIn('Path is calculated',
271 res['output']['configuration-response-common']['response-message'])
274 # Path Computation success
275 def test_12_path_computation_rdm_uni(self):
276 url = ("{}/operations/transportpce-pce:path-computation-request"
277 .format(self.restconf_baseurl))
279 "service-name": "service1",
280 "resource-reserve": "true",
281 "service-handler-header": {
282 "request-id": "request1"
287 "node-id": "OpenROADM-2-1"
292 "node-id": "OpenROADM-2-2"
294 "pce-metric": "hop-count"
297 headers = {'content-type': 'application/json',
298 "Accept": "application/json"}
299 response = requests.request(
300 "POST", url, data=json.dumps(body), headers=headers,
301 auth=('admin', 'admin'))
302 self.assertEqual(response.status_code, requests.codes.ok)
303 res = response.json()
304 self.assertIn('Path is calculated',
305 res['output']['configuration-response-common']['response-message'])
307 atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
308 ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
309 self.assertEqual(atozList,15)
310 self.assertEqual(ztoaList,15)
311 for i in range(0,15):
312 atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
313 ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
314 if (atoz['id'] == '14'):
315 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
316 if (ztoa['id'] == '0'):
317 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
321 def test_13_delete_simple_topology(self):
322 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
323 .format(self.restconf_baseurl))
324 headers = {'content-type': 'application/xml',
325 "Accept": "application/json"}
326 response = requests.request(
327 "DELETE", url, headers=headers, auth=('admin', 'admin'))
328 self.assertEqual(response.status_code, requests.codes.ok)
331 # Test deleted topology
332 def test_14_test_topology_simple_deleted(self):
333 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
334 .format(self.restconf_baseurl))
335 headers = {'content-type': 'application/json',
336 "Accept": "application/json"}
337 response = requests.request(
338 "GET", url, headers=headers, auth=('admin', 'admin'))
339 self.assertEqual(response.status_code, 404)
342 # Load simple topology
343 def test_15_load_complex_topology(self):
344 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
345 .format(self.restconf_baseurl))
346 body = self.complex_topo_uni_dir_data
347 headers = {'content-type': 'application/xml',
348 "Accept": "application/json"}
349 response = requests.request(
350 "PUT", url, data=body, headers=headers,
351 auth=('admin', 'admin'))
352 self.assertEqual(response.status_code, 201)
355 # Get existing nodeId
356 def test_16_get_nodeId(self):
357 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
358 .format(self.restconf_baseurl))
359 headers = {'content-type': 'application/json',
360 "Accept": "application/json"}
361 response = requests.request(
362 "GET", url, headers=headers, auth=('admin', 'admin'))
363 self.assertEqual(response.status_code, requests.codes.ok)
364 res = response.json()
366 res['node'][0]['node-id'],
370 # Test failed path computation
371 def test_17_fail_path_computation(self):
372 url = ("{}/operations/transportpce-pce:path-computation-request"
373 .format(self.restconf_baseurl))
375 "service-handler-header": {
376 "request-id": "request-1"
380 headers = {'content-type': 'application/json',
381 "Accept": "application/json"}
382 response = requests.request(
383 "POST", url, data=json.dumps(body), headers=headers,
384 auth=('admin', 'admin'))
385 self.assertEqual(response.status_code, requests.codes.ok)
386 res = response.json()
387 self.assertIn('Service Name is not set',
388 res['output']['configuration-response-common']['response-message'])
391 # Test1 success path computation
392 def test_18_success1_path_computation(self):
393 url = ("{}/operations/transportpce-pce:path-computation-request"
394 .format(self.restconf_baseurl))
396 "service-name": "service1",
397 "resource-reserve": "true",
398 "service-handler-header": {
399 "request-id": "request1"
402 "service-format": "Ethernet",
405 "node-id": "XPONDER-2-2",
408 "port-device-name": "Some port-device-name",
409 "port-type": "Some port-type",
410 "port-name": "Some port-name",
411 "port-rack": "Some port-rack",
412 "port-shelf": "Some port-shelf",
413 "port-slot": "Some port-slot",
414 "port-sub-slot": "Some port-sub-slot"
419 "port-device-name": "Some port-device-name",
420 "port-type": "Some port-type",
421 "port-name": "Some port-name",
422 "port-rack": "Some port-rack",
423 "port-shelf": "Some port-shelf",
424 "port-slot": "Some port-slot",
425 "port-sub-slot": "Some port-sub-slot"
430 "service-format": "Ethernet",
433 "node-id": "XPONDER-1-2",
436 "port-device-name": "Some port-device-name",
437 "port-type": "Some port-type",
438 "port-name": "Some port-name",
439 "port-rack": "Some port-rack",
440 "port-shelf": "Some port-shelf",
441 "port-slot": "Some port-slot",
442 "port-sub-slot": "Some port-sub-slot"
447 "port-device-name": "Some port-device-name",
448 "port-type": "Some port-type",
449 "port-name": "Some port-name",
450 "port-rack": "Some port-rack",
451 "port-shelf": "Some port-shelf",
452 "port-slot": "Some port-slot",
453 "port-sub-slot": "Some port-sub-slot"
457 "hard-constraints": {
462 "existing-service": [
463 "Some existing-service"
467 "soft-constraints": {
472 "existing-service": [
473 "Some existing-service"
477 "pce-metric": "hop-count",
478 "locally-protected-links": "true"
481 headers = {'content-type': 'application/json',
482 "Accept": "application/json"}
483 response = requests.request(
484 "POST", url, data=json.dumps(body), headers=headers,
485 auth=('admin', 'admin'))
486 self.assertEqual(response.status_code, requests.codes.ok)
487 res = response.json()
488 self.assertIn('Path is calculated',
489 res['output']['configuration-response-common']['response-message'])
492 # Test2 success path computation with path description
493 def test_19_success2_path_computation(self):
494 url = ("{}/operations/transportpce-pce:path-computation-request"
495 .format(self.restconf_baseurl))
497 "service-name": "service 1",
498 "resource-reserve": "true",
499 "service-handler-header": {
500 "request-id": "request 1"
504 "node-id": "XPONDER-1-2"
508 "node-id": "XPONDER-3-2"
510 "pce-metric": "hop-count"
513 headers = {'content-type': 'application/json',
514 "Accept": "application/json"}
515 response = requests.request(
516 "POST", url, data=json.dumps(body), headers=headers,
517 auth=('admin', 'admin'))
518 self.assertEqual(response.status_code, requests.codes.ok)
519 res = response.json()
520 self.assertIn('Path is calculated',
521 res['output']['configuration-response-common']['response-message'])
522 self.assertEqual(5 , res['output']['response-parameters']['path-description']
523 ['aToZ-direction']['aToZ-wavelength-number'])
524 self.assertEqual(5 , res['output']['response-parameters']['path-description']
525 ['zToA-direction']['zToA-wavelength-number'])
528 # Test3 success path computation with hard-constraints exclude
529 def test_20_success3_path_computation(self):
530 url = ("{}/operations/transportpce-pce:path-computation-request"
531 .format(self.restconf_baseurl))
533 "service-name": "service 1",
534 "resource-reserve": "true",
535 "service-handler-header": {
536 "request-id": "request 1"
540 "node-id": "XPONDER-1-2"
544 "node-id": "XPONDER-3-2"
546 "hard-constraints": {
548 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
551 "pce-metric": "hop-count"
554 headers = {'content-type': 'application/json',
555 "Accept": "application/json"}
556 response = requests.request(
557 "POST", url, data=json.dumps(body), headers=headers,
558 auth=('admin', 'admin'))
559 self.assertEqual(response.status_code, requests.codes.ok)
560 res = response.json()
561 self.assertIn('Path is calculated',
562 res['output']['configuration-response-common']['response-message'])
563 self.assertEqual(9 , res['output']['response-parameters']['path-description']
564 ['aToZ-direction']['aToZ-wavelength-number'])
565 self.assertEqual(9 , res['output']['response-parameters']['path-description']
566 ['zToA-direction']['zToA-wavelength-number'])
569 # Delete complex topology
570 def test_21_delete_complex_topology(self):
571 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
572 .format(self.restconf_baseurl))
573 headers = {'content-type': 'application/xml',
574 "Accept": "application/json"}
575 response = requests.request(
576 "DELETE", url, headers=headers, auth=('admin', 'admin'))
577 self.assertEqual(response.status_code, requests.codes.ok)
580 # Test deleted complex topology
581 def test_22_test_topology_complex_deleted(self):
582 url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
583 .format(self.restconf_baseurl))
584 headers = {'content-type': 'application/json',
585 "Accept": "application/json"}
586 response = requests.request(
587 "GET", url, headers=headers, auth=('admin', 'admin'))
588 self.assertEqual(response.status_code, 404)
592 if __name__ == "__main__":
593 unittest.main(verbosity=2)