/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: 2019-02-15 17:41:17 UTC
  • mto: (7290.1.2 work)
  • mto: This revision was merged to the branch mainline in revision 7295.
  • Revision ID: jelmer@jelmer.uk-20190215174117-o9w1am2z88mg9g1q
Update references to home location.

~/.config/breezy rather than ~/.bazaar.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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
18
17
import re
19
18
 
20
 
from bzrlib.cleanup import (
 
19
from ..cleanup import (
21
20
    _do_with_cleanups,
22
21
    _run_cleanup,
 
22
    ObjectWithCleanups,
23
23
    OperationWithCleanups,
24
24
    )
25
 
from bzrlib.tests import TestCase
26
 
from bzrlib import (
 
25
from .. import (
27
26
    debug,
28
 
    trace,
 
27
    tests,
29
28
    )
30
29
 
31
30
 
32
 
class CleanupsTestCase(TestCase):
 
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):
33
40
 
34
41
    def setUp(self):
35
42
        super(CleanupsTestCase, self).setUp()
80
87
        """The -Dcleanup debug flag causes cleanup errors to be reported to the
81
88
        user.
82
89
        """
83
 
        log = StringIO()
84
 
        trace.push_log_file(log)
85
90
        debug.debug_flags.add('cleanup')
86
91
        self.assertFalse(_run_cleanup(self.failing_cleanup))
87
92
        self.assertContainsRe(
88
 
            log.getvalue(),
89
 
            "bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
 
93
            self.get_log(),
 
94
            "brz: warning: Cleanup failed:.*failing_cleanup goes boom")
90
95
 
91
96
    def test_prior_error_cleanup_succeeds(self):
92
97
        """Calling _run_cleanup from a finally block will not interfere with an
94
99
        """
95
100
        def failing_operation():
96
101
            try:
97
 
                1/0
 
102
                1 / 0
98
103
            finally:
99
104
                _run_cleanup(self.no_op_cleanup)
100
105
        self.assertRaises(ZeroDivisionError, failing_operation)
109
114
        """
110
115
        def failing_operation():
111
116
            try:
112
 
                1/0
 
117
                1 / 0
113
118
            finally:
114
119
                _run_cleanup(self.failing_cleanup)
115
120
        self.assertRaises(ZeroDivisionError, failing_operation)
139
144
 
140
145
    def failing_func(self):
141
146
        self.call_log.append('failing_func')
142
 
        1/0
 
147
        1 / 0
143
148
 
144
149
    def test_func_error_propagates(self):
145
150
        """Errors from the main function are propagated (after running
181
186
        """
182
187
        cleanups = self.make_two_failing_cleanup_funcs()
183
188
        self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
184
 
            self.trivial_func)
 
189
                          self.trivial_func)
185
190
        self.assertLogContains('Cleanup failed:.*ErrorB')
186
 
        self.assertFalse('ErrorA' in self.get_log())
 
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$')
187
197
 
188
198
    def make_two_failing_cleanup_funcs(self):
189
199
        def raise_a():
190
200
            raise ErrorA('Error A')
 
201
 
191
202
        def raise_b():
192
203
            raise ErrorB('Error B')
193
204
        return [(raise_a, (), {}), (raise_b, (), {})]
194
205
 
195
206
    def test_multiple_cleanup_failures_debug_flag(self):
196
 
        log = StringIO()
197
 
        trace.push_log_file(log)
198
207
        debug.debug_flags.add('cleanup')
199
208
        cleanups = self.make_two_failing_cleanup_funcs()
200
209
        self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
201
 
            self.trivial_func)
 
210
                          self.trivial_func)
 
211
        trace_value = self.get_log()
202
212
        self.assertContainsRe(
203
 
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
204
 
        self.assertEqual(1, log.getvalue().count('bzr: warning:'),
205
 
                log.getvalue())
 
213
            trace_value, "brz: warning: Cleanup failed:.*Error B\n")
 
214
        self.assertEqual(1, trace_value.count('brz: warning:'))
206
215
 
207
216
    def test_func_and_cleanup_errors_debug_flag(self):
208
 
        log = StringIO()
209
 
        trace.push_log_file(log)
210
217
        debug.debug_flags.add('cleanup')
211
218
        cleanups = self.make_two_failing_cleanup_funcs()
212
219
        self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
213
 
            self.failing_func)
214
 
        self.assertContainsRe(
215
 
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
216
 
        self.assertContainsRe(
217
 
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
218
 
        self.assertEqual(2, log.getvalue().count('bzr: warning:'))
 
220
                          self.failing_func)
 
221
        trace_value = self.get_log()
 
222
        self.assertContainsRe(
 
223
            trace_value, "brz: warning: Cleanup failed:.*Error A\n")
 
224
        self.assertContainsRe(
 
225
            trace_value, "brz: warning: Cleanup failed:.*Error B\n")
 
226
        self.assertEqual(2, trace_value.count('brz: warning:'))
219
227
 
220
228
    def test_func_may_mutate_cleanups(self):
221
229
        """The main func may mutate the cleanups before it returns.
222
 
        
 
230
 
223
231
        This allows a function to gradually add cleanups as it acquires
224
232
        resources, rather than planning all the cleanups up-front.  The
225
233
        OperationWithCleanups helper relies on this working.
226
234
        """
227
235
        cleanups_list = []
 
236
 
228
237
        def func_that_adds_cleanups():
229
238
            self.call_log.append('func_that_adds_cleanups')
230
239
            cleanups_list.append((self.no_op_cleanup, (), {}))
238
247
        """The -Dcleanup debug flag causes cleanup errors to be reported to the
239
248
        user.
240
249
        """
241
 
        log = StringIO()
242
 
        trace.push_log_file(log)
243
250
        debug.debug_flags.add('cleanup')
244
251
        self.assertRaises(ZeroDivisionError, _do_with_cleanups,
245
 
            [(self.failing_cleanup, (), {})], self.failing_func)
 
252
                          [(self.failing_cleanup, (), {})], self.failing_func)
 
253
        trace_value = self.get_log()
246
254
        self.assertContainsRe(
247
 
            log.getvalue(),
248
 
            "bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
249
 
        self.assertEqual(1, log.getvalue().count('bzr: warning:'))
250
 
 
251
 
 
252
 
class ErrorA(Exception): pass
253
 
class ErrorB(Exception): pass
 
255
            trace_value,
 
256
            "brz: warning: Cleanup failed:.*failing_cleanup goes boom")
 
257
        self.assertEqual(1, trace_value.count('brz: warning:'))
254
258
 
255
259
 
256
260
class TestOperationWithCleanups(CleanupsTestCase):
262
266
        cleanup added during the func is run first.
263
267
        """
264
268
        call_log = []
 
269
 
265
270
        def func(op, foo):
266
271
            call_log.append(('func called', foo))
267
272
            op.add_cleanup(call_log.append, 'cleanup 2')
274
279
        self.assertEqual('result', result)
275
280
        self.assertEqual(
276
281
            [('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
277
 
            'cleanup 4'], call_log)
278
 
 
 
282
             'cleanup 4'], call_log)
 
283
 
 
284
 
 
285
class SampleWithCleanups(ObjectWithCleanups):
 
286
    """Minimal ObjectWithCleanups subclass."""
 
287
 
 
288
 
 
289
class TestObjectWithCleanups(tests.TestCase):
 
290
 
 
291
    def test_object_with_cleanups(self):
 
292
        a = []
 
293
        s = SampleWithCleanups()
 
294
        s.add_cleanup(a.append, 42)
 
295
        s.cleanup_now()
 
296
        self.assertEqual(a, [42])