1
# Copyright (C) 2009, 2010 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19
from ..cleanup import (
23
OperationWithCleanups,
31
class ErrorA(Exception):
32
"""Sample exception type A."""
35
class ErrorB(Exception):
36
"""Sample exception type B."""
39
class CleanupsTestCase(tests.TestCase):
42
super(CleanupsTestCase, self).setUp()
45
def no_op_cleanup(self):
46
self.call_log.append('no_op_cleanup')
48
def assertLogContains(self, regex):
49
self.assertContainsRe(self.get_log(), regex, re.DOTALL)
51
def failing_cleanup(self):
52
self.call_log.append('failing_cleanup')
53
raise Exception("failing_cleanup goes boom!")
56
class TestRunCleanup(CleanupsTestCase):
58
def test_no_errors(self):
59
"""The function passed to _run_cleanup is run."""
60
self.assertTrue(_run_cleanup(self.no_op_cleanup))
61
self.assertEqual(['no_op_cleanup'], self.call_log)
63
def test_cleanup_with_args_kwargs(self):
64
def func_taking_args_kwargs(*args, **kwargs):
65
self.call_log.append(('func', args, kwargs))
66
_run_cleanup(func_taking_args_kwargs, 'an arg', kwarg='foo')
68
[('func', ('an arg',), {'kwarg': 'foo'})], self.call_log)
70
def test_cleanup_error(self):
71
"""An error from the cleanup function is logged by _run_cleanup, but not
74
This is there's no way for _run_cleanup to know if there's an existing
75
exception in this situation::
79
_run_cleanup(cleanup_func)
80
So, the best _run_cleanup can do is always log errors but never raise
83
self.assertFalse(_run_cleanup(self.failing_cleanup))
84
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
86
def test_cleanup_error_debug_flag(self):
87
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
90
debug.debug_flags.add('cleanup')
91
self.assertFalse(_run_cleanup(self.failing_cleanup))
92
self.assertContainsRe(
94
"brz: warning: Cleanup failed:.*failing_cleanup goes boom")
96
def test_prior_error_cleanup_succeeds(self):
97
"""Calling _run_cleanup from a finally block will not interfere with an
98
exception from the try block.
100
def failing_operation():
104
_run_cleanup(self.no_op_cleanup)
105
self.assertRaises(ZeroDivisionError, failing_operation)
106
self.assertEqual(['no_op_cleanup'], self.call_log)
108
def test_prior_error_cleanup_fails(self):
109
"""Calling _run_cleanup from a finally block will not interfere with an
110
exception from the try block even when the cleanup itself raises an
113
The cleanup exception will be logged.
115
def failing_operation():
119
_run_cleanup(self.failing_cleanup)
120
self.assertRaises(ZeroDivisionError, failing_operation)
121
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
124
class TestDoWithCleanups(CleanupsTestCase):
126
def trivial_func(self):
127
self.call_log.append('trivial_func')
128
return 'trivial result'
130
def test_runs_func(self):
131
"""_do_with_cleanups runs the function it is given, and returns the
134
result = _do_with_cleanups([], self.trivial_func)
135
self.assertEqual('trivial result', result)
137
def test_runs_cleanups(self):
138
"""Cleanup functions are run (in the given order)."""
139
cleanup_func_1 = (self.call_log.append, ('cleanup 1',), {})
140
cleanup_func_2 = (self.call_log.append, ('cleanup 2',), {})
141
_do_with_cleanups([cleanup_func_1, cleanup_func_2], self.trivial_func)
143
['trivial_func', 'cleanup 1', 'cleanup 2'], self.call_log)
145
def failing_func(self):
146
self.call_log.append('failing_func')
149
def test_func_error_propagates(self):
150
"""Errors from the main function are propagated (after running
154
ZeroDivisionError, _do_with_cleanups,
155
[(self.no_op_cleanup, (), {})], self.failing_func)
156
self.assertEqual(['failing_func', 'no_op_cleanup'], self.call_log)
158
def test_func_error_trumps_cleanup_error(self):
159
"""Errors from the main function a propagated even if a cleanup raises
162
The cleanup error is be logged.
165
ZeroDivisionError, _do_with_cleanups,
166
[(self.failing_cleanup, (), {})], self.failing_func)
167
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
169
def test_func_passes_and_error_from_cleanup(self):
170
"""An error from a cleanup is propagated when the main function doesn't
171
raise an error. Later cleanups are still executed.
173
exc = self.assertRaises(
174
Exception, _do_with_cleanups,
175
[(self.failing_cleanup, (), {}), (self.no_op_cleanup, (), {})],
177
self.assertEqual('failing_cleanup goes boom!', exc.args[0])
179
['trivial_func', 'failing_cleanup', 'no_op_cleanup'],
182
def test_multiple_cleanup_failures(self):
183
"""When multiple cleanups fail (as tends to happen when something has
184
gone wrong), the first error is propagated, and subsequent errors are
187
cleanups = self.make_two_failing_cleanup_funcs()
188
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
190
self.assertLogContains('Cleanup failed:.*ErrorB')
191
# Error A may appear in the log (with Python 3 exception chaining), but
192
# Error B should be the last error recorded.
193
self.assertContainsRe(
195
'Traceback \\(most recent call last\\):\n( .*\n)+'
196
'.*ErrorB: Error B\n$')
198
def make_two_failing_cleanup_funcs(self):
200
raise ErrorA('Error A')
203
raise ErrorB('Error B')
204
return [(raise_a, (), {}), (raise_b, (), {})]
206
def test_multiple_cleanup_failures_debug_flag(self):
207
debug.debug_flags.add('cleanup')
208
cleanups = self.make_two_failing_cleanup_funcs()
209
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
211
trace_value = self.get_log()
212
self.assertContainsRe(
213
trace_value, "brz: warning: Cleanup failed:.*Error B\n")
214
self.assertEqual(1, trace_value.count('brz: warning:'))
216
def test_func_and_cleanup_errors_debug_flag(self):
217
debug.debug_flags.add('cleanup')
218
cleanups = self.make_two_failing_cleanup_funcs()
219
self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
221
trace_value = self.get_log()
222
self.assertContainsRe(
223
trace_value, "brz: warning: Cleanup failed:.*Error A\n")
224
self.assertContainsRe(
225
trace_value, "brz: warning: Cleanup failed:.*Error B\n")
226
self.assertEqual(2, trace_value.count('brz: warning:'))
228
def test_func_may_mutate_cleanups(self):
229
"""The main func may mutate the cleanups before it returns.
231
This allows a function to gradually add cleanups as it acquires
232
resources, rather than planning all the cleanups up-front. The
233
OperationWithCleanups helper relies on this working.
237
def func_that_adds_cleanups():
238
self.call_log.append('func_that_adds_cleanups')
239
cleanups_list.append((self.no_op_cleanup, (), {}))
241
result = _do_with_cleanups(cleanups_list, func_that_adds_cleanups)
242
self.assertEqual('result', result)
244
['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
246
def test_cleanup_error_debug_flag(self):
247
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
250
debug.debug_flags.add('cleanup')
251
self.assertRaises(ZeroDivisionError, _do_with_cleanups,
252
[(self.failing_cleanup, (), {})], self.failing_func)
253
trace_value = self.get_log()
254
self.assertContainsRe(
256
"brz: warning: Cleanup failed:.*failing_cleanup goes boom")
257
self.assertEqual(1, trace_value.count('brz: warning:'))
260
class TestOperationWithCleanups(CleanupsTestCase):
262
def test_cleanup_ordering(self):
263
"""Cleanups are added in LIFO order.
265
So cleanups added before run is called are run last, and the last
266
cleanup added during the func is run first.
271
call_log.append(('func called', foo))
272
op.add_cleanup(call_log.append, 'cleanup 2')
273
op.add_cleanup(call_log.append, 'cleanup 1')
275
owc = OperationWithCleanups(func)
276
owc.add_cleanup(call_log.append, 'cleanup 4')
277
owc.add_cleanup(call_log.append, 'cleanup 3')
278
result = owc.run('foo')
279
self.assertEqual('result', result)
281
[('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
282
'cleanup 4'], call_log)
285
class SampleWithCleanups(ObjectWithCleanups):
286
"""Minimal ObjectWithCleanups subclass."""
289
class TestObjectWithCleanups(tests.TestCase):
291
def test_object_with_cleanups(self):
293
s = SampleWithCleanups()
294
s.add_cleanup(a.append, 42)
296
self.assertEqual(a, [42])