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 ##############################################################################
22 class TransportPCEtesting(unittest.TestCase):
25 simple_topo_bi_dir_data = None
26 simple_topo_uni_dir_data = None
27 complex_topo_uni_dir_data = None
28 restconf_baseurl = "http://localhost:8181/restconf"
32 topo_bi_dir_file = "sample_configs/honeynode-topo.xml"
33 if os.path.isfile(topo_bi_dir_file):
34 with open(topo_bi_dir_file, 'r') as topo_bi_dir:
35 cls.simple_topo_bi_dir_data = topo_bi_dir.read();
36 topo_uni_dir_file = "sample_configs/NW-simple-topology.xml"
37 if os.path.isfile(topo_uni_dir_file):
38 with open(topo_uni_dir_file, 'r') as topo_uni_dir:
39 cls.simple_topo_uni_dir_data = topo_uni_dir.read();
40 topo_uni_dir_complex_file = "sample_configs/NW-for-test-5-4.xml"
41 if os.path.isfile(topo_uni_dir_complex_file):
42 with open(topo_uni_dir_complex_file, 'r') as topo_uni_dir_complex:
43 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read();
47 executable = "../karaf/target/assembly/bin/karaf"
48 with open('odl.log', 'w') as outfile:
49 cls.odl_process = subprocess.Popen(
50 ["bash", executable, "server"], stdout=outfile,
51 stdin=open(os.devnull))
54 def setUpClass(cls): # a class method called before tests in an individual class run.
60 def tearDownClass(cls):
61 for child in psutil.Process(cls.odl_process.pid).children():
62 child.send_signal(signal.SIGINT)
64 cls.odl_process.send_signal(signal.SIGINT)
65 cls.odl_process.wait()
67 def setUp(self): # instruction executed before each test method
70 # Load simple bidirectional topology
71 def test_01_load_simple_topology_bi(self):
72 url = ("{}/config/ietf-network:network/openroadm-topology"
73 .format(self.restconf_baseurl))
74 body = self.simple_topo_bi_dir_data
75 headers = {'content-type': 'application/xml',
76 "Accept": "application/json"}
77 response = requests.request(
78 "PUT", url, data=body, headers=headers,
79 auth=('admin', 'admin'))
80 self.assertEqual(response.status_code, requests.codes.ok)
84 def test_02_get_nodeId(self):
85 url = ("{}/config/ietf-network:network/openroadm-topology/node/ROADMA-SRG1"
86 .format(self.restconf_baseurl))
87 headers = {'content-type': 'application/json',
88 "Accept": "application/json"}
89 response = requests.request(
90 "GET", url, headers=headers, auth=('admin', 'admin'))
91 self.assertEqual(response.status_code, requests.codes.ok)
94 res['node'][0]['node-id'], 'ROADMA-SRG1')
98 def test_03_get_linkId(self):
99 url = ("{}/config/ietf-network:network/openroadm-topology/link/XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX"
100 .format(self.restconf_baseurl))
101 headers = {'content-type': 'application/json',
102 "Accept": "application/json"}
103 response = requests.request(
104 "GET", url, headers=headers, auth=('admin', 'admin'))
105 self.assertEqual(response.status_code, requests.codes.ok)
106 res = response.json()
108 res['ietf-network-topology:link'][0]['link-id'],
109 'XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX')
112 # Path Computation success
113 def test_04_path_computation_xpdr_bi(self):
114 url = ("{}/operations/transportpce-pce:path-computation-request"
115 .format(self.restconf_baseurl))
117 "service-name": "service-1",
118 "resource-reserve": "true",
119 "pce-metric": "hop-count",
120 "service-handler-header": {
121 "request-id": "request-1"
135 headers = {'content-type': 'application/json',
136 "Accept": "application/json"}
137 response = requests.request(
138 "POST", url, data=json.dumps(body), headers=headers,
139 auth=('admin', 'admin'))
140 self.assertEqual(response.status_code, requests.codes.ok)
141 res = response.json()
142 self.assertIn('Path is calculated',
143 res['output']['configuration-response-common']['response-message'])
146 # Path Computation success
147 def test_05_path_computation_rdm_bi(self):
148 url = ("{}/operations/transportpce-pce:path-computation-request"
149 .format(self.restconf_baseurl))
151 "service-name": "service-1",
152 "resource-reserve": "true",
153 "pce-metric": "hop-count",
154 "service-handler-header": {
155 "request-id": "request-1"
169 headers = {'content-type': 'application/json',
170 "Accept": "application/json"}
171 response = requests.request(
172 "POST", url, data=json.dumps(body), headers=headers,
173 auth=('admin', 'admin'))
174 self.assertEqual(response.status_code, requests.codes.ok)
175 res = response.json()
176 self.assertIn('Path is calculated',
177 res['output']['configuration-response-common']['response-message'])
181 def test_06_delete_simple_topology_bi(self):
182 url = ("{}/config/ietf-network:network/openroadm-topology"
183 .format(self.restconf_baseurl))
184 headers = {'content-type': 'application/xml',
185 "Accept": "application/json"}
186 response = requests.request(
187 "DELETE", url, headers=headers, auth=('admin', 'admin'))
188 self.assertEqual(response.status_code, requests.codes.ok)
191 # Test deleted topology
192 def test_07_test_topology_simple_bi_deleted(self):
193 url = ("{}/config/ietf-network:network/openroadm-topology/node/ROADMA-SRG1"
194 .format(self.restconf_baseurl))
195 headers = {'content-type': 'application/json',
196 "Accept": "application/json"}
197 response = requests.request(
198 "GET", url, headers=headers, auth=('admin', 'admin'))
199 self.assertEqual(response.status_code, 404)
202 # Load simple bidirectional topology
203 def test_08_load_simple_topology_uni(self):
204 url = ("{}/config/ietf-network:network/openroadm-topology"
205 .format(self.restconf_baseurl))
206 body = self.simple_topo_uni_dir_data
207 headers = {'content-type': 'application/xml',
208 "Accept": "application/json"}
209 response = requests.request(
210 "PUT", url, data=body, headers=headers,
211 auth=('admin', 'admin'))
212 self.assertEqual(response.status_code, 201)
215 # Get existing nodeId
216 def test_09_get_nodeId(self):
217 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-1-2"
218 .format(self.restconf_baseurl))
219 headers = {'content-type': 'application/json',
220 "Accept": "application/json"}
221 response = requests.request(
222 "GET", url, headers=headers, auth=('admin', 'admin'))
223 self.assertEqual(response.status_code, requests.codes.ok)
224 res = response.json()
226 res['node'][0]['node-id'],
230 # Get existing linkId
231 def test_10_get_linkId(self):
232 url = ("{}/config/ietf-network:network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
233 .format(self.restconf_baseurl))
234 headers = {'content-type': 'application/json',
235 "Accept": "application/json"}
236 response = requests.request(
237 "GET", url, headers=headers, auth=('admin', 'admin'))
238 self.assertEqual(response.status_code, requests.codes.ok)
239 res = response.json()
241 res['ietf-network-topology:link'][0]['link-id'],
242 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
245 # Path Computation success
246 def test_11_path_computation_xpdr_uni(self):
247 url = ("{}/operations/transportpce-pce:path-computation-request"
248 .format(self.restconf_baseurl))
250 "service-name": "service-1",
251 "resource-reserve": "true",
252 "pce-metric": "hop-count",
253 "service-handler-header": {
254 "request-id": "request-1"
257 "node-id": "XPONDER-1-2",
262 "node-id": "XPONDER-3-2",
268 headers = {'content-type': 'application/json',
269 "Accept": "application/json"}
270 response = requests.request(
271 "POST", url, data=json.dumps(body), headers=headers,
272 auth=('admin', 'admin'))
273 self.assertEqual(response.status_code, requests.codes.ok)
274 res = response.json()
275 self.assertIn('Path is calculated',
276 res['output']['configuration-response-common']['response-message'])
279 # Path Computation success
280 def test_12_path_computation_rdm_uni(self):
281 url = ("{}/operations/transportpce-pce:path-computation-request"
282 .format(self.restconf_baseurl))
284 "service-name": "service1",
285 "resource-reserve": "true",
286 "service-handler-header": {
287 "request-id": "request1"
292 "node-id": "OpenROADM-2-1"
297 "node-id": "OpenROADM-2-2"
299 "pce-metric": "hop-count"
302 headers = {'content-type': 'application/json',
303 "Accept": "application/json"}
304 response = requests.request(
305 "POST", url, data=json.dumps(body), headers=headers,
306 auth=('admin', 'admin'))
307 self.assertEqual(response.status_code, requests.codes.ok)
308 res = response.json()
309 self.assertIn('Path is calculated',
310 res['output']['configuration-response-common']['response-message'])
312 atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
313 ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
314 self.assertEqual(atozList,15)
315 self.assertEqual(ztoaList,15)
316 for i in range(0,15):
317 atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
318 ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
319 if (atoz['id'] == '14'):
320 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
321 if (ztoa['id'] == '0'):
322 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
326 def test_13_delete_simple_topology(self):
327 url = ("{}/config/ietf-network:network/openroadm-topology"
328 .format(self.restconf_baseurl))
329 headers = {'content-type': 'application/xml',
330 "Accept": "application/json"}
331 response = requests.request(
332 "DELETE", url, headers=headers, auth=('admin', 'admin'))
333 self.assertEqual(response.status_code, requests.codes.ok)
336 # Test deleted topology
337 def test_14_test_topology_simple_deleted(self):
338 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-1-2"
339 .format(self.restconf_baseurl))
340 headers = {'content-type': 'application/json',
341 "Accept": "application/json"}
342 response = requests.request(
343 "GET", url, headers=headers, auth=('admin', 'admin'))
344 self.assertEqual(response.status_code, 404)
347 # Load simple topology
348 def test_15_load_complex_topology(self):
349 url = ("{}/config/ietf-network:network/openroadm-topology"
350 .format(self.restconf_baseurl))
351 body = self.complex_topo_uni_dir_data
352 headers = {'content-type': 'application/xml',
353 "Accept": "application/json"}
354 response = requests.request(
355 "PUT", url, data=body, headers=headers,
356 auth=('admin', 'admin'))
357 self.assertEqual(response.status_code, 201)
360 # Get existing nodeId
361 def test_16_get_nodeId(self):
362 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
363 .format(self.restconf_baseurl))
364 headers = {'content-type': 'application/json',
365 "Accept": "application/json"}
366 response = requests.request(
367 "GET", url, headers=headers, auth=('admin', 'admin'))
368 self.assertEqual(response.status_code, requests.codes.ok)
369 res = response.json()
371 res['node'][0]['node-id'],
375 # Test failed path computation
376 def test_17_fail_path_computation(self):
377 url = ("{}/operations/transportpce-pce:path-computation-request"
378 .format(self.restconf_baseurl))
380 "service-handler-header": {
381 "request-id": "request-1"
385 headers = {'content-type': 'application/json',
386 "Accept": "application/json"}
387 response = requests.request(
388 "POST", url, data=json.dumps(body), headers=headers,
389 auth=('admin', 'admin'))
390 self.assertEqual(response.status_code, requests.codes.ok)
391 res = response.json()
392 self.assertIn('Service Name is not set',
393 res['output']['configuration-response-common']['response-message'])
396 # Test1 success path computation
397 def test_18_success1_path_computation(self):
398 url = ("{}/operations/transportpce-pce:path-computation-request"
399 .format(self.restconf_baseurl))
401 "service-name": "service1",
402 "resource-reserve": "true",
403 "service-handler-header": {
404 "request-id": "request1"
407 "service-format": "Ethernet",
410 "node-id": "XPONDER-2-2",
413 "port-device-name": "Some port-device-name",
414 "port-type": "Some port-type",
415 "port-name": "Some port-name",
416 "port-rack": "Some port-rack",
417 "port-shelf": "Some port-shelf",
418 "port-slot": "Some port-slot",
419 "port-sub-slot": "Some port-sub-slot"
424 "port-device-name": "Some port-device-name",
425 "port-type": "Some port-type",
426 "port-name": "Some port-name",
427 "port-rack": "Some port-rack",
428 "port-shelf": "Some port-shelf",
429 "port-slot": "Some port-slot",
430 "port-sub-slot": "Some port-sub-slot"
435 "service-format": "Ethernet",
438 "node-id": "XPONDER-1-2",
441 "port-device-name": "Some port-device-name",
442 "port-type": "Some port-type",
443 "port-name": "Some port-name",
444 "port-rack": "Some port-rack",
445 "port-shelf": "Some port-shelf",
446 "port-slot": "Some port-slot",
447 "port-sub-slot": "Some port-sub-slot"
452 "port-device-name": "Some port-device-name",
453 "port-type": "Some port-type",
454 "port-name": "Some port-name",
455 "port-rack": "Some port-rack",
456 "port-shelf": "Some port-shelf",
457 "port-slot": "Some port-slot",
458 "port-sub-slot": "Some port-sub-slot"
462 "hard-constraints": {
467 "existing-service": [
468 "Some existing-service"
472 "soft-constraints": {
477 "existing-service": [
478 "Some existing-service"
482 "pce-metric": "hop-count",
483 "locally-protected-links": "true"
486 headers = {'content-type': 'application/json',
487 "Accept": "application/json"}
488 response = requests.request(
489 "POST", url, data=json.dumps(body), headers=headers,
490 auth=('admin', 'admin'))
491 self.assertEqual(response.status_code, requests.codes.ok)
492 res = response.json()
493 self.assertIn('Path is calculated',
494 res['output']['configuration-response-common']['response-message'])
497 # Test2 success path computation with path description
498 def test_19_success2_path_computation(self):
499 url = ("{}/operations/transportpce-pce:path-computation-request"
500 .format(self.restconf_baseurl))
502 "service-name": "service 1",
503 "resource-reserve": "true",
504 "service-handler-header": {
505 "request-id": "request 1"
509 "node-id": "XPONDER-1-2"
513 "node-id": "XPONDER-3-2"
515 "pce-metric": "hop-count"
518 headers = {'content-type': 'application/json',
519 "Accept": "application/json"}
520 response = requests.request(
521 "POST", url, data=json.dumps(body), headers=headers,
522 auth=('admin', 'admin'))
523 self.assertEqual(response.status_code, requests.codes.ok)
524 res = response.json()
525 self.assertIn('Path is calculated',
526 res['output']['configuration-response-common']['response-message'])
527 self.assertEqual(5 , res['output']['response-parameters']['path-description']
528 ['aToZ-direction']['aToZ-wavelength-number'])
529 self.assertEqual(5 , res['output']['response-parameters']['path-description']
530 ['zToA-direction']['zToA-wavelength-number'])
533 # Test3 success path computation with hard-constraints exclude
534 def test_20_success3_path_computation(self):
535 url = ("{}/operations/transportpce-pce:path-computation-request"
536 .format(self.restconf_baseurl))
538 "service-name": "service 1",
539 "resource-reserve": "true",
540 "service-handler-header": {
541 "request-id": "request 1"
545 "node-id": "XPONDER-1-2"
549 "node-id": "XPONDER-3-2"
551 "hard-constraints": {
553 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
556 "pce-metric": "hop-count"
559 headers = {'content-type': 'application/json',
560 "Accept": "application/json"}
561 response = requests.request(
562 "POST", url, data=json.dumps(body), headers=headers,
563 auth=('admin', 'admin'))
564 self.assertEqual(response.status_code, requests.codes.ok)
565 res = response.json()
566 self.assertIn('Path is calculated',
567 res['output']['configuration-response-common']['response-message'])
568 self.assertEqual(9 , res['output']['response-parameters']['path-description']
569 ['aToZ-direction']['aToZ-wavelength-number'])
570 self.assertEqual(9 , res['output']['response-parameters']['path-description']
571 ['zToA-direction']['zToA-wavelength-number'])
574 # Delete complex topology
575 def test_21_delete_complex_topology(self):
576 url = ("{}/config/ietf-network:network/openroadm-topology"
577 .format(self.restconf_baseurl))
578 headers = {'content-type': 'application/xml',
579 "Accept": "application/json"}
580 response = requests.request(
581 "DELETE", url, headers=headers, auth=('admin', 'admin'))
582 self.assertEqual(response.status_code, requests.codes.ok)
585 # Test deleted complex topology
586 def test_22_test_topology_complex_deleted(self):
587 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
588 .format(self.restconf_baseurl))
589 headers = {'content-type': 'application/json',
590 "Accept": "application/json"}
591 response = requests.request(
592 "GET", url, headers=headers, auth=('admin', 'admin'))
593 self.assertEqual(response.status_code, 404)
597 if __name__ == "__main__":
598 unittest.main(verbosity=2)