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
20 from functest.energy.energy import EnergyRecorder
21 import functest.energy.energy as energy
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 # pylint: disable=too-many-public-methods
65 class EnergyRecorderTest(unittest.TestCase):
66 """Energy module unitary test suite."""
69 request_headers = {'content-type': 'application/json'}
70 returned_value_to_preserve = "value"
71 exception_message_to_preserve = "exception_message"
75 """Set config values."""
76 os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
77 os.environ["ENERGY_RECORDER_API_USER"] = "user"
78 os.environ["ENERGY_RECORDER_API_PASSWORD"] = "password"
81 def _set_env_nocreds():
82 """Set config values."""
83 os.environ["ENERGY_RECORDER_API_URL"] = "http://pod-uri:8888"
84 del os.environ["ENERGY_RECORDER_API_USER"]
85 del os.environ["ENERGY_RECORDER_API_PASSWORD"]
87 @mock.patch('functest.energy.energy.requests.post',
88 return_value=RECORDER_OK)
89 def test_start(self, post_mock=None, get_mock=None):
90 """EnergyRecorder.start method (regular case)."""
91 self.test_load_config()
92 self.assertTrue(EnergyRecorder.start(self.case_name))
93 post_mock.assert_called_once_with(
94 EnergyRecorder.energy_recorder_api["uri"],
95 auth=EnergyRecorder.energy_recorder_api["auth"],
97 headers=self.request_headers,
98 timeout=EnergyRecorder.CONNECTION_TIMEOUT
101 @mock.patch('functest.energy.energy.requests.post',
102 side_effect=Exception("Internal execution error (MOCK)"))
103 def test_start_error(self, post_mock=None):
104 """EnergyRecorder.start method (error in method)."""
105 self.test_load_config()
106 self.assertFalse(EnergyRecorder.start(self.case_name))
107 post_mock.assert_called_once_with(
108 EnergyRecorder.energy_recorder_api["uri"],
109 auth=EnergyRecorder.energy_recorder_api["auth"],
111 headers=self.request_headers,
112 timeout=EnergyRecorder.CONNECTION_TIMEOUT
115 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
116 side_effect=Exception("Internal execution error (MOCK)"))
117 def test_start_exception(self, conf_loader_mock=None):
118 """EnergyRecorder.start test with exception during execution."""
119 start_status = EnergyRecorder.start(CASE_NAME)
120 self.assertFalse(start_status)
122 @mock.patch('functest.energy.energy.requests.post',
123 return_value=RECORDER_KO)
124 def test_start_api_error(self, post_mock=None):
125 """EnergyRecorder.start method (API error)."""
126 self.test_load_config()
127 self.assertFalse(EnergyRecorder.start(self.case_name))
128 post_mock.assert_called_once_with(
129 EnergyRecorder.energy_recorder_api["uri"],
130 auth=EnergyRecorder.energy_recorder_api["auth"],
132 headers=self.request_headers,
133 timeout=EnergyRecorder.CONNECTION_TIMEOUT
136 @mock.patch('functest.energy.energy.requests.post',
137 return_value=RECORDER_OK)
138 def test_set_step(self, post_mock=None):
139 """EnergyRecorder.set_step method (regular case)."""
140 self.test_load_config()
141 self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
142 post_mock.assert_called_once_with(
143 EnergyRecorder.energy_recorder_api["uri"] + "/step",
144 auth=EnergyRecorder.energy_recorder_api["auth"],
146 headers=self.request_headers,
147 timeout=EnergyRecorder.CONNECTION_TIMEOUT
150 @mock.patch('functest.energy.energy.requests.post',
151 return_value=RECORDER_KO)
152 def test_set_step_api_error(self, post_mock=None):
153 """EnergyRecorder.set_step method (API error)."""
154 self.test_load_config()
155 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
156 post_mock.assert_called_once_with(
157 EnergyRecorder.energy_recorder_api["uri"] + "/step",
158 auth=EnergyRecorder.energy_recorder_api["auth"],
160 headers=self.request_headers,
161 timeout=EnergyRecorder.CONNECTION_TIMEOUT
164 @mock.patch('functest.energy.energy.requests.post',
165 side_effect=Exception("Internal execution error (MOCK)"))
166 def test_set_step_error(self, post_mock=None):
167 """EnergyRecorder.set_step method (method error)."""
168 self.test_load_config()
169 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
170 post_mock.assert_called_once_with(
171 EnergyRecorder.energy_recorder_api["uri"] + "/step",
172 auth=EnergyRecorder.energy_recorder_api["auth"],
174 headers=self.request_headers,
175 timeout=EnergyRecorder.CONNECTION_TIMEOUT
178 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
179 side_effect=requests.exceptions.ConnectionError())
180 def test_set_step_connection_error(self, conf_loader_mock=None):
181 """EnergyRecorder.start test with exception during execution."""
182 step_status = EnergyRecorder.set_step(STEP_NAME)
183 self.assertFalse(step_status)
185 @mock.patch('functest.energy.energy.requests.delete',
186 return_value=RECORDER_OK)
187 def test_stop(self, delete_mock=None):
188 """EnergyRecorder.stop method (regular case)."""
189 self.test_load_config()
190 self.assertTrue(EnergyRecorder.stop())
191 delete_mock.assert_called_once_with(
192 EnergyRecorder.energy_recorder_api["uri"],
193 auth=EnergyRecorder.energy_recorder_api["auth"],
194 headers=self.request_headers,
195 timeout=EnergyRecorder.CONNECTION_TIMEOUT
198 @mock.patch('functest.energy.energy.requests.delete',
199 return_value=RECORDER_KO)
200 def test_stop_api_error(self, delete_mock=None):
201 """EnergyRecorder.stop method (API Error)."""
202 self.test_load_config()
203 self.assertFalse(EnergyRecorder.stop())
204 delete_mock.assert_called_once_with(
205 EnergyRecorder.energy_recorder_api["uri"],
206 auth=EnergyRecorder.energy_recorder_api["auth"],
207 headers=self.request_headers,
208 timeout=EnergyRecorder.CONNECTION_TIMEOUT
211 @mock.patch('functest.energy.energy.requests.delete',
212 side_effect=Exception("Internal execution error (MOCK)"))
213 def test_stop_error(self, delete_mock=None):
214 """EnergyRecorder.stop method (method error)."""
215 self.test_load_config()
216 self.assertFalse(EnergyRecorder.stop())
217 delete_mock.assert_called_once_with(
218 EnergyRecorder.energy_recorder_api["uri"],
219 auth=EnergyRecorder.energy_recorder_api["auth"],
220 headers=self.request_headers,
221 timeout=EnergyRecorder.CONNECTION_TIMEOUT
224 @energy.enable_recording
225 def __decorated_method(self):
226 """Call with to energy recorder decorators."""
227 return self.returned_value_to_preserve
229 @energy.enable_recording
230 def __decorated_method_with_ex(self):
231 """Call with to energy recorder decorators."""
232 raise Exception(self.exception_message_to_preserve)
234 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
236 @mock.patch("functest.energy.energy.EnergyRecorder")
237 def test_decorators(self,
239 cur_scenario_mock=None):
240 """Test energy module decorators."""
241 self.__decorated_method()
242 calls = [mock.call.start(self.case_name),
244 recorder_mock.assert_has_calls(calls)
246 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
247 return_value={"scenario": PREVIOUS_SCENARIO,
248 "step": PREVIOUS_STEP})
249 @mock.patch("functest.energy.energy.EnergyRecorder")
250 def test_decorators_with_previous(self,
252 cur_scenario_mock=None):
253 """Test energy module decorators."""
254 os.environ['NODE_NAME'] = 'MOCK_POD'
255 self._set_env_creds()
256 self.__decorated_method()
257 calls = [mock.call.start(self.case_name),
258 mock.call.submit_scenario(PREVIOUS_SCENARIO,
260 recorder_mock.assert_has_calls(calls, True)
262 def test_decorator_preserve_return(self):
263 """Test that decorator preserve method returned value."""
264 self.test_load_config()
266 self.__decorated_method() == self.returned_value_to_preserve
270 "functest.energy.energy.finish_session")
271 def test_decorator_preserve_ex(self, finish_mock=None):
272 """Test that decorator preserve method exceptions."""
273 self.test_load_config()
274 with self.assertRaises(Exception) as context:
275 self.__decorated_method_with_ex()
277 self.exception_message_to_preserve in str(context.exception)
279 self.assertTrue(finish_mock.called)
281 @mock.patch("functest.energy.energy.requests.get",
283 def test_load_config(self, loader_mock=None, get_mock=None):
284 """Test load config."""
285 os.environ['NODE_NAME'] = 'MOCK_POD'
286 self._set_env_creds()
287 EnergyRecorder.energy_recorder_api = None
288 EnergyRecorder.load_config()
291 EnergyRecorder.energy_recorder_api["auth"],
295 EnergyRecorder.energy_recorder_api["uri"],
296 "http://pod-uri:8888/recorders/environment/MOCK_POD"
299 @mock.patch("functest.energy.energy.requests.get",
301 def test_load_config_no_creds(self, loader_mock=None, get_mock=None):
302 """Test load config without creds."""
303 os.environ['NODE_NAME'] = 'MOCK_POD'
304 self._set_env_nocreds()
305 EnergyRecorder.energy_recorder_api = None
306 EnergyRecorder.load_config()
307 self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
309 EnergyRecorder.energy_recorder_api["uri"],
310 "http://pod-uri:8888/recorders/environment/MOCK_POD"
313 @mock.patch("functest.energy.energy.requests.get",
315 def test_load_config_ex(self, loader_mock=None, get_mock=None):
316 """Test load config with exception."""
317 for key in ['NODE_NAME', 'ENERGY_RECORDER_API_URL']:
319 with self.assertRaises(AssertionError):
320 EnergyRecorder.energy_recorder_api = None
321 EnergyRecorder.load_config()
322 self.assertEquals(EnergyRecorder.energy_recorder_api, None)
324 @mock.patch("functest.energy.energy.requests.get",
326 def test_load_config_api_ko(self, loader_mock=None, get_mock=None):
327 """Test load config with API unavailable."""
328 os.environ['NODE_NAME'] = 'MOCK_POD'
329 self._set_env_creds()
330 EnergyRecorder.energy_recorder_api = None
331 EnergyRecorder.load_config()
332 self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
335 @mock.patch('functest.energy.energy.requests.get',
336 return_value=RECORDER_OK)
337 def test_get_current_scenario(self, loader_mock=None, get_mock=None):
338 """Test get_current_scenario."""
339 os.environ['NODE_NAME'] = 'MOCK_POD'
340 self.test_load_config()
341 scenario = EnergyRecorder.get_current_scenario()
342 self.assertTrue(scenario is not None)
344 @mock.patch('functest.energy.energy.requests.get',
345 return_value=RECORDER_NOT_FOUND)
346 def test_current_scenario_not_found(self, get_mock=None):
347 """Test get current scenario not existing."""
348 os.environ['NODE_NAME'] = 'MOCK_POD'
349 self.test_load_config()
350 scenario = EnergyRecorder.get_current_scenario()
351 self.assertTrue(scenario is None)
353 @mock.patch('functest.energy.energy.requests.get',
354 return_value=RECORDER_KO)
355 def test_current_scenario_api_error(self, get_mock=None):
356 """Test get current scenario with API error."""
357 os.environ['NODE_NAME'] = 'MOCK_POD'
358 self.test_load_config()
359 scenario = EnergyRecorder.get_current_scenario()
360 self.assertTrue(scenario is None)
362 @mock.patch('functest.energy.energy.EnergyRecorder.load_config',
363 side_effect=Exception("Internal execution error (MOCK)"))
364 def test_current_scenario_exception(self, get_mock=None):
365 """Test get current scenario with exception."""
366 scenario = EnergyRecorder.get_current_scenario()
367 self.assertTrue(scenario is None)
369 if __name__ == "__main__":
370 logging.disable(logging.CRITICAL)
371 unittest.main(verbosity=2)