/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_cleanup.py

  • Committer: Andrew Bennetts
  • Date: 2009-10-29 00:25:26 UTC
  • mfrom: (4775 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4778.
  • Revision ID: andrew.bennetts@canonical.com-20091029002526-0d0hb1jqt1qvd5a6
Merge lp:bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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