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
18 from functest.energy.energy import EnergyRecorder
19 import functest.energy.energy as energy
22 CASE_NAME = "UNIT_TEST_CASE"
23 STEP_NAME = "UNIT_TEST_STEP"
25 PREVIOUS_SCENARIO = "previous_scenario"
26 PREVIOUS_STEP = "previous_step"
29 class MockHttpResponse(object): # pylint: disable=too-few-public-methods
30 """Mock response for Energy recorder API."""
32 def __init__(self, text, status_code):
33 """Create an instance of MockHttpResponse."""
35 self.status_code = status_code
38 API_OK = MockHttpResponse(
42 API_KO = MockHttpResponse(
43 '{"message": "API-KO"}',
47 RECORDER_OK = MockHttpResponse(
48 '{"environment": "UNIT_TEST",'
50 ' "scenario": "' + CASE_NAME + '"}',
53 RECORDER_KO = MockHttpResponse(
54 '{"message": "An unhandled API exception occurred (MOCK)"}',
59 def config_loader_mock(config_key):
60 """Return mocked config values."""
61 if config_key == "energy_recorder.api_url":
62 return "http://pod-uri:8888"
63 elif config_key == "energy_recorder.api_user":
65 elif config_key == "energy_recorder.api_password":
68 raise Exception("Config not mocked")
71 def config_loader_mock_no_creds(config_key):
72 """Return mocked config values."""
73 if config_key == "energy_recorder.api_url":
74 return "http://pod-uri:8888"
75 elif config_key == "energy_recorder.api_user":
77 elif config_key == "energy_recorder.api_password":
80 raise Exception("Config not mocked:" + config_key)
83 class EnergyRecorderTest(unittest.TestCase):
84 """Energy module unitary test suite."""
87 request_headers = {'content-type': 'application/json'}
88 returned_value_to_preserve = "value"
89 exception_message_to_preserve = "exception_message"
91 @mock.patch('functest.energy.energy.requests.post',
92 return_value=RECORDER_OK)
93 def test_start(self, post_mock=None, get_mock=None):
94 """EnergyRecorder.start method (regular case)."""
95 self.test_load_config()
96 self.assertTrue(EnergyRecorder.start(self.case_name))
97 post_mock.assert_called_once_with(
98 EnergyRecorder.energy_recorder_api["uri"],
99 auth=EnergyRecorder.energy_recorder_api["auth"],
101 headers=self.request_headers,
102 timeout=EnergyRecorder.CONNECTION_TIMOUT
105 @mock.patch('functest.energy.energy.requests.post',
106 side_effect=Exception("Internal execution error (MOCK)"))
107 def test_start_error(self, post_mock=None):
108 """EnergyRecorder.start method (error in method)."""
109 self.test_load_config()
110 self.assertFalse(EnergyRecorder.start(self.case_name))
111 post_mock.assert_called_once_with(
112 EnergyRecorder.energy_recorder_api["uri"],
113 auth=EnergyRecorder.energy_recorder_api["auth"],
115 headers=self.request_headers,
116 timeout=EnergyRecorder.CONNECTION_TIMOUT
119 @mock.patch('functest.energy.energy.requests.post',
120 return_value=RECORDER_KO)
121 def test_start_api_error(self, post_mock=None):
122 """EnergyRecorder.start method (API error)."""
123 self.test_load_config()
124 self.assertFalse(EnergyRecorder.start(self.case_name))
125 post_mock.assert_called_once_with(
126 EnergyRecorder.energy_recorder_api["uri"],
127 auth=EnergyRecorder.energy_recorder_api["auth"],
129 headers=self.request_headers,
130 timeout=EnergyRecorder.CONNECTION_TIMOUT
133 @mock.patch('functest.energy.energy.requests.post',
134 return_value=RECORDER_OK)
135 def test_set_step(self, post_mock=None):
136 """EnergyRecorder.set_step method (regular case)."""
137 self.test_load_config()
138 self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
139 post_mock.assert_called_once_with(
140 EnergyRecorder.energy_recorder_api["uri"] + "/step",
141 auth=EnergyRecorder.energy_recorder_api["auth"],
143 headers=self.request_headers,
144 timeout=EnergyRecorder.CONNECTION_TIMOUT
147 @mock.patch('functest.energy.energy.requests.post',
148 return_value=RECORDER_KO)
149 def test_set_step_api_error(self, post_mock=None):
150 """EnergyRecorder.set_step method (API error)."""
151 self.test_load_config()
152 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
153 post_mock.assert_called_once_with(
154 EnergyRecorder.energy_recorder_api["uri"] + "/step",
155 auth=EnergyRecorder.energy_recorder_api["auth"],
157 headers=self.request_headers,
158 timeout=EnergyRecorder.CONNECTION_TIMOUT
161 @mock.patch('functest.energy.energy.requests.post',
162 side_effect=Exception("Internal execution error (MOCK)"))
163 def test_set_step_error(self, post_mock=None):
164 """EnergyRecorder.set_step method (method error)."""
165 self.test_load_config()
166 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
167 post_mock.assert_called_once_with(
168 EnergyRecorder.energy_recorder_api["uri"] + "/step",
169 auth=EnergyRecorder.energy_recorder_api["auth"],
171 headers=self.request_headers,
172 timeout=EnergyRecorder.CONNECTION_TIMOUT
175 @mock.patch('functest.energy.energy.requests.delete',
176 return_value=RECORDER_OK)
177 def test_stop(self, delete_mock=None):
178 """EnergyRecorder.stop method (regular case)."""
179 self.test_load_config()
180 self.assertTrue(EnergyRecorder.stop())
181 delete_mock.assert_called_once_with(
182 EnergyRecorder.energy_recorder_api["uri"],
183 auth=EnergyRecorder.energy_recorder_api["auth"],
184 headers=self.request_headers,
185 timeout=EnergyRecorder.CONNECTION_TIMOUT
188 @mock.patch('functest.energy.energy.requests.delete',
189 return_value=RECORDER_KO)
190 def test_stop_api_error(self, delete_mock=None):
191 """EnergyRecorder.stop method (API Error)."""
192 self.test_load_config()
193 self.assertFalse(EnergyRecorder.stop())
194 delete_mock.assert_called_once_with(
195 EnergyRecorder.energy_recorder_api["uri"],
196 auth=EnergyRecorder.energy_recorder_api["auth"],
197 headers=self.request_headers,
198 timeout=EnergyRecorder.CONNECTION_TIMOUT
201 @mock.patch('functest.energy.energy.requests.delete',
202 side_effect=Exception("Internal execution error (MOCK)"))
203 def test_stop_error(self, delete_mock=None):
204 """EnergyRecorder.stop method (method error)."""
205 self.test_load_config()
206 self.assertFalse(EnergyRecorder.stop())
207 delete_mock.assert_called_once_with(
208 EnergyRecorder.energy_recorder_api["uri"],
209 auth=EnergyRecorder.energy_recorder_api["auth"],
210 headers=self.request_headers,
211 timeout=EnergyRecorder.CONNECTION_TIMOUT
214 @energy.enable_recording
215 def __decorated_method(self):
216 """Call with to energy recorder decorators."""
217 return self.returned_value_to_preserve
219 @energy.enable_recording
220 def __decorated_method_with_ex(self):
221 """Call with to energy recorder decorators."""
222 raise Exception(self.exception_message_to_preserve)
224 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
226 @mock.patch("functest.energy.energy.EnergyRecorder")
227 def test_decorators(self,
229 cur_scenario_mock=None):
230 """Test energy module decorators."""
231 self.__decorated_method()
232 calls = [mock.call.start(self.case_name),
234 recorder_mock.assert_has_calls(calls)
236 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
237 return_value={"scenario": PREVIOUS_SCENARIO,
238 "step": PREVIOUS_STEP})
239 @mock.patch("functest.energy.energy.EnergyRecorder")
240 @mock.patch("functest.utils.functest_utils.get_pod_name",
241 return_value="MOCK_POD")
242 @mock.patch("functest.utils.functest_utils.get_functest_config",
243 side_effect=config_loader_mock)
244 def test_decorators_with_previous(self,
248 cur_scenario_mock=None):
249 """Test energy module decorators."""
250 self.__decorated_method()
251 calls = [mock.call.start(self.case_name),
252 mock.call.submit_scenario(PREVIOUS_SCENARIO,
254 recorder_mock.assert_has_calls(calls)
256 def test_decorator_preserve_return(self):
257 """Test that decorator preserve method returned value."""
258 self.test_load_config()
260 self.__decorated_method() == self.returned_value_to_preserve
264 "functest.energy.energy.finish_session")
265 def test_decorator_preserve_ex(self, finish_mock=None):
266 """Test that decorator preserve method exceptions."""
267 self.test_load_config()
268 with self.assertRaises(Exception) as context:
269 self.__decorated_method_with_ex()
271 self.exception_message_to_preserve in context.exception
273 self.assertTrue(finish_mock.called)
275 @mock.patch("functest.utils.functest_utils.get_functest_config",
276 side_effect=config_loader_mock)
277 @mock.patch("functest.utils.functest_utils.get_pod_name",
278 return_value="MOCK_POD")
279 @mock.patch("functest.energy.energy.requests.get",
281 def test_load_config(self, loader_mock=None, pod_mock=None,
283 """Test load config."""
284 EnergyRecorder.energy_recorder_api = None
285 EnergyRecorder.load_config()
288 EnergyRecorder.energy_recorder_api["auth"],
292 EnergyRecorder.energy_recorder_api["uri"],
293 "http://pod-uri:8888/recorders/environment/MOCK_POD"
296 @mock.patch("functest.utils.functest_utils.get_functest_config",
297 side_effect=config_loader_mock_no_creds)
298 @mock.patch("functest.utils.functest_utils.get_pod_name",
299 return_value="MOCK_POD")
300 @mock.patch("functest.energy.energy.requests.get",
302 def test_load_config_no_creds(self, loader_mock=None, pod_mock=None,
304 """Test load config without creds."""
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.utils.functest_utils.get_functest_config",
315 @mock.patch("functest.utils.functest_utils.get_pod_name",
316 return_value="MOCK_POD")
317 @mock.patch("functest.energy.energy.requests.get",
319 def test_load_config_ex(self, loader_mock=None, pod_mock=None,
321 """Test load config with exception."""
322 with self.assertRaises(AssertionError):
323 EnergyRecorder.energy_recorder_api = None
324 EnergyRecorder.load_config()
325 self.assertEquals(EnergyRecorder.energy_recorder_api, None)
327 @mock.patch("functest.utils.functest_utils.get_functest_config",
328 side_effect=config_loader_mock)
329 @mock.patch("functest.utils.functest_utils.get_pod_name",
330 return_value="MOCK_POD")
331 @mock.patch("functest.energy.energy.requests.get",
333 def test_load_config_api_ko(self, loader_mock=None, pod_mock=None,
335 """Test load config with API unavailable."""
336 EnergyRecorder.energy_recorder_api = None
337 EnergyRecorder.load_config()
338 self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
341 @mock.patch("functest.utils.functest_utils.get_functest_config",
343 @mock.patch("functest.utils.functest_utils.get_pod_name",
344 return_value="MOCK_POD")
345 @mock.patch('functest.energy.energy.requests.get',
346 return_value=RECORDER_OK)
347 def test_get_current_scenario(self, loader_mock=None,
348 pod_mock=None, get_mock=None):
349 """Test get_current_scenario."""
350 self.test_load_config()
351 scenario = EnergyRecorder.get_current_scenario()
352 self.assertTrue(scenario is not None)
355 if __name__ == "__main__":
356 logging.disable(logging.CRITICAL)
357 unittest.main(verbosity=2)