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
20 from functest.utils.constants import CONST
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_functest_config",
241 side_effect=config_loader_mock)
242 def test_decorators_with_previous(self,
245 cur_scenario_mock=None):
246 """Test energy module decorators."""
247 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
248 self.__decorated_method()
249 calls = [mock.call.start(self.case_name),
250 mock.call.submit_scenario(PREVIOUS_SCENARIO,
252 recorder_mock.assert_has_calls(calls)
254 def test_decorator_preserve_return(self):
255 """Test that decorator preserve method returned value."""
256 self.test_load_config()
258 self.__decorated_method() == self.returned_value_to_preserve
262 "functest.energy.energy.finish_session")
263 def test_decorator_preserve_ex(self, finish_mock=None):
264 """Test that decorator preserve method exceptions."""
265 self.test_load_config()
266 with self.assertRaises(Exception) as context:
267 self.__decorated_method_with_ex()
269 self.exception_message_to_preserve in context.exception
271 self.assertTrue(finish_mock.called)
273 @mock.patch("functest.utils.functest_utils.get_functest_config",
274 side_effect=config_loader_mock)
275 @mock.patch("functest.energy.energy.requests.get",
277 def test_load_config(self, loader_mock=None, get_mock=None):
278 """Test load config."""
279 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
280 EnergyRecorder.energy_recorder_api = None
281 EnergyRecorder.load_config()
284 EnergyRecorder.energy_recorder_api["auth"],
288 EnergyRecorder.energy_recorder_api["uri"],
289 "http://pod-uri:8888/recorders/environment/MOCK_POD"
292 @mock.patch("functest.utils.functest_utils.get_functest_config",
293 side_effect=config_loader_mock_no_creds)
294 @mock.patch("functest.energy.energy.requests.get",
296 def test_load_config_no_creds(self, loader_mock=None, get_mock=None):
297 """Test load config without creds."""
298 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
299 EnergyRecorder.energy_recorder_api = None
300 EnergyRecorder.load_config()
301 self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
303 EnergyRecorder.energy_recorder_api["uri"],
304 "http://pod-uri:8888/recorders/environment/MOCK_POD"
307 @mock.patch("functest.utils.functest_utils.get_functest_config",
309 @mock.patch("functest.energy.energy.requests.get",
311 def test_load_config_ex(self, loader_mock=None, get_mock=None):
312 """Test load config with exception."""
313 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
314 with self.assertRaises(AssertionError):
315 EnergyRecorder.energy_recorder_api = None
316 EnergyRecorder.load_config()
317 self.assertEquals(EnergyRecorder.energy_recorder_api, None)
319 @mock.patch("functest.utils.functest_utils.get_functest_config",
320 side_effect=config_loader_mock)
321 @mock.patch("functest.energy.energy.requests.get",
323 def test_load_config_api_ko(self, loader_mock=None, get_mock=None):
324 """Test load config with API unavailable."""
325 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
326 EnergyRecorder.energy_recorder_api = None
327 EnergyRecorder.load_config()
328 self.assertEquals(EnergyRecorder.energy_recorder_api["available"],
331 @mock.patch("functest.utils.functest_utils.get_functest_config",
333 @mock.patch('functest.energy.energy.requests.get',
334 return_value=RECORDER_OK)
335 def test_get_current_scenario(self, loader_mock=None, get_mock=None):
336 """Test get_current_scenario."""
337 CONST.__setattr__('NODE_NAME', 'MOCK_POD')
338 self.test_load_config()
339 scenario = EnergyRecorder.get_current_scenario()
340 self.assertTrue(scenario is not None)
343 if __name__ == "__main__":
344 logging.disable(logging.CRITICAL)
345 unittest.main(verbosity=2)