/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 breezy/tests/test_cleanup.py

  • Committer: Jelmer Vernooij
  • Date: 2017-07-21 13:20:17 UTC
  • mfrom: (6733.1.1 move-errors-config)
  • Revision ID: jelmer@jelmer.uk-20170721132017-oratmmxasovq4r1q
Merge lp:~jelmer/brz/move-errors-config.

Show diffs side-by-side

added added

removed removed

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