/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: Marius Kruger
  • Date: 2010-07-10 21:28:56 UTC
  • mto: (5384.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5385.
  • Revision ID: marius.kruger@enerweb.co.za-20100710212856-uq4ji3go0u5se7hx
* Update documentation
* add NEWS

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_cleanup.py
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
16
16
 
 
17
from cStringIO import StringIO
17
18
import re
18
19
 
19
 
from ..cleanup import (
20
 
    ExitStack,
21
 
    )
22
 
from ..sixish import PY3
23
 
from .. import (
24
 
    tests,
25
 
    )
26
 
 
27
 
from contextlib import contextmanager
28
 
 
29
 
 
30
 
check_exception_chaining = PY3
31
 
 
32
 
 
33
 
# Imported from contextlib2's test_contextlib2.py
34
 
# Copyright: Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008,
35
 
#   2009, 2010, 2011 Python Software Foundation
36
 
#
37
 
# PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2
38
 
# --------------------------------------------
39
 
# .
40
 
# 1. This LICENSE AGREEMENT is between the Python Software Foundation
41
 
# ("PSF"), and the Individual or Organization ("Licensee") accessing and
42
 
# otherwise using this software ("Python") in source or binary form and
43
 
# its associated documentation.
44
 
# .
45
 
# 2. Subject to the terms and conditions of this License Agreement, PSF hereby
46
 
# grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce,
47
 
# analyze, test, perform and/or display publicly, prepare derivative works,
48
 
# distribute, and otherwise use Python alone or in any derivative version,
49
 
# provided, however, that PSF's License Agreement and PSF's notice of copyright,
50
 
# i.e., "Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
51
 
# 2011 Python Software Foundation; All Rights Reserved" are retained in Python
52
 
# alone or in any derivative version prepared by Licensee.
53
 
# .
54
 
# 3. In the event Licensee prepares a derivative work that is based on
55
 
# or incorporates Python or any part thereof, and wants to make
56
 
# the derivative work available to others as provided herein, then
57
 
# Licensee hereby agrees to include in any such work a brief summary of
58
 
# the changes made to Python.
59
 
# .
60
 
# 4. PSF is making Python available to Licensee on an "AS IS"
61
 
# basis.  PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR
62
 
# IMPLIED.  BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND
63
 
# DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS
64
 
# FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT
65
 
# INFRINGE ANY THIRD PARTY RIGHTS.
66
 
# .
67
 
# 5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON
68
 
# FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS
69
 
# A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON,
70
 
# OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF.
71
 
# .
72
 
# 6. This License Agreement will automatically terminate upon a material
73
 
# breach of its terms and conditions.
74
 
# .
75
 
# 7. Nothing in this License Agreement shall be deemed to create any
76
 
# relationship of agency, partnership, or joint venture between PSF and
77
 
# Licensee.  This License Agreement does not grant permission to use PSF
78
 
# trademarks or trade name in a trademark sense to endorse or promote
79
 
# products or services of Licensee, or any third party.
80
 
# .
81
 
# 8. By copying, installing or otherwise using Python, Licensee
82
 
# agrees to be bound by the terms and conditions of this License
83
 
# Agreement.
84
 
 
85
 
 
86
 
class TestExitStack(tests.TestCase):
87
 
 
88
 
    def test_no_resources(self):
89
 
        with ExitStack():
90
 
            pass
91
 
 
92
 
    def test_callback(self):
93
 
        expected = [
94
 
            ((), {}),
95
 
            ((1,), {}),
96
 
            ((1, 2), {}),
97
 
            ((), dict(example=1)),
98
 
            ((1,), dict(example=1)),
99
 
            ((1, 2), dict(example=1)),
100
 
        ]
101
 
        result = []
102
 
        def _exit(*args, **kwds):
103
 
            """Test metadata propagation"""
104
 
            result.append((args, kwds))
105
 
        with ExitStack() as stack:
106
 
            for args, kwds in reversed(expected):
107
 
                if args and kwds:
108
 
                    f = stack.callback(_exit, *args, **kwds)
109
 
                elif args:
110
 
                    f = stack.callback(_exit, *args)
111
 
                elif kwds:
112
 
                    f = stack.callback(_exit, **kwds)
113
 
                else:
114
 
                    f = stack.callback(_exit)
115
 
                self.assertIs(f, _exit)
116
 
        self.assertEqual(result, expected)
117
 
 
118
 
    def test_push(self):
119
 
        exc_raised = ZeroDivisionError
120
 
        def _expect_exc(exc_type, exc, exc_tb):
121
 
            self.assertIs(exc_type, exc_raised)
122
 
        def _suppress_exc(*exc_details):
123
 
            return True
124
 
        def _expect_ok(exc_type, exc, exc_tb):
125
 
            self.assertIsNone(exc_type)
126
 
            self.assertIsNone(exc)
127
 
            self.assertIsNone(exc_tb)
128
 
        class ExitCM(object):
129
 
            def __init__(self, check_exc):
130
 
                self.check_exc = check_exc
131
 
            def __enter__(self):
132
 
                self.fail("Should not be called!")
133
 
            def __exit__(self, *exc_details):
134
 
                self.check_exc(*exc_details)
135
 
        with ExitStack() as stack:
136
 
            stack.push(_expect_ok)
137
 
            cm = ExitCM(_expect_ok)
138
 
            stack.push(cm)
139
 
            stack.push(_suppress_exc)
140
 
            cm = ExitCM(_expect_exc)
141
 
            stack.push(cm)
142
 
            stack.push(_expect_exc)
143
 
            stack.push(_expect_exc)
144
 
            1 / 0
145
 
 
146
 
    def test_enter_context(self):
147
 
        class TestCM(object):
148
 
            def __enter__(self):
149
 
                result.append(1)
150
 
            def __exit__(self, *exc_details):
151
 
                result.append(3)
152
 
 
153
 
        result = []
154
 
        cm = TestCM()
155
 
        with ExitStack() as stack:
156
 
            @stack.callback  # Registered first => cleaned up last
157
 
            def _exit():
158
 
                result.append(4)
159
 
            self.assertIsNotNone(_exit)
160
 
            stack.enter_context(cm)
161
 
            result.append(2)
162
 
        self.assertEqual(result, [1, 2, 3, 4])
163
 
 
164
 
    def test_close(self):
165
 
        result = []
166
 
        with ExitStack() as stack:
167
 
            @stack.callback
168
 
            def _exit():
169
 
                result.append(1)
170
 
            self.assertIsNotNone(_exit)
171
 
            stack.close()
172
 
            result.append(2)
173
 
        self.assertEqual(result, [1, 2])
174
 
 
175
 
    def test_pop_all(self):
176
 
        result = []
177
 
        with ExitStack() as stack:
178
 
            @stack.callback
179
 
            def _exit():
180
 
                result.append(3)
181
 
            self.assertIsNotNone(_exit)
182
 
            new_stack = stack.pop_all()
183
 
            result.append(1)
184
 
        result.append(2)
185
 
        new_stack.close()
186
 
        self.assertEqual(result, [1, 2, 3])
187
 
 
188
 
    def test_exit_raise(self):
189
 
        def _raise():
190
 
            with ExitStack() as stack:
191
 
                stack.push(lambda *exc: False)
192
 
                1 / 0
193
 
        self.assertRaises(ZeroDivisionError, _raise)
194
 
 
195
 
    def test_exit_suppress(self):
196
 
        with ExitStack() as stack:
197
 
            stack.push(lambda *exc: True)
198
 
            1 / 0
199
 
 
200
 
    def test_exit_exception_chaining_reference(self):
201
 
        # Sanity check to make sure that ExitStack chaining matches
202
 
        # actual nested with statements
203
 
        class RaiseExc:
204
 
            def __init__(self, exc):
205
 
                self.exc = exc
206
 
            def __enter__(self):
207
 
                return self
208
 
            def __exit__(self, *exc_details):
209
 
                raise self.exc
210
 
 
211
 
        class RaiseExcWithContext:
212
 
            def __init__(self, outer, inner):
213
 
                self.outer = outer
214
 
                self.inner = inner
215
 
            def __enter__(self):
216
 
                return self
217
 
            def __exit__(self, *exc_details):
218
 
                try:
219
 
                    raise self.inner
220
 
                except:
221
 
                    raise self.outer
222
 
 
223
 
        class SuppressExc:
224
 
            def __enter__(self):
225
 
                return self
226
 
            def __exit__(self, *exc_details):
227
 
                self.__class__.saved_details = exc_details
228
 
                return True
229
 
 
230
 
        try:
231
 
            with RaiseExc(IndexError):
232
 
                with RaiseExcWithContext(KeyError, AttributeError):
233
 
                    with SuppressExc():
234
 
                        with RaiseExc(ValueError):
235
 
                            1 / 0
236
 
        except IndexError as exc:
237
 
            if check_exception_chaining:
238
 
                self.assertIsInstance(exc.__context__, KeyError)
239
 
                self.assertIsInstance(exc.__context__.__context__, AttributeError)
240
 
                # Inner exceptions were suppressed
241
 
                self.assertIsNone(exc.__context__.__context__.__context__)
242
 
        else:
243
 
            self.fail("Expected IndexError, but no exception was raised")
244
 
        # Check the inner exceptions
245
 
        inner_exc = SuppressExc.saved_details[1]
246
 
        self.assertIsInstance(inner_exc, ValueError)
247
 
        if check_exception_chaining:
248
 
            self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
249
 
 
250
 
    def test_exit_exception_chaining(self):
251
 
        # Ensure exception chaining matches the reference behaviour
252
 
        def raise_exc(exc):
253
 
            raise exc
254
 
 
255
 
        saved_details = [None]
256
 
        def suppress_exc(*exc_details):
257
 
            saved_details[0] = exc_details
258
 
            return True
259
 
 
260
 
        try:
261
 
            with ExitStack() as stack:
262
 
                stack.callback(raise_exc, IndexError)
263
 
                stack.callback(raise_exc, KeyError)
264
 
                stack.callback(raise_exc, AttributeError)
265
 
                stack.push(suppress_exc)
266
 
                stack.callback(raise_exc, ValueError)
267
 
                1 / 0
268
 
        except IndexError as exc:
269
 
            if check_exception_chaining:
270
 
                self.assertIsInstance(exc.__context__, KeyError)
271
 
                self.assertIsInstance(exc.__context__.__context__, AttributeError)
272
 
                # Inner exceptions were suppressed
273
 
                self.assertIsNone(exc.__context__.__context__.__context__)
274
 
        else:
275
 
            self.fail("Expected IndexError, but no exception was raised")
276
 
        # Check the inner exceptions
277
 
        inner_exc = saved_details[0][1]
278
 
        self.assertIsInstance(inner_exc, ValueError)
279
 
        if check_exception_chaining:
280
 
            self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
281
 
 
282
 
    def test_exit_exception_non_suppressing(self):
283
 
        # http://bugs.python.org/issue19092
284
 
        def raise_exc(exc):
285
 
            raise exc
286
 
 
287
 
        def suppress_exc(*exc_details):
288
 
            return True
289
 
 
290
 
        try:
291
 
            with ExitStack() as stack:
292
 
                stack.callback(lambda: None)
293
 
                stack.callback(raise_exc, IndexError)
294
 
        except Exception as exc:
295
 
            self.assertIsInstance(exc, IndexError)
296
 
        else:
297
 
            self.fail("Expected IndexError, but no exception was raised")
298
 
 
299
 
        try:
300
 
            with ExitStack() as stack:
301
 
                stack.callback(raise_exc, KeyError)
302
 
                stack.push(suppress_exc)
303
 
                stack.callback(raise_exc, IndexError)
304
 
        except Exception as exc:
305
 
            self.assertIsInstance(exc, KeyError)
306
 
        else:
307
 
            self.fail("Expected KeyError, but no exception was raised")
308
 
 
309
 
    def test_exit_exception_with_correct_context(self):
310
 
        # http://bugs.python.org/issue20317
311
 
        @contextmanager
312
 
        def gets_the_context_right(exc):
313
 
            try:
314
 
                yield
315
 
            finally:
316
 
                raise exc
317
 
 
318
 
        exc1 = Exception(1)
319
 
        exc2 = Exception(2)
320
 
        exc3 = Exception(3)
321
 
        exc4 = Exception(4)
322
 
 
323
 
        # The contextmanager already fixes the context, so prior to the
324
 
        # fix, ExitStack would try to fix it *again* and get into an
325
 
        # infinite self-referential loop
326
 
        try:
327
 
            with ExitStack() as stack:
328
 
                stack.enter_context(gets_the_context_right(exc4))
329
 
                stack.enter_context(gets_the_context_right(exc3))
330
 
                stack.enter_context(gets_the_context_right(exc2))
331
 
                raise exc1
332
 
        except Exception as exc:
333
 
            self.assertIs(exc, exc4)
334
 
            if check_exception_chaining:
335
 
                self.assertIs(exc.__context__, exc3)
336
 
                self.assertIs(exc.__context__.__context__, exc2)
337
 
                self.assertIs(exc.__context__.__context__.__context__, exc1)
338
 
                self.assertIsNone(
339
 
                    exc.__context__.__context__.__context__.__context__)
340
 
 
341
 
    def test_exit_exception_with_existing_context(self):
342
 
        # Addresses a lack of test coverage discovered after checking in a
343
 
        # fix for issue 20317 that still contained debugging code.
344
 
        def raise_nested(inner_exc, outer_exc):
345
 
            try:
346
 
                raise inner_exc
347
 
            finally:
348
 
                raise outer_exc
349
 
        exc1 = Exception(1)
350
 
        exc2 = Exception(2)
351
 
        exc3 = Exception(3)
352
 
        exc4 = Exception(4)
353
 
        exc5 = Exception(5)
354
 
        try:
355
 
            with ExitStack() as stack:
356
 
                stack.callback(raise_nested, exc4, exc5)
357
 
                stack.callback(raise_nested, exc2, exc3)
358
 
                raise exc1
359
 
        except Exception as exc:
360
 
            self.assertIs(exc, exc5)
361
 
            if check_exception_chaining:
362
 
                self.assertIs(exc.__context__, exc4)
363
 
                self.assertIs(exc.__context__.__context__, exc3)
364
 
                self.assertIs(exc.__context__.__context__.__context__, exc2)
365
 
                self.assertIs(
366
 
                    exc.__context__.__context__.__context__.__context__, exc1)
367
 
                self.assertIsNone(
368
 
                    exc.__context__.__context__.__context__.__context__.__context__)
369
 
 
370
 
    def test_body_exception_suppress(self):
371
 
        def suppress_exc(*exc_details):
372
 
            return True
373
 
        try:
374
 
            with ExitStack() as stack:
375
 
                stack.push(suppress_exc)
376
 
                1 / 0
377
 
        except IndexError as exc:
378
 
            self.fail("Expected no exception, got IndexError")
379
 
 
380
 
    def test_exit_exception_chaining_suppress(self):
381
 
        with ExitStack() as stack:
382
 
            stack.push(lambda *exc: True)
383
 
            stack.push(lambda *exc: 1 / 0)
384
 
            stack.push(lambda *exc: {}[1])
385
 
 
386
 
    def test_excessive_nesting(self):
387
 
        # The original implementation would die with RecursionError here
388
 
        with ExitStack() as stack:
389
 
            for i in range(10000):
390
 
                stack.callback(int)
391
 
 
392
 
    def test_instance_bypass(self):
393
 
        class Example(object):
394
 
            pass
395
 
        cm = Example()
396
 
        cm.__exit__ = object()
397
 
        stack = ExitStack()
398
 
        self.assertRaises(AttributeError, stack.enter_context, cm)
399
 
        stack.push(cm)
400
 
        # self.assertIs(stack._exit_callbacks[-1], cm)
 
20
from bzrlib.cleanup import (
 
21
    _do_with_cleanups,
 
22
    _run_cleanup,
 
23
    ObjectWithCleanups,
 
24
    OperationWithCleanups,
 
25
    )
 
26
from bzrlib.tests import TestCase
 
27
from bzrlib import (
 
28
    debug,
 
29
    trace,
 
30
    )
 
31
 
 
32
 
 
33
class CleanupsTestCase(TestCase):
 
34
 
 
35
    def setUp(self):
 
36
        super(CleanupsTestCase, self).setUp()
 
37
        self.call_log = []
 
38
 
 
39
    def no_op_cleanup(self):
 
40
        self.call_log.append('no_op_cleanup')
 
41
 
 
42
    def assertLogContains(self, regex):
 
43
        self.assertContainsRe(self.get_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
        self.assertFalse('ErrorA' in self.get_log())
 
188
 
 
189
    def make_two_failing_cleanup_funcs(self):
 
190
        def raise_a():
 
191
            raise ErrorA('Error A')
 
192
        def raise_b():
 
193
            raise ErrorB('Error B')
 
194
        return [(raise_a, (), {}), (raise_b, (), {})]
 
195
 
 
196
    def test_multiple_cleanup_failures_debug_flag(self):
 
197
        log = StringIO()
 
198
        trace.push_log_file(log)
 
199
        debug.debug_flags.add('cleanup')
 
200
        cleanups = self.make_two_failing_cleanup_funcs()
 
201
        self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
 
202
            self.trivial_func)
 
203
        self.assertContainsRe(
 
204
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
 
205
        self.assertEqual(1, log.getvalue().count('bzr: warning:'),
 
206
                log.getvalue())
 
207
 
 
208
    def test_func_and_cleanup_errors_debug_flag(self):
 
209
        log = StringIO()
 
210
        trace.push_log_file(log)
 
211
        debug.debug_flags.add('cleanup')
 
212
        cleanups = self.make_two_failing_cleanup_funcs()
 
213
        self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
 
214
            self.failing_func)
 
215
        self.assertContainsRe(
 
216
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
 
217
        self.assertContainsRe(
 
218
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
 
219
        self.assertEqual(2, log.getvalue().count('bzr: warning:'))
 
220
 
 
221
    def test_func_may_mutate_cleanups(self):
 
222
        """The main func may mutate the cleanups before it returns.
 
223
        
 
224
        This allows a function to gradually add cleanups as it acquires
 
225
        resources, rather than planning all the cleanups up-front.  The
 
226
        OperationWithCleanups helper relies on this working.
 
227
        """
 
228
        cleanups_list = []
 
229
        def func_that_adds_cleanups():
 
230
            self.call_log.append('func_that_adds_cleanups')
 
231
            cleanups_list.append((self.no_op_cleanup, (), {}))
 
232
            return 'result'
 
233
        result = _do_with_cleanups(cleanups_list, func_that_adds_cleanups)
 
234
        self.assertEqual('result', result)
 
235
        self.assertEqual(
 
236
            ['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
 
237
 
 
238
    def test_cleanup_error_debug_flag(self):
 
239
        """The -Dcleanup debug flag causes cleanup errors to be reported to the
 
240
        user.
 
241
        """
 
242
        log = StringIO()
 
243
        trace.push_log_file(log)
 
244
        debug.debug_flags.add('cleanup')
 
245
        self.assertRaises(ZeroDivisionError, _do_with_cleanups,
 
246
            [(self.failing_cleanup, (), {})], self.failing_func)
 
247
        self.assertContainsRe(
 
248
            log.getvalue(),
 
249
            "bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
 
250
        self.assertEqual(1, log.getvalue().count('bzr: warning:'))
 
251
 
 
252
 
 
253
class ErrorA(Exception): pass
 
254
class ErrorB(Exception): pass
 
255
 
 
256
 
 
257
class TestOperationWithCleanups(CleanupsTestCase):
 
258
 
 
259
    def test_cleanup_ordering(self):
 
260
        """Cleanups are added in LIFO order.
 
261
 
 
262
        So cleanups added before run is called are run last, and the last
 
263
        cleanup added during the func is run first.
 
264
        """
 
265
        call_log = []
 
266
        def func(op, foo):
 
267
            call_log.append(('func called', foo))
 
268
            op.add_cleanup(call_log.append, 'cleanup 2')
 
269
            op.add_cleanup(call_log.append, 'cleanup 1')
 
270
            return 'result'
 
271
        owc = OperationWithCleanups(func)
 
272
        owc.add_cleanup(call_log.append, 'cleanup 4')
 
273
        owc.add_cleanup(call_log.append, 'cleanup 3')
 
274
        result = owc.run('foo')
 
275
        self.assertEqual('result', result)
 
276
        self.assertEqual(
 
277
            [('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
 
278
            'cleanup 4'], call_log)
 
279
 
 
280
 
 
281
class SampleWithCleanups(ObjectWithCleanups):
 
282
 
 
283
    pass
 
284
 
 
285
 
 
286
class TestObjectWithCleanups(TestCase):
 
287
 
 
288
    def test_object_with_cleanups(self):
 
289
        a = []
 
290
        s = SampleWithCleanups()
 
291
        s.add_cleanup(a.append, 42)
 
292
        s.cleanup_now()
 
293
        self.assertEqual(a, [42])