/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-15 02:53:30 UTC
  • mto: This revision was merged to the branch mainline in revision 4775.
  • Revision ID: andrew.bennetts@canonical.com-20091015025330-9cwu80sdkttu7v58
Add OperationWithCleanups helper, use it to make commit.py simpler and more robust.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_cleanup.py
1
 
# Copyright (C) 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
18
18
import re
19
19
 
20
20
from bzrlib.cleanup import (
21
 
    _do_with_cleanups,
 
21
    do_with_cleanups,
22
22
    _run_cleanup,
23
 
    OperationWithCleanups,
24
23
    )
25
24
from bzrlib.tests import TestCase
26
25
from bzrlib import (
39
38
        self.call_log.append('no_op_cleanup')
40
39
 
41
40
    def assertLogContains(self, regex):
42
 
        self.assertContainsRe(self.get_log(), regex, re.DOTALL)
 
41
        log = self._get_log(keep_log_file=True)
 
42
        self.assertContainsRe(log, regex, re.DOTALL)
43
43
 
44
44
    def failing_cleanup(self):
45
45
        self.call_log.append('failing_cleanup')
116
116
        self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
117
117
 
118
118
 
 
119
#class TestRunCleanupReportingErrors(CleanupsTestCase):
 
120
#
 
121
#    def test_cleanup_error_reported(self):
 
122
#        xxx
 
123
 
 
124
 
119
125
class TestDoWithCleanups(CleanupsTestCase):
120
126
 
121
127
    def trivial_func(self):
123
129
        return 'trivial result'
124
130
 
125
131
    def test_runs_func(self):
126
 
        """_do_with_cleanups runs the function it is given, and returns the
 
132
        """do_with_cleanups runs the function it is given, and returns the
127
133
        result.
128
134
        """
129
 
        result = _do_with_cleanups([], self.trivial_func)
 
135
        result = do_with_cleanups(self.trivial_func, [])
130
136
        self.assertEqual('trivial result', result)
131
137
 
132
138
    def test_runs_cleanups(self):
133
139
        """Cleanup functions are run (in the given order)."""
134
 
        cleanup_func_1 = (self.call_log.append, ('cleanup 1',), {})
135
 
        cleanup_func_2 = (self.call_log.append, ('cleanup 2',), {})
136
 
        _do_with_cleanups([cleanup_func_1, cleanup_func_2], self.trivial_func)
 
140
        cleanup_func_1 = lambda: self.call_log.append('cleanup 1')
 
141
        cleanup_func_2 = lambda: self.call_log.append('cleanup 2')
 
142
        do_with_cleanups(self.trivial_func, [cleanup_func_1, cleanup_func_2])
137
143
        self.assertEqual(
138
144
            ['trivial_func', 'cleanup 1', 'cleanup 2'], self.call_log)
139
145
 
146
152
        cleanups).
147
153
        """
148
154
        self.assertRaises(
149
 
            ZeroDivisionError, _do_with_cleanups,
150
 
            [(self.no_op_cleanup, (), {})], self.failing_func)
 
155
            ZeroDivisionError, do_with_cleanups, self.failing_func,
 
156
            [self.no_op_cleanup])
151
157
        self.assertEqual(['failing_func', 'no_op_cleanup'], self.call_log)
152
158
 
153
159
    def test_func_error_trumps_cleanup_error(self):
157
163
        The cleanup error is be logged.
158
164
        """
159
165
        self.assertRaises(
160
 
            ZeroDivisionError, _do_with_cleanups,
161
 
            [(self.failing_cleanup, (), {})], self.failing_func)
 
166
            ZeroDivisionError, do_with_cleanups, self.failing_func,
 
167
            [self.failing_cleanup])
162
168
        self.assertLogContains('Cleanup failed:.*failing_cleanup goes boom')
163
169
 
164
170
    def test_func_passes_and_error_from_cleanup(self):
166
172
        raise an error.  Later cleanups are still executed.
167
173
        """
168
174
        exc = self.assertRaises(
169
 
            Exception, _do_with_cleanups,
170
 
            [(self.failing_cleanup, (), {}), (self.no_op_cleanup, (), {})],
171
 
            self.trivial_func)
 
175
            Exception, do_with_cleanups, self.trivial_func,
 
176
            [self.failing_cleanup, self.no_op_cleanup])
172
177
        self.assertEqual('failing_cleanup goes boom!', exc.args[0])
