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')
202
raise ErrorB('Error B')
203
return [(raise_a, (), {}), (raise_b, (), {})]
205
def test_multiple_cleanup_failures_debug_flag(self):
206
debug.debug_flags.add('cleanup')
207
cleanups = self.make_two_failing_cleanup_funcs()
208
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
210
trace_value = self.get_log()
211
self.assertContainsRe(
212
trace_value, "brz: warning: Cleanup failed:.*Error B\n")
213
self.assertEqual(1, trace_value.count('brz: warning:'))
215
def test_func_and_cleanup_errors_debug_flag(self):
216
debug.debug_flags.add('cleanup')
217
cleanups = self.make_two_failing_cleanup_funcs()
218
self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
220
trace_value = self.get_log()
221
self.assertContainsRe(
222
trace_value, "brz: warning: Cleanup failed:.*Error A\n")
223
self.assertContainsRe(
224
trace_value, "brz: warning: Cleanup failed:.*Error B\n")
225
self.assertEqual(2, trace_value.count('brz: warning:'))
227
def test_func_may_mutate_cleanups(self):
228
"""The main func may mutate the cleanups before it returns.
230
This allows a function to gradually add cleanups as it acquires
231
resources, rather than planning all the cleanups up-front. The
232
OperationWithCleanups helper relies on this working.
235
def func_that_adds_cleanups():
236
self.call_log.append('func_that_adds_cleanups')
237
cleanups_list.append((self.no_op_cleanup, (), {}))
239
result = _do_with_cleanups(cleanups_list, func_that_adds_cleanups)
240
self.assertEqual('result', result)
242
['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
244
def test_cleanup_error_debug_flag(self):
245
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
248
debug.debug_flags.add('cleanup')
249
self.assertRaises(ZeroDivisionError, _do_with_cleanups,
250
[(self.failing_cleanup, (), {})], self.failing_func)
251
trace_value = self.get_log()
252
self.assertContainsRe(
254
"brz: warning: Cleanup failed:.*failing_cleanup goes boom")
255
self.assertEqual(1, trace_value.count('brz: warning:'))
258
class TestOperationWithCleanups(CleanupsTestCase):
260
def test_cleanup_ordering(self):
261
"""Cleanups are added in LIFO order.
263
So cleanups added before run is called are run last, and the last
264
cleanup added during the func is run first.
268
call_log.append(('func called', foo))
269
op.add_cleanup(call_log.append, 'cleanup 2')
270
op.add_cleanup(call_log.append, 'cleanup 1')
272
owc = OperationWithCleanups(func)
273
owc.add_cleanup(call_log.append, 'cleanup 4')
274
owc.add_cleanup(call_log.append, 'cleanup 3')
275
result = owc.run('foo')
276
self.assertEqual('result', result)
278
[('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
279
'cleanup 4'], call_log)
282
class SampleWithCleanups(ObjectWithCleanups):
283
"""Minimal ObjectWithCleanups subclass."""
286
class TestObjectWithCleanups(tests.TestCase):
288
def test_object_with_cleanups(self):
290
s = SampleWithCleanups()
291
s.add_cleanup(a.append, 42)
293
self.assertEqual(a, [42])
added
removed
breezy/tests/test_cleanup.py
