Coverage for manila/tests/share/test_hook.py: 100%
164 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2015 Mirantis Inc.
2# All Rights Reserved.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16from unittest import mock
18import ddt
20from manila import context
21from manila.share import hook
22from manila import test
25class FakeHookImplementation(hook.HookBase):
26 def _execute_pre_hook(self, context, func_name, *args, **kwargs):
27 """Fake implementation of a pre hook action."""
29 def _execute_post_hook(self, context, func_name, pre_hook_data,
30 driver_action_results, *args, **kwargs):
31 """Fake implementation of a post hook action."""
33 def _execute_periodic_hook(self, context, periodic_hook_data,
34 *args, **kwargs):
35 """Fake implementation of a periodic hook action."""
38@ddt.ddt
39class HookBaseTestCase(test.TestCase):
41 def setUp(self):
42 super(HookBaseTestCase, self).setUp()
43 self.context = context.get_admin_context()
44 self.default_config = {
45 "enable_pre_hooks": True,
46 "enable_post_hooks": True,
47 "enable_periodic_hooks": True,
48 "suppress_pre_hooks_errors": True,
49 "suppress_post_hooks_errors": True,
50 }
51 for k, v in self.default_config.items():
52 hook.CONF.set_default(k, v)
54 def _fake_safe_get(self, key):
55 return self.default_config.get(key)
57 def _get_hook_instance(self, set_configuration=True, host="fake_host"):
58 if set_configuration:
59 configuration = mock.Mock()
60 configuration.safe_get.side_effect = self._fake_safe_get
61 else:
62 configuration = None
63 instance = FakeHookImplementation(
64 configuration=configuration, host=host)
65 return instance
67 def test_instantiate_hook_fail(self):
68 self.assertRaises(TypeError, hook.HookBase)
70 @ddt.data(True, False)
71 def test_instantiate_hook_successfully_and_set_configuration(
72 self, set_configuration):
73 instance = self._get_hook_instance(set_configuration)
75 self.assertTrue(hasattr(instance, 'host'))
76 self.assertEqual("fake_host", instance.host)
77 self.assertTrue(hasattr(instance, 'configuration'))
78 if not set_configuration:
79 self.assertIsNone(instance.configuration)
80 for attr_name in ("pre_hooks_enabled",
81 "post_hooks_enabled",
82 "periodic_hooks_enabled",
83 "suppress_pre_hooks_errors",
84 "suppress_post_hooks_errors"):
85 self.assertTrue(hasattr(instance, attr_name))
86 if set_configuration:
87 instance.configuration.append_config_values.assert_has_calls([
88 mock.call(hook.hook_options)])
89 conf_func = self._fake_safe_get
90 else:
91 conf_func = self.default_config.get
92 self.assertEqual(
93 conf_func("enable_pre_hooks"), instance.pre_hooks_enabled)
94 self.assertEqual(
95 conf_func("enable_post_hooks"), instance.post_hooks_enabled)
96 self.assertEqual(
97 conf_func("enable_periodic_hooks"),
98 instance.periodic_hooks_enabled)
99 self.assertEqual(
100 conf_func("suppress_pre_hooks_errors"),
101 instance.suppress_pre_hooks_errors)
102 self.assertEqual(
103 conf_func("suppress_post_hooks_errors"),
104 instance.suppress_post_hooks_errors)
106 def test_execute_pre_hook_disabled(self):
107 instance = self._get_hook_instance()
108 instance.pre_hooks_enabled = False
109 self.mock_object(
110 instance, "_execute_pre_hook",
111 mock.Mock(side_effect=Exception("I should not be raised.")))
113 result = instance.execute_pre_hook(
114 self.context, "fake_func_name", "some_arg", some_kwarg="foo")
116 self.assertIsNone(result)
118 @ddt.data(True, False)
119 def test_execute_pre_hook_success(self, provide_context):
120 instance = self._get_hook_instance()
121 instance.pre_hooks_enabled = True
122 instance.suppress_pre_hooks_errors = True
123 expected = "fake_expected_result"
124 some_arg = "some_arg"
125 func_name = "fake_func_name"
126 self.mock_object(hook.LOG, 'error')
127 self.mock_object(
128 instance, "_execute_pre_hook", mock.Mock(return_value=expected))
129 mock_ctxt = self.mock_object(context, 'get_admin_context')
130 ctxt = self.context if provide_context else mock_ctxt
132 result = instance.execute_pre_hook(
133 ctxt, func_name, some_arg, some_kwarg="foo")
135 self.assertEqual(expected, result)
136 instance._execute_pre_hook.assert_called_once_with(
137 some_arg,
138 context=self.context if provide_context else mock_ctxt,
139 func_name=func_name,
140 some_kwarg="foo")
141 self.assertFalse(hook.LOG.error.called)
143 def test_execute_pre_hook_exception_with_suppression(self):
144 instance = self._get_hook_instance()
145 instance.pre_hooks_enabled = True
146 instance.suppress_pre_hooks_errors = True
147 some_arg = "some_arg"
148 func_name = "fake_func_name"
149 FakeException = type("FakeException", (Exception, ), {})
150 self.mock_object(hook.LOG, 'warning')
151 self.mock_object(
152 instance, "_execute_pre_hook", mock.Mock(side_effect=(
153 FakeException("Some exception that should be suppressed."))))
155 result = instance.execute_pre_hook(
156 self.context, func_name, some_arg, some_kwarg="foo")
158 self.assertIsInstance(result, FakeException)
159 instance._execute_pre_hook.assert_called_once_with(
160 some_arg,
161 context=self.context,
162 func_name=func_name,
163 some_kwarg="foo")
164 self.assertTrue(hook.LOG.warning.called)
166 def test_execute_pre_hook_exception_without_suppression(self):
167 instance = self._get_hook_instance()
168 instance.pre_hooks_enabled = True
169 instance.suppress_pre_hooks_errors = False
170 some_arg = "some_arg"
171 func_name = "fake_func_name"
172 FakeException = type("FakeException", (Exception, ), {})
173 self.mock_object(hook.LOG, 'warning')
174 self.mock_object(
175 instance, "_execute_pre_hook", mock.Mock(side_effect=(
176 FakeException(
177 "Some exception that should NOT be suppressed."))))
179 self.assertRaises(
180 FakeException,
181 instance.execute_pre_hook,
182 self.context, func_name, some_arg, some_kwarg="foo")
184 instance._execute_pre_hook.assert_called_once_with(
185 some_arg,
186 context=self.context,
187 func_name=func_name,
188 some_kwarg="foo")
189 self.assertFalse(hook.LOG.warning.called)
191 def test_execute_post_hook_disabled(self):
192 instance = self._get_hook_instance()
193 instance.post_hooks_enabled = False
194 self.mock_object(
195 instance, "_execute_post_hook",
196 mock.Mock(side_effect=Exception("I should not be raised.")))
198 result = instance.execute_post_hook(
199 self.context, "fake_func_name", "some_pre_hook_data",
200 "some_driver_action_results", "some_arg", some_kwarg="foo")
202 self.assertIsNone(result)
204 @ddt.data(True, False)
205 def test_execute_post_hook_success(self, provide_context):
206 instance = self._get_hook_instance()
207 instance.post_hooks_enabled = True
208 instance.suppress_post_hooks_errors = True
209 expected = "fake_expected_result"
210 some_arg = "some_arg"
211 func_name = "fake_func_name"
212 pre_hook_data = "some_pre_hook_data"
213 driver_action_results = "some_driver_action_results"
214 self.mock_object(hook.LOG, 'warning')
215 self.mock_object(
216 instance, "_execute_post_hook", mock.Mock(return_value=expected))
217 mock_ctxt = self.mock_object(context, 'get_admin_context')
218 ctxt = self.context if provide_context else mock_ctxt
220 result = instance.execute_post_hook(
221 ctxt, func_name, pre_hook_data, driver_action_results,
222 some_arg, some_kwarg="foo")
224 self.assertEqual(expected, result)
225 instance._execute_post_hook.assert_called_once_with(
226 some_arg,
227 context=self.context if provide_context else mock_ctxt,
228 func_name=func_name,
229 pre_hook_data=pre_hook_data,
230 driver_action_results=driver_action_results,
231 some_kwarg="foo")
232 self.assertFalse(hook.LOG.warning.called)
234 def test_execute_post_hook_exception_with_suppression(self):
235 instance = self._get_hook_instance()
236 instance.post_hooks_enabled = True
237 instance.suppress_post_hooks_errors = True
238 some_arg = "some_arg"
239 func_name = "fake_func_name"
240 pre_hook_data = "some_pre_hook_data"
241 driver_action_results = "some_driver_action_results"
242 FakeException = type("FakeException", (Exception, ), {})
243 self.mock_object(hook.LOG, 'warning')
244 self.mock_object(
245 instance, "_execute_post_hook", mock.Mock(side_effect=(
246 FakeException("Some exception that should be suppressed."))))
248 result = instance.execute_post_hook(
249 self.context, func_name, pre_hook_data, driver_action_results,
250 some_arg, some_kwarg="foo")
252 self.assertIsInstance(result, FakeException)
253 instance._execute_post_hook.assert_called_once_with(
254 some_arg,
255 context=self.context,
256 func_name=func_name,
257 pre_hook_data=pre_hook_data,
258 driver_action_results=driver_action_results,
259 some_kwarg="foo")
260 self.assertTrue(hook.LOG.warning.called)
262 def test_execute_post_hook_exception_without_suppression(self):
263 instance = self._get_hook_instance()
264 instance.post_hooks_enabled = True
265 instance.suppress_post_hooks_errors = False
266 some_arg = "some_arg"
267 func_name = "fake_func_name"
268 pre_hook_data = "some_pre_hook_data"
269 driver_action_results = "some_driver_action_results"
270 FakeException = type("FakeException", (Exception, ), {})
271 self.mock_object(hook.LOG, 'error')
272 self.mock_object(
273 instance, "_execute_post_hook", mock.Mock(side_effect=(
274 FakeException(
275 "Some exception that should NOT be suppressed."))))
277 self.assertRaises(
278 FakeException,
279 instance.execute_post_hook,
280 self.context, func_name, pre_hook_data, driver_action_results,
281 some_arg, some_kwarg="foo")
283 instance._execute_post_hook.assert_called_once_with(
284 some_arg,
285 context=self.context,
286 func_name=func_name,
287 pre_hook_data=pre_hook_data,
288 driver_action_results=driver_action_results,
289 some_kwarg="foo")
290 self.assertFalse(hook.LOG.error.called)
292 def test_execute_periodic_hook_disabled(self):
293 instance = self._get_hook_instance()
294 instance.periodic_hooks_enabled = False
295 self.mock_object(instance, "_execute_periodic_hook")
297 instance.execute_periodic_hook(
298 self.context, "fake_periodic_hook_data",
299 "some_arg", some_kwarg="foo")
301 self.assertFalse(instance._execute_periodic_hook.called)
303 @ddt.data(True, False)
304 def test_execute_periodic_hook_enabled(self, provide_context):
305 instance = self._get_hook_instance()
306 instance.periodic_hooks_enabled = True
307 expected = "some_expected_result"
308 self.mock_object(
309 instance,
310 "_execute_periodic_hook",
311 mock.Mock(return_value=expected))
312 mock_ctxt = self.mock_object(context, 'get_admin_context')
313 ctxt = self.context if provide_context else mock_ctxt
315 result = instance.execute_periodic_hook(
316 ctxt, "fake_periodic_hook_data",
317 "some_arg", some_kwarg="foo")
319 instance._execute_periodic_hook.assert_called_once_with(
320 ctxt, "fake_periodic_hook_data",
321 "some_arg", some_kwarg="foo")
322 self.assertEqual(expected, result)