« back to all changes in this revision
Viewing changes to bzrlib/tests/test_cleanup.py
- browse files at revision 4775
- compare with another revision
- compare with revision 1156
- download diff
- download tarball
- view history from revision 4775
- Committer: Canonical.com Patch Queue Manager
- Date: 2009-10-28 01:08:22 UTC
- mfrom: (4744.3.6 robust-cleanup-in-commit)
- Revision ID: pqm@pqm.ubuntu.com-20091028010822-2qrdd38w5rcfki8s
(andrew) Add bzrlib.cleanup.OperationWithCleanups,
and use it to make commit.py simpler and more robust.
and use it to make commit.py simpler and more robust.
- files added:
- bzrlib/cleanup.py
- files modified:
- NEWS
added
removed
bzrlib/tests/test_cleanup.py
1
# Copyright (C) 2009 Canonical Ltd
2
#
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.
7
#
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.
12
#
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
16
17
from cStringIO import StringIO
18
import re
19
20
from bzrlib.cleanup import (
21
_do_with_cleanups,
22
_run_cleanup,
23
OperationWithCleanups,
24
)
25
from bzrlib.tests import TestCase
26
from bzrlib import (
27
debug,
28
trace,
29
)
30
31
32
class CleanupsTestCase(TestCase):
33
34
def setUp(self):
35
super(CleanupsTestCase, self).setUp()
36
self.call_log = []
37
38
def no_op_cleanup(self):
39
self.call_log.append('no_op_cleanup')
40
41
def assertLogContains(self, regex):
42
log = self._get_log(keep_log_file=True)
43
self.assertContainsRe(log, regex, re.DOTALL)
44
45
def failing_cleanup(self):
46
self.call_log.append('failing_cleanup')
47
raise Exception("failing_cleanup goes boom!")
48
49
50
class TestRunCleanup(CleanupsTestCase):
51
52
def test_no_errors(self):
53
"""The function passed to _run_cleanup is run."""
54
self.assertTrue(_run_cleanup(self.no_op_cleanup))
55
self.assertEqual(['no_op_cleanup'], self.call_log)
56
57
def test_cleanup_with_args_kwargs(self):
58
def func_taking_args_kwargs(*args, **kwargs):
59
self.call_log.append(('func', args, kwargs))
60
_run_cleanup(func_taking_args_kwargs, 'an arg', kwarg='foo')
61
self.assertEqual(
62
[('func', ('an arg',), {'kwarg': 'foo'})], self.call_log)
63
64
def test_cleanup_error(self):
65
"""An error from the cleanup function is logged by _run_cleanup, but not
66
propagated.
67
68
This is there's no way for _run_cleanup to know if there's an existing
69
exception in this situation::
70
try:
71
some_func()
72
finally:
73
_run_cleanup(cleanup_func)
74
So, the best _run_cleanup can do is always log errors but never raise
75
them.
76
"""
77
self.assertFalse(_run_cleanup(self.failing_cleanup))
78
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
79
80
def test_cleanup_error_debug_flag(self):
81
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
82
user.
83
"""
84
log = StringIO()
85
trace.push_log_file(log)
86
debug.debug_flags.add('cleanup')
87
self.assertFalse(_run_cleanup(self.failing_cleanup))
88
self.assertContainsRe(
89
log.getvalue(),
90
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
91
92
def test_prior_error_cleanup_succeeds(self):
93
"""Calling _run_cleanup from a finally block will not interfere with an
94
exception from the try block.
95
"""
96
def failing_operation():
97
try:
98
1/0
99
finally:
100
_run_cleanup(self.no_op_cleanup)
101
self.assertRaises(ZeroDivisionError, failing_operation)
102
self.assertEqual(['no_op_cleanup'], self.call_log)
103
104
def test_prior_error_cleanup_fails(self):
105
"""Calling _run_cleanup from a finally block will not interfere with an
106
exception from the try block even when the cleanup itself raises an
107
exception.
108
109
The cleanup exception will be logged.
110
"""
111
def failing_operation():
112
try:
113
1/0
114
finally:
115
_run_cleanup(self.failing_cleanup)
116
self.assertRaises(ZeroDivisionError, failing_operation)
117
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
118
119
120
class TestDoWithCleanups(CleanupsTestCase):
121
122
def trivial_func(self):
123
self.call_log.append('trivial_func')
124
return 'trivial result'
125
126
def test_runs_func(self):
127
"""_do_with_cleanups runs the function it is given, and returns the
128
result.
129
"""
130
result = _do_with_cleanups([], self.trivial_func)
131
self.assertEqual('trivial result', result)
132
133
def test_runs_cleanups(self):
134
"""Cleanup functions are run (in the given order)."""
135
cleanup_func_1 = (self.call_log.append, ('cleanup 1',), {})
136
cleanup_func_2 = (self.call_log.append, ('cleanup 2',), {})
137
_do_with_cleanups([cleanup_func_1, cleanup_func_2], self.trivial_func)
138
self.assertEqual(
139
['trivial_func', 'cleanup 1', 'cleanup 2'], self.call_log)
140
141
def failing_func(self):
142
self.call_log.append('failing_func')
143
1/0
144
145
def test_func_error_propagates(self):
146
"""Errors from the main function are propagated (after running
147
cleanups).
148
"""
149
self.assertRaises(
150
ZeroDivisionError, _do_with_cleanups,
151
[(self.no_op_cleanup, (), {})], self.failing_func)
152
self.assertEqual(['failing_func', 'no_op_cleanup'], self.call_log)
153
154
def test_func_error_trumps_cleanup_error(self):
155
"""Errors from the main function a propagated even if a cleanup raises
156
an error.
157
158
The cleanup error is be logged.
159
"""
160
self.assertRaises(
161
ZeroDivisionError, _do_with_cleanups,
162
[(self.failing_cleanup, (), {})], self.failing_func)
163
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
164
165
def test_func_passes_and_error_from_cleanup(self):
166
"""An error from a cleanup is propagated when the main function doesn't
167
raise an error. Later cleanups are still executed.
168
"""
169
exc = self.assertRaises(
170
Exception, _do_with_cleanups,
171
[(self.failing_cleanup, (), {}), (self.no_op_cleanup, (), {})],
172
self.trivial_func)
173
self.assertEqual('failing_cleanup goes boom!', exc.args[0])
174
self.assertEqual(
175
['trivial_func', 'failing_cleanup', 'no_op_cleanup'],
176
self.call_log)
177
178
def test_multiple_cleanup_failures(self):
179
"""When multiple cleanups fail (as tends to happen when something has
180
gone wrong), the first error is propagated, and subsequent errors are
181
logged.
182
"""
183
cleanups = self.make_two_failing_cleanup_funcs()
184
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
185
self.trivial_func)
186
self.assertLogContains('Cleanup failed:.*ErrorB')
187
log = self._get_log(keep_log_file=True)
188
self.assertFalse('ErrorA' in log)
189
190
def make_two_failing_cleanup_funcs(self):
191
def raise_a():
192
raise ErrorA('Error A')
193
def raise_b():
194
raise ErrorB('Error B')
195
return [(raise_a, (), {}), (raise_b, (), {})]
196
197
def test_multiple_cleanup_failures_debug_flag(self):
198
log = StringIO()
199
trace.push_log_file(log)
200
debug.debug_flags.add('cleanup')
201
cleanups = self.make_two_failing_cleanup_funcs()
202
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
203
self.trivial_func)
204
self.assertContainsRe(
205
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
206
self.assertEqual(1, log.getvalue().count('bzr: warning:'),
207
log.getvalue())
208
209
def test_func_and_cleanup_errors_debug_flag(self):
210
log = StringIO()
211
trace.push_log_file(log)
212
debug.debug_flags.add('cleanup')
213
cleanups = self.make_two_failing_cleanup_funcs()
214
self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
215
self.failing_func)
216
self.assertContainsRe(
217
log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
218
self.assertContainsRe(
219
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
220
self.assertEqual(2, log.getvalue().count('bzr: warning:'))
221
222
def test_func_may_mutate_cleanups(self):
223
"""The main func may mutate the cleanups before it returns.
224
225
This allows a function to gradually add cleanups as it acquires
226
resources, rather than planning all the cleanups up-front. The
227
OperationWithCleanups helper relies on this working.
228
"""
229
cleanups_list = []
230
def func_that_adds_cleanups():
231
self.call_log.append('func_that_adds_cleanups')
232
cleanups_list.append((self.no_op_cleanup, (), {}))
233
return 'result'
234
result = _do_with_cleanups(cleanups_list, func_that_adds_cleanups)
235
self.assertEqual('result', result)
236
self.assertEqual(
237
['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
238
239
def test_cleanup_error_debug_flag(self):
240
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
241
user.
242
"""
243
log = StringIO()
244
trace.push_log_file(log)
245
debug.debug_flags.add('cleanup')
246
self.assertRaises(ZeroDivisionError, _do_with_cleanups,
247
[(self.failing_cleanup, (), {})], self.failing_func)
248
self.assertContainsRe(
249
log.getvalue(),
250
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
251
self.assertEqual(1, log.getvalue().count('bzr: warning:'))
252
253
254
class ErrorA(Exception): pass
255
class ErrorB(Exception): pass
256
257
258
class TestOperationWithCleanups(CleanupsTestCase):
259
260
def test_cleanup_ordering(self):
261
"""Cleanups are added in LIFO order.
262
263
So cleanups added before run is called are run last, and the last
264
cleanup added during the func is run first.
265
"""
266
call_log = []
267
def func(op, foo):
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')
271
return 'result'
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)
277
self.assertEqual(
278
[('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
279
'cleanup 4'], call_log)
280
Loggerhead is a web-based interface for Breezy
Version: 2.0.1