/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: 2018-02-18 21:42:57 UTC
  • mto: This revision was merged to the branch mainline in revision 6859.
  • Revision ID: jelmer@jelmer.uk-20180218214257-jpevutp1wa30tz3v
Update TODO to reference Breezy, not Bazaar.

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