3 ##############################################################################
4 #Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
24 class TransportGNPYtesting(unittest.TestCase):
28 restconf_baseurl = "http://localhost:8181/restconf"
31 def __init_logfile(cls):
32 if os.path.isfile("./transportpce_tests/gnpy.log"):
33 os.remove("transportpce_tests/gnpy.log")
37 print ("starting opendaylight...")
38 cls.odl_process = test_utils.start_tpce()
40 print ("opendaylight started")
43 def tearDownClass(cls):
44 for child in psutil.Process(cls.odl_process.pid).children():
45 child.send_signal(signal.SIGINT)
47 cls.odl_process.send_signal(signal.SIGINT)
48 cls.odl_process.wait()
53 #Mount the different topologies
54 def test_01_connect_clliNetwork(self):
55 url = ("{}/config/ietf-network:networks/network/clli-network"
56 .format(self.restconf_baseurl))
57 topo_clliNet_file = "sample_configs/gnpy/clliNetwork.json"
58 if os.path.isfile(topo_clliNet_file):
59 with open(topo_clliNet_file, 'r') as clli_net:
60 body = clli_net.read()
61 headers = {'content-type': 'application/json'}
62 response = requests.request(
63 "PUT", url, data=body, headers=headers,
64 auth=('admin', 'admin'))
65 self.assertEqual(response.status_code, requests.codes.ok)
68 def test_02_connect_openroadmNetwork(self):
69 url = ("{}/config/ietf-network:networks/network/openroadm-network"
70 .format(self.restconf_baseurl))
71 topo_ordNet_file = "sample_configs/gnpy/openroadmNetwork.json"
72 if os.path.isfile(topo_ordNet_file):
73 with open(topo_ordNet_file, 'r') as ord_net:
75 headers = {'content-type': 'application/json'}
76 response = requests.request(
77 "PUT", url, data=body, headers=headers,
78 auth=('admin', 'admin'))
79 self.assertEqual(response.status_code, requests.codes.ok)
82 def test_03_connect_openroadmTopology(self):
83 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
84 .format(self.restconf_baseurl))
85 topo_ordTopo_file = "sample_configs/gnpy/openroadmTopology.json"
86 if os.path.isfile(topo_ordTopo_file):
87 with open(topo_ordTopo_file, 'r') as ord_topo:
88 body = ord_topo.read()
89 headers = {'content-type': 'application/json'}
90 response = requests.request(
91 "PUT", url, data=body, headers=headers,
92 auth=('admin', 'admin'))
93 self.assertEqual(response.status_code, requests.codes.ok)
96 #Path computed by PCE is feasible according to Gnpy
97 def test_04_path_computation_FeasibleWithPCE(self):
98 url = ("{}/operations/transportpce-pce:path-computation-request"
99 .format(self.restconf_baseurl))
101 "service-name": "service-1",
102 "resource-reserve": "true",
103 "pce-metric": "hop-count",
104 "service-handler-header": {
105 "request-id": "request-1"
108 "node-id": "XPONDER-1",
109 "service-rate": "100",
110 "service-format": "Ethernet",
114 "node-id": "XPONDER-5",
115 "service-rate": "100",
116 "service-format": "Ethernet",
121 headers = {'content-type': 'application/json',
122 "Accept": "application/json"}
123 response = requests.request(
124 "POST", url, data=json.dumps(body), headers=headers,
125 auth=('admin', 'admin'))
126 self.assertEqual(response.status_code, requests.codes.ok)
127 res = response.json()
128 self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
129 self.assertEqual(res['output']['configuration-response-common']['response-message'],
130 'Path is calculated by PCE')
131 self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
132 self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
133 self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
134 self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
137 #Path computed by PCE is not feasible by GNPy and GNPy cannot find another one (low SNR)
138 def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
139 url = ("{}/operations/transportpce-pce:path-computation-request"
140 .format(self.restconf_baseurl))
142 "service-name": "service-2",
143 "resource-reserve": "true",
144 "pce-metric": "hop-count",
145 "service-handler-header": {
146 "request-id": "request-2"
149 "node-id": "XPONDER-1",
150 "service-rate": "100",
151 "service-format": "Ethernet",
155 "node-id": "XPONDER-5",
156 "service-rate": "100",
157 "service-format": "Ethernet",
160 "hard-constraints": {
166 "node-id": "OpenROADM-2"
172 "node-id": "OpenROADM-3"
178 "node-id": "OpenROADM-4"
186 headers = {'content-type': 'application/json',
187 "Accept": "application/json"}
188 response = requests.request(
189 "POST", url, data=json.dumps(body), headers=headers,
190 auth=('admin', 'admin'))
191 self.assertEqual(response.status_code, requests.codes.ok)
192 res = response.json()
193 self.assertEqual(res['output']['configuration-response-common']['response-code'], '500')
194 self.assertEqual(res['output']['configuration-response-common']['response-message'],
195 'No path available by PCE and GNPy ')
196 self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
197 self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],False)
198 self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
199 self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],False)
202 # #PCE cannot find a path while GNPy finds a feasible one
203 def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
204 url = ("{}/operations/transportpce-pce:path-computation-request"
205 .format(self.restconf_baseurl))
207 "service-name": "service-3",
208 "resource-reserve": "true",
209 "pce-metric": "hop-count",
210 "service-handler-header": {
211 "request-id": "request-3"
214 "node-id": "XPONDER-1",
215 "service-rate": "100",
216 "service-format": "Ethernet",
220 "node-id": "XPONDER-4",
221 "service-rate": "100",
222 "service-format": "Ethernet",
225 "hard-constraints": {
231 "node-id": "OpenROADM-2"
237 "node-id": "OpenROADM-3"
245 headers = {'content-type': 'application/json',
246 "Accept": "application/json"}
247 response = requests.request(
248 "POST", url, data=json.dumps(body), headers=headers,
249 auth=('admin', 'admin'))
250 self.assertEqual(response.status_code, requests.codes.ok)
251 res = response.json()
252 self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
253 self.assertEqual(res['output']['configuration-response-common']['response-message'],
254 'Path is calculated by GNPy')
255 self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
256 self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
257 self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
258 self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
261 #Not found path by PCE and GNPy cannot find another one
262 def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
263 url = ("{}/operations/transportpce-pce:path-computation-request"
264 .format(self.restconf_baseurl))
266 "service-name": "service-4",
267 "resource-reserve": "true",
268 "pce-metric": "hop-count",
269 "service-handler-header": {
270 "request-id": "request-4"
273 "node-id": "XPONDER-1",
274 "service-rate": "100",
275 "service-format": "Ethernet",
279 "node-id": "XPONDER-4",
280 "service-rate": "100",
281 "service-format": "Ethernet",
284 "hard-constraints": {
290 "node-id": "OpenROADM-2"
296 "node-id": "OpenROADM-3"
302 "node-id": "OpenROADM-4"
308 "node-id": "OpenROADM-3"
316 headers = {'content-type': 'application/json',
317 "Accept": "application/json"}
318 response = requests.request(
319 "POST", url, data=json.dumps(body), headers=headers,
320 auth=('admin', 'admin'))
321 self.assertEqual(response.status_code, requests.codes.ok)
322 res = response.json()
323 self.assertEqual(res['output']['configuration-response-common']['response-code'], '500')
324 self.assertEqual(res['output']['configuration-response-common']['response-message'],
325 'No path available by PCE and GNPy ')
328 #Disconnect the different topologies
329 def test_08_disconnect_openroadmTopology(self):
330 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
331 .format(self.restconf_baseurl))
333 headers = {'content-type': 'application/json'}
334 response = requests.request(
335 "DELETE", url, data=json.dumps(data), headers=headers,
336 auth=('admin', 'admin'))
337 self.assertEqual(response.status_code, requests.codes.ok)
340 def test_09_disconnect_openroadmNetwork(self):
341 url = ("{}/config/ietf-network:networks/network/openroadm-network"
342 .format(self.restconf_baseurl))
344 headers = {'content-type': 'application/json'}
345 response = requests.request(
346 "DELETE", url, data=json.dumps(data), headers=headers,
347 auth=('admin', 'admin'))
348 self.assertEqual(response.status_code, requests.codes.ok)
351 def test_10_disconnect_clliNetwork(self):
352 url = ("{}/config/ietf-network:networks/network/clli-network"
353 .format(self.restconf_baseurl))
355 headers = {'content-type': 'application/json'}
356 response = requests.request(
357 "DELETE", url, data=json.dumps(data), headers=headers,
358 auth=('admin', 'admin'))
359 self.assertEqual(response.status_code, requests.codes.ok)
362 if __name__ == "__main__":
363 #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
364 unittest.main(verbosity=2)