173
178
        self.assertEqual(
174
179
            ['trivial_func', 'failing_cleanup', 'no_op_cleanup'],
180
185
        logged.
181
186
        """
182
187
        cleanups = self.make_two_failing_cleanup_funcs()
183
 
        self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
184
 
            self.trivial_func)
 
188
        self.assertRaises(ErrorA, do_with_cleanups, self.trivial_func,
 
189
            cleanups)
185
190
        self.assertLogContains('Cleanup failed:.*ErrorB')
186
 
        self.assertFalse('ErrorA' in self.get_log())
 
191
        log = self._get_log(keep_log_file=True)
 
192
        self.assertFalse('ErrorA' in log)
187
193
 
188
194
    def make_two_failing_cleanup_funcs(self):
189
195
        def raise_a():
190
196
            raise ErrorA('Error A')
191
197
        def raise_b():
192
198
            raise ErrorB('Error B')
193
 
        return [(raise_a, (), {}), (raise_b, (), {})]
 
199
        return [raise_a, raise_b]
194
200
 
195
201
    def test_multiple_cleanup_failures_debug_flag(self):
196
202
        log = StringIO()
197
203
        trace.push_log_file(log)
198
204
        debug.debug_flags.add('cleanup')
199
205
        cleanups = self.make_two_failing_cleanup_funcs()
200
 
        self.assertRaises(ErrorA, _do_with_cleanups, cleanups,
201
 
            self.trivial_func)
 
206
        self.assertRaises(ErrorA, do_with_cleanups, self.trivial_func, cleanups)
202
207
        self.assertContainsRe(
203
208
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error B\n")
204
209
        self.assertEqual(1, log.getvalue().count('bzr: warning:'),
209
214
        trace.push_log_file(log)
210
215
        debug.debug_flags.add('cleanup')
211
216
        cleanups = self.make_two_failing_cleanup_funcs()
212
 
        self.assertRaises(ZeroDivisionError, _do_with_cleanups, cleanups,
213
 
            self.failing_func)
 
217
        self.assertRaises(ZeroDivisionError, do_with_cleanups,
 
218
            self.failing_func, cleanups)
214
219
        self.assertContainsRe(
215
220
            log.getvalue(), "bzr: warning: Cleanup failed:.*Error A\n")
216
221
        self.assertContainsRe(
221
226
        """The main func may mutate the cleanups before it returns.
222
227
        
223
228
        This allows a function to gradually add cleanups as it acquires
224
 
        resources, rather than planning all the cleanups up-front.  The
225
 
        OperationWithCleanups helper relies on this working.
 
229
        resources, rather than planning all the cleanups up-front.
226
230
        """
 
231
        # XXX: this is cute, but an object with an 'add_cleanup' method may
 
232
        # make a better API?
227
233
        cleanups_list = []
228
234
        def func_that_adds_cleanups():
229
235
            self.call_log.append('func_that_adds_cleanups')
230
 
            cleanups_list.append((self.no_op_cleanup, (), {}))
 
236
            cleanups_list.append(self.no_op_cleanup)
231
237
            return 'result'
232
 
        result = _do_with_cleanups(cleanups_list, func_that_adds_cleanups)
 
238
        result = do_with_cleanups(func_that_adds_cleanups, cleanups_list)
233
239
        self.assertEqual('result', result)
234
240
        self.assertEqual(
235
241
            ['func_that_adds_cleanups', 'no_op_cleanup'], self.call_log)
241
247
        log = StringIO()
242
248
        trace.push_log_file(log)
243
249
        debug.debug_flags.add('cleanup')
244
 
        self.assertRaises(ZeroDivisionError, _do_with_cleanups,
245
 
            [(self.failing_cleanup, (), {})], self.failing_func)
 
250
        self.assertRaises(ZeroDivisionError, do_with_cleanups,
 
251
            self.failing_func, [self.failing_cleanup])
246
252
        self.assertContainsRe(
247
253
            log.getvalue(),
248
254
            "bzr: warning: Cleanup failed:.*failing_cleanup goes boom")
251
257
 
252
258
class ErrorA(Exception): pass
253
259
class ErrorB(Exception): pass
254
 
 
255
 
 
256
 
class TestOperationWithCleanups(CleanupsTestCase):
257
 
 
258
 
    def test_cleanup_ordering(self):
259
 
        """Cleanups are added in LIFO order.
260
 
 
261
 
        So cleanups added before run is called are run last, and the last
262
 
        cleanup added during the func is run first.
263
 
        """
264
 
        call_log = []
265
 
        def func(op, foo):
266
 
            call_log.append(('func called', foo))
267
 
            op.add_cleanup(call_log.append, 'cleanup 2')
268
 
            op.add_cleanup(call_log.append, 'cleanup 1')
269
 
            return 'result'
270
 
        owc = OperationWithCleanups(func)
271
 
        owc.add_cleanup(call_log.append, 'cleanup 4')
272
 
        owc.add_cleanup(call_log.append, 'cleanup 3')
273
 
        result = owc.run('foo')
274
 
        self.assertEqual('result', result)
275
 
        self.assertEqual(
276
 
            [('func called', 'foo'), 'cleanup 1', 'cleanup 2', 'cleanup 3',
277
 
            'cleanup 4'], call_log)
278