« back to all changes in this revision
Viewing changes to bzrlib/tests/test_cleanup.py
- browse files at revision 4634.85.3
- compare with another revision
- compare with revision 6011
- download diff
- download tarball
- view history from revision 4634.85.3
- Committer: Andrew Bennetts
- Date: 2009-09-23 04:21:10 UTC
- mto: (4744.3.1 robust-cleanup-in-commit)
- mto: This revision was merged to the branch mainline in revision 4775.
- Revision ID: andrew.bennetts@canonical.com-20090923042110-gyf7tk9gs7iwi0r1
Lots more tests.
- files modified:
- bzrlib/cleanup.py
added
removed
bzrlib/tests/test_cleanup.py
17
17
import re
18
18
19
19
from bzrlib.tests import TestCase
20
from bzrlib.cleanup import run_cleanup
21
22
23
class TestCleanup(TestCase):
20
from bzrlib.cleanup import (
21
do_with_cleanups,
22
run_cleanup,
23
)
24
25
26
class CleanupsTestCase(TestCase):
27
28
def setUp(self):
29
super(CleanupsTestCase, self).setUp()
30
self.call_log = []
24
31
25
32
def no_op_cleanup(self):
26
self.cleanup_was_run = True
33
self.call_log.append('no_op_cleanup')
34
35
def assertLogContains(self, regex):
36
log = self._get_log(keep_log_file=True)
37
self.assertContainsRe(log, regex, re.DOTALL)
38
39
def failing_cleanup(self):
40
self.call_log.append('failing_cleanup')
41
raise Exception("failing_cleanup goes boom!")
42
43
44
class TestRunCleanup(CleanupsTestCase):
27
45
28
46
def test_no_errors(self):
47
"""The function passed to run_cleanup is run."""
29
48
self.assertTrue(run_cleanup(self.no_op_cleanup))
30
self.assertTrue(self.cleanup_was_run)
49
self.assertEqual(['no_op_cleanup'], self.call_log)
31
50
32
def assertLogContains(self, regex):
33
log = self._get_log(keep_log_file=True)
34
self.assertContainsRe(log, regex, re.DOTALL)
51
# def test_cleanup_with_args_kwargs(self):
35
52
36
53
def test_cleanup_error(self):
37
# Because python sucks, there's no way for run_cleanup to know if
38
# there's an existing exception in this situation:
39
# try:
40
# some_func()
41
# finally:
42
# run_cleanup(cleanup_func)
43
# So, the best run_cleanup can do is always log errors but never raise
44
# them.
54
"""An error from the cleanup function is logged by run_cleanup, but not
55
propagated.
56
57
This is there's no way for run_cleanup to know if there's an existing
58
exception in this situation::
59
try:
60
some_func()
61
finally:
62
run_cleanup(cleanup_func)
63
So, the best run_cleanup can do is always log errors but never raise
64
them.
65
"""
45
66
self.assertFalse(run_cleanup(self.failing_cleanup))
46
67
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
47
68
48
69
def test_prior_error_cleanup_succeeds(self):
70
"""Calling run_cleanup from a finally block will not interfere with an
71
exception from the try block.
72
"""
49
73
def failing_operation():
50
74
try:
51
75
1/0
52
76
finally:
53
77
run_cleanup(self.no_op_cleanup)
54
78
self.assertRaises(ZeroDivisionError, failing_operation)
55
self.assertTrue(self.cleanup_was_run)
56
57
def failing_cleanup(self):
58
raise Exception("failing_cleanup goes boom!")
79
self.assertEqual(['no_op_cleanup'], self.call_log)
59
80
60
81
def test_prior_error_cleanup_fails(self):
82
"""Calling run_cleanup from a finally block will not interfere with an
83
exception from the try block even when the cleanup itself raises an
84
exception.
85
86
The cleanup exception will be logged.
87
"""
61
88
def failing_operation():
62
89
try:
63
90
1/0
66
93
self.assertRaises(ZeroDivisionError, failing_operation)
67
94
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
68
95
96
97
#class TestRunCleanupReportingErrors(CleanupsTestCase):
98
#
69
99
# def test_cleanup_error_reported(self):
70
71
72
73
# do_with_cleanups tests to write:
74
# - runs func (& returns result)
75
# - runs cleanups, in order
76
# - error from func propagated with healthy cleanups
77
# - error from func trumps error from cleanups (cleanup errs logged, all
78
# cleanups run)
79
# - healthy func, one error from cleanups => all cleanups run, error
80
# propagated (& returns func result)
81
# - healthy func, multiple errs from cleanups => all cleanups run, first err
82
# propagated, subsequent logged (& returns func result)
83
# - ? what about -Dcleanup, does it influnced do_with_cleanups' behaviour?
84
# - func appending new cleanups to cleanup_funcs
85
100
# xxx
101
102
103
class TestDoWithCleanups(CleanupsTestCase):
104
105
def trivial_func(self):
106
self.call_log.append('trivial_func')
107
return 'trivial result'
108
109
def test_runs_func(self):
110
"""do_with_cleanups runs the function it is given, and returns the
111
result.
112
"""
113
result = do_with_cleanups(self.trivial_func, [])
114
self.assertEqual('trivial result', result)
115
116
def test_runs_cleanups(self):
117
"""Cleanup functions are run (in the given order)."""
118
cleanup_func_1 = lambda: self.call_log.append('cleanup 1')
119
cleanup_func_2 = lambda: self.call_log.append('cleanup 2')
120
do_with_cleanups(self.trivial_func, [cleanup_func_1, cleanup_func_2])
121
self.assertEqual(
122
['trivial_func', 'cleanup 1', 'cleanup 2'], self.call_log)
123
124
def failing_func(self):
125
self.call_log.append('failing_func')
126
1/0
127
128
def test_func_error_propagates(self):
129
"""Errors from the main function are propagated (after running
130
cleanups).
131
"""
132
self.assertRaises(
133
ZeroDivisionError, do_with_cleanups, self.failing_func,
134
[self.no_op_cleanup])
135
self.assertEqual(['failing_func', 'no_op_cleanup'], self.call_log)
136
137
def test_func_error_trumps_cleanup_error(self):
138
"""Errors from the main function a propagated even if a cleanup raises
139
an error.
140
141
The cleanup error is be logged.
142
"""
143
self.assertRaises(
144
ZeroDivisionError, do_with_cleanups, self.failing_func,
145
[self.failing_cleanup])
146
self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
147
148
def test_func_passes_and_error_from_cleanup(self):
149
"""An error from a cleanup is propagated when the main function doesn't
150
raise an error. Later cleanups are still executed.
151
"""
152
exc = self.assertRaises(
153
Exception, do_with_cleanups, self.trivial_func,
154
[self.failing_cleanup, self.no_op_cleanup])
155
self.assertEqual('failing_cleanup goes boom!', exc.args[0])
156
self.assertEqual(
157
['trivial_func', 'failing_cleanup', 'no_op_cleanup'],
158
self.call_log)
159
160
def test_multiple_cleanup_failures(self):
161
"""When multiple cleanups fail (as tends to happen when something has
162
gone wrong), the first error is propagated, and subsequent errors are
163
logged.
164
"""
165
class ErrorA(Exception): pass
166
class ErrorB(Exception): pass
167
def raise_a():
168
raise ErrorA()
169
def raise_b():
170
raise ErrorB()
171
self.assertRaises(ErrorA, do_with_cleanups, self.trivial_func,
172
[raise_a, raise_b])
173
self.assertLogContains('Cleanup failed:.*ErrorB')
174
log = self._get_log(keep_log_file=True)
175
self.assertFalse('ErrorA' in log)
176
177
def test_func_may_mutate_cleanups(self):
178
"""The main func may mutate the cleanups before it returns.
179
180
This allows a function to gradually add cleanups as it acquires
181
resources, rather than planning all the cleanups up-front.
182
"""
183
# XXX: this is cute, but an object with an 'add_cleanup' method may
184
# make a better API?
185
cleanups_list = []
186
def func_that_adds_cleanups():
187
self.call_log.append('func_that_adds_cleanups')
188
cleanups_list.append(self.no_op_cleanup)
189
return 'result'
190
result = do_with_cleanups(func_that_adds_cleanups, cleanups_list)
191
self.assertEqual('result', result)
192
self.assertEqual(
193
['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1