2 # -*- coding: UTF-8 -*-
4 # Copyright (c) 2017 Orange and others.
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
11 """Unitary test for energy module."""
12 # pylint: disable=unused-argument
19 from functest.energy.energy import EnergyRecorder
20 import functest.energy.energy as energy
21 from functest.utils.constants import CONST
23 CASE_NAME = "UNIT_TEST_CASE"
24 STEP_NAME = "UNIT_TEST_STEP"
26 PREVIOUS_SCENARIO = "previous_scenario"
27 PREVIOUS_STEP = "previous_step"
30 class MockHttpResponse(object): # pylint: disable=too-few-public-methods
31 """Mock response for Energy recorder API."""
33 def __init__(self, text, status_code):
34 """Create an instance of MockHttpResponse."""
36 self.status_code = status_code
39 API_OK = MockHttpResponse(
43 API_KO = MockHttpResponse(
44 '{"message": "API-KO"}',
48 RECORDER_OK = MockHttpResponse(
49 '{"environment": "UNIT_TEST",'
51 ' "scenario": "' + CASE_NAME + '"}',
54 RECORDER_KO = MockHttpResponse(
55 '{"message": "An unhandled API exception occurred (MOCK)"}',
58 RECORDER_NOT_FOUND = MockHttpResponse(
59 '{"message": "Recorder not found (MOCK)"}',
64 def config_loader_mock(config_key):
65 """Return mocked config values."""
66 if config_key == "energy_recorder.api_url":
67 return "http://pod-uri:8888"
68 elif config_key == "energy_recorder.api_user":
70 elif config_key == "energy_recorder.api_password":
74 def config_loader_mock_no_creds(config_key):
75 """Return mocked config values."""
76 if config_key == "energy_recorder.api_url":
77 return "http://pod-uri:8888"
78 elif config_key == "energy_recorder.api_user":
80 elif config_key == "energy_recorder.api_password":
84 # pylint: disable=too-many-public-methods
85 class EnergyRecorderTest(unittest.TestCase):
86 """Energy module unitary test suite."""
89 request_headers = {'content-type': 'application/json'}
90 returned_value_to_preserve = "value"
91 exception_message_to_preserve = "exception_message"
93 @mock.patch('functest.energy.energy.requests.post',
94 return_value=RECORDER_OK)
95 def test_start(self, post_mock=None, get_mock=None):
96 """EnergyRecorder.start method (regular case)."""
97 self.test_load_config()
98 self.assertTrue(EnergyRecorder.start(self.case_name))
99 post_mock.assert_called_once_with(
100 EnergyRecorder.energy_recorder_api["uri"],
101 auth=EnergyRecorder.energy_recorder_api["auth"],
103 headers=self.request_headers,
104 timeout=EnergyRecorder.CONNECTION_TIMEOUT
107 @mock.patch('functest.energy.energy.requests.post',
108 side_effect=Exception("Internal execution error (MOCK)"))
109 def test_start_error(self, post_mock=None):
110 """EnergyRecorder.start method (error in method)."""
111 self.test_load_config()
112 self.assertFalse(EnergyRecorder.start(self.case_name))
113 post_mock.assert_called_once_with(
114 EnergyRecorder.energy_recorder_api["uri"],
115 auth=EnergyRecorder.energy_recorder_api["auth"],
117 headers=self.request_headers,
118 timeout=EnergyRecorder.CONNECTION_TIMEOUT
121 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
122 side_effect=Exception("Internal execution error (MOCK)"))
123 def test_start_exception(self, conf_loader_mock=None):
124 """EnergyRecorder.start test with exception during execution."""
125 start_status = EnergyRecorder.start(CASE_NAME)
126 self.assertFalse(start_status)
128 @mock.patch('functest.energy.energy.requests.post',
129 return_value=RECORDER_KO)
130 def test_start_api_error(self, post_mock=None):
131 """EnergyRecorder.start method (API error)."""
132 self.test_load_config()
133 self.assertFalse(EnergyRecorder.start(self.case_name))
134 post_mock.assert_called_once_with(
135 EnergyRecorder.energy_recorder_api["uri"],
136 auth=EnergyRecorder.energy_recorder_api["auth"],
138 headers=self.request_headers,
139 timeout=EnergyRecorder.CONNECTION_TIMEOUT
142 @mock.patch('functest.energy.energy.requests.post',
143 return_value=RECORDER_OK)
144 def test_set_step(self, post_mock=None):
145 """EnergyRecorder.set_step method (regular case)."""
146 self.test_load_config()
147 self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
148 post_mock.assert_called_once_with(
149 EnergyRecorder.energy_recorder_api["uri"] + "/step",
150 auth=EnergyRecorder.energy_recorder_api["auth"],
152 headers=self.request_headers,
153 timeout=EnergyRecorder.CONNECTION_TIMEOUT
156 @mock.patch('functest.energy.energy.requests.post',
157 return_value=RECORDER_KO)
158 def test_set_step_api_error(self, post_mock=None):
159 """EnergyRecorder.set_step method (API error)."""
160 self.test_load_config()
161 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
162 post_mock.assert_called_once_with(
163 EnergyRecorder.energy_recorder_api["uri"] + "/step",
164 auth=EnergyRecorder.energy_recorder_api["auth"],
166 headers=self.request_headers,
167 timeout=EnergyRecorder.CONNECTION_TIMEOUT
170 @mock.patch('functest.energy.energy.requests.post',
171 side_effect=Exception("Internal execution error (MOCK)"))
172 def test_set_step_error(self, post_mock=None):
173 """EnergyRecorder.set_step method (method error)."""
174 self.test_load_config()
175 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
176 post_mock.assert_called_once_with(
177 EnergyRecorder.energy_recorder_api["uri"] + "/step",
178 auth=EnergyRecorder.energy_recorder_api["auth"],
180 headers=self.request_headers,
181 timeout=EnergyRecorder.CONNECTION_TIMEOUT
184 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
185 side_effect=requests.exceptions.ConnectionError())
186 def test_set_step_connection_error(self, conf_loader_mock=None):
187 """EnergyRecorder.start test with exception during execution."""
188 step_status = EnergyRecorder.set_step(STEP_NAME)
189 self.assertFalse(step_status)
191 @mock.patch('functest.energy.energy.requests.delete',
192 return_value=RECORDER_OK)
193 def test_stop(self, delete_mock=None):
194 """EnergyRecorder.stop method (regular case)."""
195 self.test_load_config()
196 self.assertTrue(EnergyRecorder.stop())
197 delete_mock.assert_called_once_with(
198 EnergyRecorder.energy_recorder_api["uri"],
199 auth=EnergyRecorder.energy_recorder_api["auth"],
200 headers=self.request_headers,
201 timeout=EnergyRecorder.CONNECTION_TIMEOUT
204 @mock.patch('functest.energy.energy.requests.delete',
205 return_value=RECORDER_KO)
206 def test_stop_api_error(self, delete_mock=None):
207 """EnergyRecorder.stop method (API Error)."""
208 self.test_load_config()
209 self.assertFalse(EnergyRecorder.stop())
210 delete_mock.assert_called_once_with(
211 EnergyRecorder.energy_recorder_api["uri"],
212 auth=EnergyRecorder.energy_recorder_api["auth"],
213 headers=self.request_headers,
214 timeout=EnergyRecorder.CONNECTION_TIMEOUT
217 @mock.patch('functest.energy.energy.requests.delete',
218 side_effect=Exception("Internal execution error (MOCK)"))
219 def test_stop_error(self, delete_mock=None):
220 """EnergyRecorder.stop method (method error)."""
221 self.test_load_config()
222 self.assertFalse(EnergyRecorder.stop())
223 delete_mock.assert_called_once_with(
224 EnergyRecorder.energy_recorder_api["uri"],
225 auth=EnergyRecorder.energy_recorder_api["auth"],
226 headers=self.request_headers,
227 timeout=EnergyRecorder.CONNECTION_TIMEOUT
230 @energy.enable_recording
231 def __decorated_method(self):
232 """Call with to energy recorder decorators."""
233 return self.returned_value_to_preserve
235 @energy.enable_recording
236 def __decorated_method_with_ex(self):
237 """Call with to energy recorder decorators."""
238 raise Exception(self.exception_message_to_preserve)
240 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
242 @mock.patch("functest.energy.energy.EnergyRecorder")
243 def test_decorators(self,
245 cur_scenario_mock=None):
246 """Test energy module decorators."""
247 self.__decorated_method()
248 calls = [mock.call.start(self.case_name),
250 recorder_mock.assert_has_calls(calls)
252 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
253 return_value={"scenario": PREVIOUS_SCENARIO,
254 "step": PREVIOUS_STEP})
255 @mock.patch("functest.energy.energy.EnergyRecorder")
256 @mock.patch("functest.utils.functest_utils.get_functest_config",
257 side_effect=config_loader_mock)
258 def test_decorators_with_previous(self,
261 cur_scenario_mock=None):
262 """Test energy module decorators."""
263 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
264 self.__decorated_method()
265 calls = [mock.call.start(self.case_name),
266 mock.call.submit_scenario(PREVIOUS_SCENARIO,
268 recorder_mock.assert_has_calls(calls, True)
270 def test_decorator_preserve_return(self):
271 """Test that decorator preserve method returned value."""
272 self.test_load_config()
274 self.__decorated_method() == self.returned_value_to_preserve
278 "functest.energy.energy.finish_session")
279 def test_decorator_preserve_ex(self, finish_mock=None):
280 """Test that decorator preserve method exceptions."""
281 self.test_load_config()
282 with self.assertRaises(Exception) as context:
283 self.__decorated_method_with_ex()
285 self.exception_message_to_preserve in str(context.exception)
287 self.assertTrue(finish_mock.called)
289 @mock.patch("functest.utils.functest_utils.get_functest_config",
290 side_effect=config_loader_mock)
291 @mock.patch("functest.energy.energy.requests.get",
293 def test_load_config(self, loader_mock=None, get_mock=None):
294 """Test load config."""
295 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
296 EnergyRecorder.energy_recorder_api = None
297 EnergyRecorder.load_config()
300 EnergyRecorder.energy_recorder_api["auth"],
304 EnergyRecorder.energy_recorder_api["uri"],
305 "http://pod-uri:8888/recorders/environment/MOCK_POD"
308 @mock.patch("functest.utils.functest_utils.get_functest_config",
309 side_effect=config_loader_mock_no_creds)
310 @mock.patch("functest.energy.energy.requests.get",
312 def test_load_config_no_creds(self, loader_mock=None, get_mock=None):
313 """Test load config without creds."""
314 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
315 EnergyRecorder.energy_recorder_api = None
316 EnergyRecorder.load_config()
317 self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
319 EnergyRecorder.energy_recorder_api["uri"],
320 "http://pod-uri:8888/recorders/environment/MOCK_POD"
323 @mock.patch("functest.utils.functest_utils.get_functest_config",
325 @mock.patch("functest.energy.energy.requests.get",
327 def test_load_config_ex(self, loader_mock=None, get_mock=None):
328 """Test load config with exception."""
329 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
330 with self.assertRaises(AssertionError):
331 EnergyRecorder.energy_recorder_api = None
332 EnergyRecorder.load_config()
333 self.assertEquals(EnergyRecorder.energy_recorder_api, None)
335 @mock.patch("functest.utils.functest_utils.get_functest_config",
336 side_effect=config_loader_mock)
337 @mock.patch("functest.energy.energy.requests.get",
339 def test_load_config_api_ko(self, loader_mock=None, get_mock=None):
340 """Test load config with API unavailable."""
341 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
342 EnergyRecorder.energy_recorder_api = None
343 EnergyRecorder.load_config()
344 self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
347 @mock.patch("functest.utils.functest_utils.get_functest_config",
349 @mock.patch('functest.energy.energy.requests.get',
350 return_value=RECORDER_OK)
351 def test_get_current_scenario(self, loader_mock=None, get_mock=None):
352 """Test get_current_scenario."""
353 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
354 self.test_load_config()
355 scenario = EnergyRecorder.get_current_scenario()
356 self.assertTrue(scenario is not None)
358 @mock.patch('functest.energy.energy.requests.get',
359 return_value=RECORDER_NOT_FOUND)
360 def test_current_scenario_not_found(self, get_mock=None):
361 """Test get current scenario not existing."""
362 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
363 self.test_load_config()
364 scenario = EnergyRecorder.get_current_scenario()
365 self.assertTrue(scenario is None)
367 @mock.patch('functest.energy.energy.requests.get',
368 return_value=RECORDER_KO)
369 def test_current_scenario_api_error(self, get_mock=None):
370 """Test get current scenario with API error."""
371 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
372 self.test_load_config()
373 scenario = EnergyRecorder.get_current_scenario()
374 self.assertTrue(scenario is None)
376 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
377 side_effect=Exception("Internal execution error (MOCK)"))
378 def test_current_scenario_exception(self, get_mock=None):
379 """Test get current scenario with exception."""
380 scenario = EnergyRecorder.get_current_scenario()
381 self.assertTrue(scenario is None)
383 if __name__ == "__main__":
384 logging.disable(logging.CRITICAL)
385 unittest.main(verbosity=2)