add methods to manage ordm topo in func tests
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_gnpy.py
1 #!/usr/bin/env python
2
3 ##############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
5 #
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 ##############################################################################
11
12 import unittest
13 import json
14 import os
15 import sys
16 import time
17 import requests
18 from common import test_utils
19
20
21 class TransportGNPYtesting(unittest.TestCase):
22
23     @classmethod
24     def __init_logfile(cls):
25         GNPY_LOGFILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
26                                     "..", "..", "transportpce_tests", "gnpy.log")
27         if os.path.isfile(GNPY_LOFGILE):
28             os.remove(GNPY_LOFGILE)
29
30     topo_cllinet_data = None
31     topo_ordnet_data = None
32     topo_ordtopo_data = None
33     processes = None
34
35     @classmethod
36     def setUpClass(cls):
37         try:
38             sample_files_parsed = False
39             TOPO_CLLINET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
40                                              "..", "..", "sample_configs", "gnpy", "clliNetwork.json")
41             with open(TOPO_CLLINET_FILE, 'r') as topo_cllinet:
42                 cls.topo_cllinet_data = topo_cllinet.read()
43
44             TOPO_ORDNET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
45                                             "..", "..", "sample_configs", "gnpy", "openroadmNetwork.json")
46             with open(TOPO_ORDNET_FILE, 'r') as topo_ordnet:
47                 cls.topo_ordnet_data = topo_ordnet.read()
48
49             TOPO_ORDTOPO_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
50                                              "..", "..", "sample_configs", "gnpy", "openroadmTopology.json")
51             with open(TOPO_ORDTOPO_FILE, 'r') as topo_ordtopo:
52                 cls.topo_ordtopo_data = topo_ordtopo.read()
53             sample_files_parsed = True
54         except PermissionError as err:
55             print("Permission Error when trying to read sample files\n", err)
56             sys.exit(2)
57         except FileNotFoundError as err:
58             print("File Not found Error when trying to read sample files\n", err)
59             sys.exit(2)
60         except:
61             print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
62             sys.exit(2)
63         finally:
64             if sample_files_parsed:
65                 print("sample files content loaded")
66
67         cls.processes = test_utils.start_tpce()
68
69     @classmethod
70     def tearDownClass(cls):
71         for process in cls.processes:
72             test_utils.shutdown_process(process)
73         print("all processes killed")
74
75     def setUp(self):
76         time.sleep(2)
77
78     # Mount the different topologies
79     def test_01_connect_clliNetwork(self):
80         url = "{}/config/ietf-network:networks/network/clli-network"
81         response = test_utils.rawput_request(url, self.topo_cllinet_data)
82         self.assertEqual(response.status_code, requests.codes.ok)
83         time.sleep(3)
84
85     def test_02_connect_openroadmNetwork(self):
86         url = "{}/config/ietf-network:networks/network/openroadm-network"
87         response = test_utils.rawput_request(url, self.topo_ordnet_data)
88         self.assertEqual(response.status_code, requests.codes.ok)
89         time.sleep(3)
90
91     def test_03_connect_openroadmTopology(self):
92         response = test_utils.rawput_request(test_utils.URL_CONFIG_ORDM_TOPO, self.topo_ordtopo_data)
93         self.assertEqual(response.status_code, requests.codes.ok)
94         time.sleep(3)
95
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         data = {
100             "input": {
101                 "service-name": "service-1",
102                 "resource-reserve": "true",
103                 "pce-metric": "hop-count",
104                 "service-handler-header": {
105                     "request-id": "request-1"
106                 },
107                 "service-a-end": {
108                     "node-id": "XPONDER-1",
109                     "service-rate": "100",
110                     "service-format": "Ethernet",
111                     "clli": "Node1"
112                 },
113                 "service-z-end": {
114                     "node-id": "XPONDER-5",
115                     "service-rate": "100",
116                     "service-format": "Ethernet",
117                     "clli": "Node5"
118                 }
119             }
120         }
121         response = test_utils.post_request(url, data)
122         self.assertEqual(response.status_code, requests.codes.ok)
123         res = response.json()
124         self.assertEqual(res['output']['configuration-response-common'][
125             'response-code'], '200')
126         self.assertEqual(res['output']['configuration-response-common'][
127             'response-message'],
128             'Path is calculated by PCE')
129         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
130                          'A-to-Z')
131         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
132         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
133                          'Z-to-A')
134         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
135         time.sleep(5)
136
137     # Path computed by PCE is not feasible by GNPy and GNPy cannot find
138     # another one (low SNR)
139     def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
140         url = "{}/operations/transportpce-pce:path-computation-request"
141         data = {
142             "input": {
143                 "service-name": "service-2",
144                 "resource-reserve": "true",
145                 "pce-metric": "hop-count",
146                 "service-handler-header": {
147                     "request-id": "request-2"
148                 },
149                 "service-a-end": {
150                     "node-id": "XPONDER-1",
151                     "service-rate": "100",
152                     "service-format": "Ethernet",
153                     "clli": "Node1"
154                 },
155                 "service-z-end": {
156                     "node-id": "XPONDER-5",
157                     "service-rate": "100",
158                     "service-format": "Ethernet",
159                     "clli": "Node5"
160                 },
161                 "hard-constraints": {
162                     "include_": {
163                         "ordered-hops": [
164                             {
165                                 "hop-number": "0",
166                                 "hop-type": {
167                                     "node-id": "OpenROADM-2"
168                                 }
169                             },
170                             {
171                                 "hop-number": "1",
172                                 "hop-type": {
173                                     "node-id": "OpenROADM-3"
174                                 }
175                             },
176                             {
177                                 "hop-number": "2",
178                                 "hop-type": {
179                                     "node-id": "OpenROADM-4"
180                                 }
181                             }
182                         ]
183                     }
184                 }
185             }
186         }
187         response = test_utils.post_request(url, data)
188         self.assertEqual(response.status_code, requests.codes.ok)
189         res = response.json()
190         self.assertEqual(res['output']['configuration-response-common'][
191             'response-code'], '500')
192         self.assertEqual(res['output']['configuration-response-common'][
193             'response-message'],
194             'No path available by PCE and GNPy ')
195         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
196                          'A-to-Z')
197         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],
198                          False)
199         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
200                          'Z-to-A')
201         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],
202                          False)
203         time.sleep(5)
204
205     # #PCE cannot find a path while GNPy finds a feasible one
206     def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
207         url = "{}/operations/transportpce-pce:path-computation-request"
208         data = {
209             "input": {
210                 "service-name": "service-3",
211                 "resource-reserve": "true",
212                 "pce-metric": "hop-count",
213                 "service-handler-header": {
214                     "request-id": "request-3"
215                 },
216                 "service-a-end": {
217                     "node-id": "XPONDER-1",
218                     "service-rate": "100",
219                     "service-format": "Ethernet",
220                     "clli": "Node1"
221                 },
222                 "service-z-end": {
223                     "node-id": "XPONDER-4",
224                     "service-rate": "100",
225                     "service-format": "Ethernet",
226                     "clli": "Node5"
227                 },
228                 "hard-constraints": {
229                     "include_": {
230                         "ordered-hops": [
231                             {
232                                 "hop-number": "0",
233                                 "hop-type": {
234                                     "node-id": "OpenROADM-2"
235                                 }
236                             },
237                             {
238                                 "hop-number": "1",
239                                 "hop-type": {
240                                     "node-id": "OpenROADM-3"
241                                 }
242                             }
243                         ]
244                     }
245                 }
246             }
247         }
248         response = test_utils.post_request(url, data)
249         self.assertEqual(response.status_code, requests.codes.ok)
250         res = response.json()
251         self.assertEqual(res['output']['configuration-response-common'][
252             'response-code'], '200')
253         self.assertEqual(res['output']['configuration-response-common'][
254             'response-message'],
255             'Path is calculated by GNPy')
256         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
257                          'A-to-Z')
258         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
259         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
260                          'Z-to-A')
261         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
262         time.sleep(5)
263
264     # Not found path by PCE and GNPy cannot find another one
265     def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
266         url = "{}/operations/transportpce-pce:path-computation-request"
267         data = {
268             "input": {
269                 "service-name": "service-4",
270                 "resource-reserve": "true",
271                 "pce-metric": "hop-count",
272                 "service-handler-header": {
273                     "request-id": "request-4"
274                 },
275                 "service-a-end": {
276                     "node-id": "XPONDER-1",
277                     "service-rate": "100",
278                     "service-format": "Ethernet",
279                     "clli": "Node1"
280                 },
281                 "service-z-end": {
282                     "node-id": "XPONDER-4",
283                     "service-rate": "100",
284                     "service-format": "Ethernet",
285                     "clli": "Node5"
286                 },
287                 "hard-constraints": {
288                     "include_": {
289                         "ordered-hops": [
290                             {
291                                 "hop-number": "0",
292                                 "hop-type": {
293                                     "node-id": "OpenROADM-2"
294                                 }
295                             },
296                             {
297                                 "hop-number": "1",
298                                 "hop-type": {
299                                     "node-id": "OpenROADM-3"
300                                 }
301                             },
302                             {
303                                 "hop-number": "2",
304                                 "hop-type": {
305                                     "node-id": "OpenROADM-4"
306                                 }
307                             },
308                             {
309                                 "hop-number": "3",
310                                 "hop-type": {
311                                     "node-id": "OpenROADM-3"
312                                 }
313                             }
314                         ]
315                     }
316                 }
317             }
318         }
319         response = test_utils.post_request(url, data)
320         self.assertEqual(response.status_code, requests.codes.ok)
321         res = response.json()
322         self.assertEqual(res['output']['configuration-response-common'][
323             'response-code'], '500')
324         self.assertEqual(res['output']['configuration-response-common'][
325             'response-message'],
326             'No path available by PCE and GNPy ')
327         time.sleep(5)
328
329     # Disconnect the different topologies
330     def test_08_disconnect_openroadmTopology(self):
331         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
332         self.assertEqual(response.status_code, requests.codes.ok)
333         time.sleep(3)
334
335     def test_09_disconnect_openroadmNetwork(self):
336         url = "{}/config/ietf-network:networks/network/openroadm-network"
337         response = test_utils.delete_request(url)
338         self.assertEqual(response.status_code, requests.codes.ok)
339         time.sleep(3)
340
341     def test_10_disconnect_clliNetwork(self):
342         url = "{}/config/ietf-network:networks/network/clli-network"
343         response = test_utils.delete_request(url)
344         self.assertEqual(response.status_code, requests.codes.ok)
345         time.sleep(3)
346
347
348 if __name__ == "__main__":
349     # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
350     unittest.main(verbosity=2)