14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
from cStringIO import StringIO
20
from bzrlib.cleanup import (
19
from ..cleanup import (
23
23
OperationWithCleanups,
25
from bzrlib.tests import TestCase
32
class CleanupsTestCase(TestCase):
31
class ErrorA(Exception):
32
"""Sample exception type A."""
35
class ErrorB(Exception):
36
"""Sample exception type B."""
39
class CleanupsTestCase(tests.TestCase):
35
42
super(CleanupsTestCase, self).setUp()
80
87
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
84
trace.push_log_file(log)
85
90
debug.debug_flags.add('cleanup')
86
91
self.assertFalse(_run_cleanup(self.failing_cleanup))
87
92
self.assertContainsRe(
89
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
94
"brz: warning: Cleanup failed:.*failing_cleanup goes boom")
91
96
def test_prior_error_cleanup_succeeds(self):
92
97
"""Calling _run_cleanup from a finally block will not interfere with an
182
187
cleanups = self.make_two_failing_cleanup_funcs()
183
188
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
185
190
self.assertLogContains('Cleanup failed:.*ErrorB')
186
self.assertFalse('ErrorA' in self.get_log())
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$')
188
198
def make_two_failing_cleanup_funcs(self):
190
200
raise ErrorA('Error A')
192
203
raise ErrorB('Error B')
193
204
return [(raise_a, (), {}), (raise_b, (), {})]
195
206
def test_multiple_cleanup_failures_debug_flag(self):
197
trace.push_log_file(log)
198
207
debug.debug_flags.add('cleanup')
199
208
cleanups = self.make_two_failing_cleanup_funcs()
200
209
self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
211
trace_value = self.get_log()
202
212
self.assertContainsRe(
203
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
204
self.assertEqual(1, log.getvalue().count('bzr: warning:'),
213
trace_value, "brz: warning: Cleanup failed:.*Error B\n")
214
self.assertEqual(1, trace_value.count('brz: warning:'))
207
216
def test_func_and_cleanup_errors_debug_flag(self):
209
trace.push_log_file(log)
210
217
debug.debug_flags.add('cleanup')
211
218
cleanups = self.make_two_failing_cleanup_funcs()
212
219
self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
214
self.assertContainsRe(
215
log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
216
self.assertContainsRe(
217
log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
218
self.assertEqual(2, log.getvalue().count('bzr: warning:'))
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:'))
220
228
def test_func_may_mutate_cleanups(self):
221
229
"""The main func may mutate the cleanups before it returns.
223
231
This allows a function to gradually add cleanups as it acquires
224
232
resources, rather than planning all the cleanups up-front. The
225
233
OperationWithCleanups helper relies on this working.
227
235
cleanups_list = []
228
237
def func_that_adds_cleanups():
229
238
self.call_log.append('func_that_adds_cleanups')
230
239
cleanups_list.append((self.no_op_cleanup, (), {}))
238
247
"""The -Dcleanup debug flag causes cleanup errors to be reported to the
242
trace.push_log_file(log)
243
250
debug.debug_flags.add('cleanup')
244
251
self.assertRaises(ZeroDivisionError, _do_with_cleanups,
245
[(self.failing_cleanup, (), {})], self.failing_func)
252
[(self.failing_cleanup, (), {})], self.failing_func)
253
trace_value = self.get_log()
246
254
self.assertContainsRe(
248
"bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
249
self.assertEqual(1, log.getvalue().count('bzr: warning:'))
252
class ErrorA(Exception): pass
253
class ErrorB(Exception): pass
256
"brz: warning: Cleanup failed:.*failing_cleanup goes boom")
257
self.assertEqual(1, trace_value.count('brz: warning:'))
256
260
class TestOperationWithCleanups(CleanupsTestCase):
274
279
self.assertEqual('result', result)
275
280
self.assertEqual(
276
281
[('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
277
'cleanup 4'], call_log)
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])