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 RECORDER_OK = MockHttpResponse(
39 '{"environment": "UNIT_TEST",'
41 ' "scenario": "' + CASE_NAME + '"}',
44 RECORDER_KO = MockHttpResponse(
45 '{"message": "An unhandled API exception occurred (MOCK)"}',
50 def config_loader_mock(config_key):
51 """Return mocked config values."""
52 if config_key == "energy_recorder.api_url":
53 return "http://pod-uri:8888"
54 elif config_key == "energy_recorder.api_user":
56 elif config_key == "energy_recorder.api_password":
59 raise Exception("Config not mocked")
62 def config_loader_mock_no_creds(config_key):
63 """Return mocked config values."""
64 if config_key == "energy_recorder.api_url":
65 return "http://pod-uri:8888"
66 elif config_key == "energy_recorder.api_user":
68 elif config_key == "energy_recorder.api_password":
71 raise Exception("Config not mocked:" + config_key)
74 class EnergyRecorderTest(unittest.TestCase):
75 """Energy module unitary test suite."""
78 request_headers = {'content-type': 'application/json'}
79 returned_value_to_preserve = "value"
80 exception_message_to_preserve = "exception_message"
82 @mock.patch('functest.energy.energy.requests.post',
83 return_value=RECORDER_OK)
84 def test_start(self, post_mock=None):
85 """EnergyRecorder.start method (regular case)."""
86 self.test_load_config()
87 self.assertTrue(EnergyRecorder.start(self.case_name))
88 post_mock.assert_called_once_with(
89 EnergyRecorder.energy_recorder_api["uri"],
90 auth=EnergyRecorder.energy_recorder_api["auth"],
92 headers=self.request_headers
95 @mock.patch('functest.energy.energy.requests.post',
96 side_effect=Exception("Internal execution error (MOCK)"))
97 def test_start_error(self, post_mock=None):
98 """EnergyRecorder.start method (error in method)."""
99 self.test_load_config()
100 self.assertFalse(EnergyRecorder.start(self.case_name))
101 post_mock.assert_called_once_with(
102 EnergyRecorder.energy_recorder_api["uri"],
103 auth=EnergyRecorder.energy_recorder_api["auth"],
105 headers=self.request_headers
108 @mock.patch('functest.energy.energy.requests.post',
109 return_value=RECORDER_KO)
110 def test_start_api_error(self, post_mock=None):
111 """EnergyRecorder.start method (API error)."""
112 self.test_load_config()
113 self.assertFalse(EnergyRecorder.start(self.case_name))
114 post_mock.assert_called_once_with(
115 EnergyRecorder.energy_recorder_api["uri"],
116 auth=EnergyRecorder.energy_recorder_api["auth"],
118 headers=self.request_headers
121 @mock.patch('functest.energy.energy.requests.post',
122 return_value=RECORDER_OK)
123 def test_set_step(self, post_mock=None):
124 """EnergyRecorder.set_step method (regular case)."""
125 self.test_load_config()
126 self.assertTrue(EnergyRecorder.set_step(STEP_NAME))
127 post_mock.assert_called_once_with(
128 EnergyRecorder.energy_recorder_api["uri"] + "/step",
129 auth=EnergyRecorder.energy_recorder_api["auth"],
131 headers=self.request_headers
134 @mock.patch('functest.energy.energy.requests.post',
135 return_value=RECORDER_KO)
136 def test_set_step_api_error(self, post_mock=None):
137 """EnergyRecorder.set_step method (API error)."""
138 self.test_load_config()
139 self.assertFalse(EnergyRecorder.set_step(STEP_NAME))
140 post_mock.assert_called_once_with(
141 EnergyRecorder.energy_recorder_api["uri"] + "/step",
142 auth=EnergyRecorder.energy_recorder_api["auth"],
144 headers=self.request_headers
147 @mock.patch('functest.energy.energy.requests.post',
148 side_effect=Exception("Internal execution error (MOCK)"))
149 def test_set_step_error(self, post_mock=None):
150 """EnergyRecorder.set_step method (method 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
160 @mock.patch('functest.energy.energy.requests.delete',
161 return_value=RECORDER_OK)
162 def test_stop(self, delete_mock=None):
163 """EnergyRecorder.stop method (regular case)."""
164 self.test_load_config()
165 self.assertTrue(EnergyRecorder.stop())
166 delete_mock.assert_called_once_with(
167 EnergyRecorder.energy_recorder_api["uri"],
168 auth=EnergyRecorder.energy_recorder_api["auth"],
169 headers=self.request_headers
172 @mock.patch('functest.energy.energy.requests.delete',
173 return_value=RECORDER_KO)
174 def test_stop_api_error(self, delete_mock=None):
175 """EnergyRecorder.stop method (API Error)."""
176 self.test_load_config()
177 self.assertFalse(EnergyRecorder.stop())
178 delete_mock.assert_called_once_with(
179 EnergyRecorder.energy_recorder_api["uri"],
180 auth=EnergyRecorder.energy_recorder_api["auth"],
181 headers=self.request_headers
184 @mock.patch('functest.energy.energy.requests.delete',
185 side_effect=Exception("Internal execution error (MOCK)"))
186 def test_stop_error(self, delete_mock=None):
187 """EnergyRecorder.stop method (method error)."""
188 self.test_load_config()
189 self.assertFalse(EnergyRecorder.stop())
190 delete_mock.assert_called_once_with(
191 EnergyRecorder.energy_recorder_api["uri"],
192 auth=EnergyRecorder.energy_recorder_api["auth"],
193 headers=self.request_headers
196 @energy.enable_recording
197 def __decorated_method(self):
198 """Call with to energy recorder decorators."""
199 return self.returned_value_to_preserve
201 @energy.enable_recording
202 def __decorated_method_with_ex(self):
203 """Call with to energy recorder decorators."""
204 raise Exception(self.exception_message_to_preserve)
206 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
208 @mock.patch("functest.energy.energy.EnergyRecorder")
209 @mock.patch("functest.utils.functest_utils.get_pod_name",
210 return_value="MOCK_POD")
211 @mock.patch("functest.utils.functest_utils.get_functest_config",
212 side_effect=config_loader_mock)
213 def test_decorators(self,
217 cur_scenario_mock=None):
218 """Test energy module decorators."""
219 self.__decorated_method()
220 calls = [mock.call.start(self.case_name),
222 recorder_mock.assert_has_calls(calls)
224 @mock.patch("functest.energy.energy.EnergyRecorder.get_current_scenario",
225 return_value={"scenario": PREVIOUS_SCENARIO,
226 "step": PREVIOUS_STEP})
227 @mock.patch("functest.energy.energy.EnergyRecorder")
228 @mock.patch("functest.utils.functest_utils.get_pod_name",
229 return_value="MOCK_POD")
230 @mock.patch("functest.utils.functest_utils.get_functest_config",
231 side_effect=config_loader_mock)
232 def test_decorators_with_previous(self,
236 cur_scenario_mock=None):
237 """Test energy module decorators."""
238 self.__decorated_method()
239 calls = [mock.call.start(self.case_name),
240 mock.call.submit_scenario(PREVIOUS_SCENARIO,
242 recorder_mock.assert_has_calls(calls)
244 def test_decorator_preserve_return(self):
245 """Test that decorator preserve method returned value."""
246 self.test_load_config()
248 self.__decorated_method() == self.returned_value_to_preserve
252 "functest.energy.energy.finish_session")
253 def test_decorator_preserve_ex(self, finish_mock=None):
254 """Test that decorator preserve method exceptions."""
255 self.test_load_config()
256 with self.assertRaises(Exception) as context:
257 self.__decorated_method_with_ex()
259 self.exception_message_to_preserve in context.exception
261 self.assertTrue(finish_mock.called)
263 @mock.patch("functest.utils.functest_utils.get_functest_config",
264 side_effect=config_loader_mock)
265 @mock.patch("functest.utils.functest_utils.get_pod_name",
266 return_value="MOCK_POD")
267 def test_load_config(self, loader_mock=None, pod_mock=None):
268 """Test load config."""
269 EnergyRecorder.energy_recorder_api = None
270 EnergyRecorder.load_config()
272 EnergyRecorder.energy_recorder_api["auth"],
276 EnergyRecorder.energy_recorder_api["uri"],
277 "http://pod-uri:8888/recorders/environment/MOCK_POD"
280 @mock.patch("functest.utils.functest_utils.get_functest_config",
281 side_effect=config_loader_mock_no_creds)
282 @mock.patch("functest.utils.functest_utils.get_pod_name",
283 return_value="MOCK_POD")
284 def test_load_config_no_creds(self, loader_mock=None, pod_mock=None):
285 """Test load config without creds."""
286 EnergyRecorder.energy_recorder_api = None
287 EnergyRecorder.load_config()
288 self.assertEquals(EnergyRecorder.energy_recorder_api["auth"], None)
290 EnergyRecorder.energy_recorder_api["uri"],
291 "http://pod-uri:8888/recorders/environment/MOCK_POD"
294 @mock.patch("functest.utils.functest_utils.get_functest_config",
296 @mock.patch("functest.utils.functest_utils.get_pod_name",
297 return_value="MOCK_POD")
298 def test_load_config_ex(self, loader_mock=None, pod_mock=None):
299 """Test load config with exception."""
300 with self.assertRaises(AssertionError):
301 EnergyRecorder.energy_recorder_api = None
302 EnergyRecorder.load_config()
303 self.assertEquals(EnergyRecorder.energy_recorder_api, None)
305 @mock.patch("functest.utils.functest_utils.get_functest_config",
307 @mock.patch("functest.utils.functest_utils.get_pod_name",
308 return_value="MOCK_POD")
309 @mock.patch('functest.energy.energy.requests.get',
310 return_value=RECORDER_OK)
311 def test_get_current_scenario(self, loader_mock=None,
312 pod_mock=None, get_mock=None):
313 """Test get_current_scenario."""
314 self.test_load_config()
315 scenario = EnergyRecorder.get_current_scenario()
316 self.assertTrue(scenario is not None)
319 if __name__ == "__main__":
320 logging.disable(logging.CRITICAL)
321 unittest.main(verbosity=2)