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):
27 restconf_baseurl = "http://localhost:8181/restconf"
31 simple_topology_file = "sample_configs/NW-simple-topology.xml"
32 if os.path.isfile(simple_topology_file):
33 with open(simple_topology_file, 'r') as simple_file:
34 cls.simple_data = simple_file.read();
35 complex_topology_file = "sample_configs/NW-for-test-5-4.xml"
36 if os.path.isfile(complex_topology_file):
37 with open(complex_topology_file, 'r') as complex_file:
38 cls.complex_data = complex_file.read();
42 executable = "../karaf/target/assembly/bin/karaf"
43 with open('odl.log', 'w') as outfile:
44 cls.odl_process = subprocess.Popen(
45 ["bash", executable, "server"], stdout=outfile,
46 stdin=open(os.devnull))
49 def setUpClass(cls): # a class method called before tests in an individual class run.
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 topology
66 def test_01_load_simple_topology(self):
67 url = ("{}/config/ietf-network:network/openroadm-topology"
68 .format(self.restconf_baseurl))
69 body = self.simple_data
70 headers = {'content-type': 'application/xml',
71 "Accept": "application/json"}
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:network/openroadm-topology/node/XPONDER-1-2"
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'],
94 def test_03_get_linkId(self):
95 url = ("{}/config/ietf-network:network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
96 .format(self.restconf_baseurl))
97 headers = {'content-type': 'application/json',
98 "Accept": "application/json"}
99 response = requests.request(
100 "GET", url, headers=headers, auth=('admin', 'admin'))
101 self.assertEqual(response.status_code, requests.codes.ok)
102 res = response.json()
104 res['ietf-network-topology:link'][0]['link-id'],
105 'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
108 # Path Computation success
109 def test_04_path_computation(self):
110 url = ("{}/operations/pce:path-computation-request"
111 .format(self.restconf_baseurl))
113 "service-name": "service-1",
114 "resource-reserve": "true",
115 "pce-metric": "hop-count",
116 "service-handler-header": {
117 "request-id": "request-1"
120 "node-id": "XPONDER-1-2",
125 "node-id": "XPONDER-3-2",
131 headers = {'content-type': 'application/json',
132 "Accept": "application/json"}
133 response = requests.request(
134 "POST", url, data=json.dumps(body), headers=headers,
135 auth=('admin', 'admin'))
136 self.assertEqual(response.status_code, requests.codes.ok)
137 res = response.json()
138 self.assertIn('Path is calculated',
139 res['output']['configuration-response-common']['response-message'])
143 def test_05_delete_simple_topology(self):
144 url = ("{}/config/ietf-network:network/openroadm-topology"
145 .format(self.restconf_baseurl))
146 headers = {'content-type': 'application/xml',
147 "Accept": "application/json"}
148 response = requests.request(
149 "DELETE", url, headers=headers, auth=('admin', 'admin'))
150 self.assertEqual(response.status_code, requests.codes.ok)
153 # Test deleted topology
154 def test_06_test_topology_simple_deleted(self):
155 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-1-2"
156 .format(self.restconf_baseurl))
157 headers = {'content-type': 'application/json',
158 "Accept": "application/json"}
159 response = requests.request(
160 "GET", url, headers=headers, auth=('admin', 'admin'))
161 self.assertEqual(response.status_code, 404)
164 # Load simple topology
165 def test_07_load_complex_topology(self):
166 url = ("{}/config/ietf-network:network/openroadm-topology"
167 .format(self.restconf_baseurl))
168 body = self.complex_data
169 headers = {'content-type': 'application/xml',
170 "Accept": "application/json"}
171 response = requests.request(
172 "PUT", url, data=body, headers=headers,
173 auth=('admin', 'admin'))
174 self.assertEqual(response.status_code, 201)
177 # Get existing nodeId
178 def test_08_get_nodeId(self):
179 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
180 .format(self.restconf_baseurl))
181 headers = {'content-type': 'application/json',
182 "Accept": "application/json"}
183 response = requests.request(
184 "GET", url, headers=headers, auth=('admin', 'admin'))
185 self.assertEqual(response.status_code, requests.codes.ok)
186 res = response.json()
188 res['node'][0]['node-id'],
192 # Test failed path computation
193 def test_09_fail_path_computation(self):
194 url = ("{}/operations/pce:path-computation-request"
195 .format(self.restconf_baseurl))
197 "service-handler-header": {
198 "request-id": "request-1"
202 headers = {'content-type': 'application/json',
203 "Accept": "application/json"}
204 response = requests.request(
205 "POST", url, data=json.dumps(body), headers=headers,
206 auth=('admin', 'admin'))
207 self.assertEqual(response.status_code, requests.codes.ok)
208 res = response.json()
209 self.assertIn('Service Name is not set',
210 res['output']['configuration-response-common']['response-message'])
213 # Test1 success path computation
214 def test_10_success1_path_computation(self):
215 url = ("{}/operations/pce:path-computation-request"
216 .format(self.restconf_baseurl))
218 "service-name": "service1",
219 "resource-reserve": "true",
220 "service-handler-header": {
221 "request-id": "request1"
224 "service-format": "Ethernet",
227 "node-id": "XPONDER-2-2",
230 "port-device-name": "Some port-device-name",
231 "port-type": "Some port-type",
232 "port-name": "Some port-name",
233 "port-rack": "Some port-rack",
234 "port-shelf": "Some port-shelf",
235 "port-slot": "Some port-slot",
236 "port-sub-slot": "Some port-sub-slot"
241 "port-device-name": "Some port-device-name",
242 "port-type": "Some port-type",
243 "port-name": "Some port-name",
244 "port-rack": "Some port-rack",
245 "port-shelf": "Some port-shelf",
246 "port-slot": "Some port-slot",
247 "port-sub-slot": "Some port-sub-slot"
252 "service-format": "Ethernet",
255 "node-id": "XPONDER-1-2",
258 "port-device-name": "Some port-device-name",
259 "port-type": "Some port-type",
260 "port-name": "Some port-name",
261 "port-rack": "Some port-rack",
262 "port-shelf": "Some port-shelf",
263 "port-slot": "Some port-slot",
264 "port-sub-slot": "Some port-sub-slot"
269 "port-device-name": "Some port-device-name",
270 "port-type": "Some port-type",
271 "port-name": "Some port-name",
272 "port-rack": "Some port-rack",
273 "port-shelf": "Some port-shelf",
274 "port-slot": "Some port-slot",
275 "port-sub-slot": "Some port-sub-slot"
279 "hard-constraints": {
284 "existing-service": [
285 "Some existing-service"
289 "soft-constraints": {
294 "existing-service": [
295 "Some existing-service"
299 "pce-metric": "hop-count",
300 "locally-protected-links": "true"
303 headers = {'content-type': 'application/json',
304 "Accept": "application/json"}
305 response = requests.request(
306 "POST", url, data=json.dumps(body), headers=headers,
307 auth=('admin', 'admin'))
308 self.assertEqual(response.status_code, requests.codes.ok)
309 res = response.json()
310 self.assertIn('Path is calculated',
311 res['output']['configuration-response-common']['response-message'])
314 # Test2 success path computation with path description
315 def test_11_success2_path_computation(self):
316 url = ("{}/operations/pce:path-computation-request"
317 .format(self.restconf_baseurl))
319 "service-name": "service 1",
320 "resource-reserve": "true",
321 "service-handler-header": {
322 "request-id": "request 1"
326 "node-id": "XPONDER-1-2"
330 "node-id": "XPONDER-3-2"
332 "pce-metric": "hop-count"
335 headers = {'content-type': 'application/json',
336 "Accept": "application/json"}
337 response = requests.request(
338 "POST", url, data=json.dumps(body), headers=headers,
339 auth=('admin', 'admin'))
340 self.assertEqual(response.status_code, requests.codes.ok)
341 res = response.json()
342 self.assertIn('Path is calculated',
343 res['output']['configuration-response-common']['response-message'])
344 self.assertEqual(5 , res['output']['response-parameters']['path-description']
345 ['aToZ-direction']['aToZ-wavelength-number'])
346 self.assertEqual(5 , res['output']['response-parameters']['path-description']
347 ['zToA-direction']['zToA-wavelength-number'])
350 # Test3 success path computation with hard-constraints exclude
351 def test_12_success3_path_computation(self):
352 url = ("{}/operations/pce:path-computation-request"
353 .format(self.restconf_baseurl))
355 "service-name": "service 1",
356 "resource-reserve": "true",
357 "service-handler-header": {
358 "request-id": "request 1"
362 "node-id": "XPONDER-1-2"
366 "node-id": "XPONDER-3-2"
368 "hard-constraints": {
370 "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
373 "pce-metric": "hop-count"
376 headers = {'content-type': 'application/json',
377 "Accept": "application/json"}
378 response = requests.request(
379 "POST", url, data=json.dumps(body), headers=headers,
380 auth=('admin', 'admin'))
381 self.assertEqual(response.status_code, requests.codes.ok)
382 res = response.json()
383 self.assertIn('Path is calculated',
384 res['output']['configuration-response-common']['response-message'])
385 self.assertEqual(9 , res['output']['response-parameters']['path-description']
386 ['aToZ-direction']['aToZ-wavelength-number'])
387 self.assertEqual(9 , res['output']['response-parameters']['path-description']
388 ['zToA-direction']['zToA-wavelength-number'])
391 # Delete complex topology
392 def test_13_delete_complex_topology(self):
393 url = ("{}/config/ietf-network:network/openroadm-topology"
394 .format(self.restconf_baseurl))
395 headers = {'content-type': 'application/xml',
396 "Accept": "application/json"}
397 response = requests.request(
398 "DELETE", url, headers=headers, auth=('admin', 'admin'))
399 self.assertEqual(response.status_code, requests.codes.ok)
402 # Test deleted complex topology
403 def test_14_test_topology_complex_deleted(self):
404 url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
405 .format(self.restconf_baseurl))
406 headers = {'content-type': 'application/json',
407 "Accept": "application/json"}
408 response = requests.request(
409 "GET", url, headers=headers, auth=('admin', 'admin'))
410 self.assertEqual(response.status_code, 404)
414 if __name__ == "__main__":
415 unittest.main(verbosity=2)