1
# Copyright (C) 2009 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
17
from cStringIO import StringIO
20
from bzrlib.cleanup import (
24
from bzrlib.tests import TestCase
31
class CleanupsTestCase(TestCase):
34
super(CleanupsTestCase, self).setUp()
37
def no_op_cleanup(self):
38
self.call_log.append('no_op_cleanup')
40
def assertLogContains(self, regex):
41
log = self._get_log(keep_log_file=True)
42
self.assertContainsRe(log, regex, re.DOTALL)
44
def failing_cleanup(self):
45
self.call_log.append('failing_cleanup')
46
raise Exception("failing_cleanup goes boom!")
49
class TestRunCleanup(CleanupsTestCase):
51
def test_no_errors(self):
52
"""The function passed to run_cleanup is run."""
53
self.assertTrue(run_cleanup(self.no_op_cleanup))
54
self.assertEqual(['no_op_cleanup'], self.call_log)
56
def test_cleanup_with_args_kwargs(self):
57
def func_taking_args_kwargs(*args, **kwargs):
58
self.call_log.append(('func', args, kwargs))
59
run_cleanup(func_taking_args_kwargs, 'an arg', kwarg='foo')
61
[('func', ('an arg',), {'kwarg': 'foo'})], self.call_log)
63
def test_cleanup_error(self):
64
"""An error from the cleanup function is logged by run_cleanup, but not
67
This is there's no way for run_cleanup to know if there's an existing
68
exception in this situation::
72
run_cleanup(cleanup_func)
73
So, the best run_cleanup can do is always log errors but never raise
76
self.assertFalse(run_cleanup(self.failing_cleanup))
77
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
79
def test_cleanup_error_debug_flag(self):
80
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
84
trace.push_log_file(log)
85
debug.debug_flags.add('cleanup')
86
self.assertFalse(run_cleanup(self.failing_cleanup))
87
self.assertContainsRe(
89
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
91
def test_prior_error_cleanup_succeeds(self):
92
"""Calling run_cleanup from a finally block will not interfere with an
93
exception from the try block.
95
def failing_operation():
99
run_cleanup(self.no_op_cleanup)
100
self.assertRaises(ZeroDivisionError, failing_operation)
101
self.assertEqual(['no_op_cleanup'], self.call_log)
103
def test_prior_error_cleanup_fails(self):
104
"""Calling run_cleanup from a finally block will not interfere with an
105
exception from the try block even when the cleanup itself raises an
108
The cleanup exception will be logged.
110
def failing_operation():
114
run_cleanup(self.failing_cleanup)
115
self.assertRaises(ZeroDivisionError, failing_operation)
116
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
119
#class TestRunCleanupReportingErrors(CleanupsTestCase):
121
# def test_cleanup_error_reported(self):
125
class TestDoWithCleanups(CleanupsTestCase):
127
def trivial_func(self):
128
self.call_log.append('trivial_func')
129
return 'trivial result'
131
def test_runs_func(self):
132
"""do_with_cleanups runs the function it is given, and returns the
135
result = do_with_cleanups(self.trivial_func, [])
136
self.assertEqual('trivial result', result)
138
def test_runs_cleanups(self):
139
"""Cleanup functions are run (in the given order)."""
140
cleanup_func_1 = lambda: self.call_log.append('cleanup 1')
141
cleanup_func_2 = lambda: self.call_log.append('cleanup 2')
142
do_with_cleanups(self.trivial_func, [cleanup_func_1, cleanup_func_2])
144
['trivial_func', 'cleanup 1', 'cleanup 2'], self.call_log)
146
def failing_func(self):
147
self.call_log.append('failing_func')
150
def test_func_error_propagates(self):
151
"""Errors from the main function are propagated (after running
155
ZeroDivisionError, do_with_cleanups, self.failing_func,
156
[self.no_op_cleanup])
157
self.assertEqual(['failing_func', 'no_op_cleanup'], self.call_log)
159
def test_func_error_trumps_cleanup_error(self):
160
"""Errors from the main function a propagated even if a cleanup raises
163
The cleanup error is be logged.
166
ZeroDivisionError, do_with_cleanups, self.failing_func,
167
[self.failing_cleanup])
168
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
170
def test_func_passes_and_error_from_cleanup(self):
171
"""An error from a cleanup is propagated when the main function doesn't
172
raise an error. Later cleanups are still executed.
174
exc = self.assertRaises(
175
Exception, do_with_cleanups, self.trivial_func,
176
[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, self.trivial_func,
190
self.assertLogContains('Cleanup failed:.*ErrorB')
191
log = self._get_log(keep_log_file=True)
192
self.assertFalse('ErrorA' in log)
194
def make_two_failing_cleanup_funcs(self):
196
raise ErrorA('Error A')
198
raise ErrorB('Error B')
199
return [raise_a, raise_b]
201
def test_multiple_cleanup_failures_debug_flag(self):
203
trace.push_log_file(log)
204
debug.debug_flags.add('cleanup')
205
cleanups = self.make_two_failing_cleanup_funcs()
206
self.assertRaises(ErrorA, do_with_cleanups, self.trivial_func, cleanups)
207
self.assertContainsRe(
208
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
209
self.assertEqual(1, log.getvalue().count('bzr: warning:'),
212
def test_func_and_cleanup_errors_debug_flag(self):
214
trace.push_log_file(log)
215
debug.debug_flags.add('cleanup')
216
cleanups = self.make_two_failing_cleanup_funcs()
217
self.assertRaises(ZeroDivisionError, do_with_cleanups,
218
self.failing_func, cleanups)
219
self.assertContainsRe(
220
log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
221
self.assertContainsRe(
222
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
223
self.assertEqual(2, log.getvalue().count('bzr: warning:'))
225
def test_func_may_mutate_cleanups(self):
226
"""The main func may mutate the cleanups before it returns.
228
This allows a function to gradually add cleanups as it acquires
229
resources, rather than planning all the cleanups up-front.
231
# XXX: this is cute, but an object with an 'add_cleanup' method may
234
def func_that_adds_cleanups():
235
self.call_log.append('func_that_adds_cleanups')
236
cleanups_list.append(self.no_op_cleanup)
238
result = do_with_cleanups(func_that_adds_cleanups, cleanups_list)
239
self.assertEqual('result', result)
241
['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
243
def test_cleanup_error_debug_flag(self):
244
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
248
trace.push_log_file(log)
249
debug.debug_flags.add('cleanup')
250
self.assertRaises(ZeroDivisionError, do_with_cleanups,
251
self.failing_func, [self.failing_cleanup])
252
self.assertContainsRe(
254
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
255
self.assertEqual(1, log.getvalue().count('bzr: warning:'))
258
class ErrorA(Exception): pass
259
class ErrorB(Exception): pass
added
removed
bzrlib/tests/test_cleanup.py
