← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_branch/test_hooks.py

  • Committer: Robert Collins
  • Date: 2010-04-08 04:34:03 UTC
  • mfrom: (5138 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5139.
  • Revision ID: robertc@robertcollins.net-20100408043403-56z0d07vdqrx7f3t
Update bugfix for 528114 to trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_branch/test_hooks.py
16
16
 
17
17
"""Tests that branch classes implement hook callouts correctly."""
18
18
 
19
 
from bzrlib.branch import Branch, ChangeBranchTipParams
20
 
from bzrlib.errors import HookFailed, TipChangeRejected
21
 
from bzrlib.remote import RemoteBranch
22
 
from bzrlib.revision import NULL_REVISION
23
 
from bzrlib.smart import server
24
 
from bzrlib.tests import TestCaseWithMemoryTransport
25
 
 
26
 
 
27
 
class ChangeBranchTipTestCase(TestCaseWithMemoryTransport):
 
19
from bzrlib import (
 
20
    branch as _mod_branch,
 
21
    errors,
 
22
    remote,
 
23
    revision,
 
24
    tests,
 
25
    )
 
26
from bzrlib.tests import test_server
 
27
 
 
28
class ChangeBranchTipTestCase(tests.TestCaseWithMemoryTransport):
28
29
    """Base TestCase for testing pre/post_change_branch_tip hooks."""
29
30
 
30
31
    def install_logging_hook(self, prefix):
33
34
        :returns: the list that the calls will be appended to.
34
35
        """
35
36
        hook_calls = []
36
 
        Branch.hooks.install_named_hook(
 
37
        _mod_branch.Branch.hooks.install_named_hook(
37
38
            prefix + '_change_branch_tip', hook_calls.append, None)
38
39
        return hook_calls
39
40
 
53
54
        pre=False):
54
55
        if hook_calls is None:
55
56
            hook_calls = self.hook_calls
56
 
        if isinstance(branch, RemoteBranch):
 
57
        if isinstance(branch, remote.RemoteBranch):
57
58
            # For a remote branch, both the server and the client will raise
58
59
            # this hook, and we see both in the test environment. The remote
59
60
            # instance comes in between the clients - the client doe pre, the
72
73
 
73
74
    def setUp(self):
74
75
        self.hook_calls = []
75
 
        TestCaseWithMemoryTransport.setUp(self)
 
76
        super(TestSetRevisionHistoryHook, self).setUp()
76
77
 
77
78
    def capture_set_rh_hook(self, branch, rev_history):
78
79
        """Capture post set-rh hook calls to self.hook_calls.
84
85
 
85
86
    def test_set_rh_empty_history(self):
86
87
        branch = self.make_branch('source')
87
 
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
88
 
                                        None)
 
88
        _mod_branch.Branch.hooks.install_named_hook(
 
89
            'set_rh', self.capture_set_rh_hook, None)
89
90
        branch.set_revision_history([])
90
91
        expected_params = ('set_rh', branch, [], True)
91
92
        self.assertHookCalls(expected_params, branch)
98
99
        tree.commit('empty commit', rev_id='foo')
99
100
        tree.unlock()
100
101
        branch = tree.branch
101
 
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
102
 
                                        None)
 
102
        _mod_branch.Branch.hooks.install_named_hook(
 
103
            'set_rh', self.capture_set_rh_hook, None)
103
104
        # some branches require that their history be set to a revision in the
104
105
        # repository
105
106
        branch.set_revision_history(['f\xc2\xb5'])
108
109
 
109
110
    def test_set_rh_branch_is_locked(self):
110
111
        branch = self.make_branch('source')
111
 
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
112
 
                                        None)
 
112
        _mod_branch.Branch.hooks.install_named_hook(
 
113
            'set_rh', self.capture_set_rh_hook, None)
113
114
        branch.set_revision_history([])
114
115
        expected_params = ('set_rh', branch, [], True)
115
116
        self.assertHookCalls(expected_params, branch)
116
117
 
117
118
    def test_set_rh_calls_all_hooks_no_errors(self):
118
119
        branch = self.make_branch('source')
119
 
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
120
 
                                        None)
121
 
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
122
 
                                        None)
 
120
        _mod_branch.Branch.hooks.install_named_hook(
 
121
            'set_rh', self.capture_set_rh_hook, None)
 
122
        _mod_branch.Branch.hooks.install_named_hook(
 
123
            'set_rh', self.capture_set_rh_hook, None)
123
124
        branch.set_revision_history([])
124
125
        expected_calls = [('set_rh', branch, [], True),
125
126
            ('set_rh', branch, [], True),
126
127
            ]
127
 
        if isinstance(branch, RemoteBranch):
 
128
        if isinstance(branch, remote.RemoteBranch):
128
129
            # For a remote branch, both the server and the client will raise
129
130
            # set_rh, and the server will do so first because that is where
130
131
            # the change takes place.
134
135
            self.assertEqual(expected_calls, self.hook_calls)
135
136
 
136
137
 
137
 
class TestOpen(TestCaseWithMemoryTransport):
 
138
class TestOpen(tests.TestCaseWithMemoryTransport):
138
139
 
139
140
    def capture_hook(self, branch):
140
141
        self.hook_calls.append(branch)
141
142
 
142
143
    def install_hook(self):
143
144
        self.hook_calls = []
144
 
        Branch.hooks.install_named_hook('open', self.capture_hook, None)
 
145
        _mod_branch.Branch.hooks.install_named_hook(
 
146
            'open', self.capture_hook, None)
145
147
 
146
148
    def test_create(self):
147
149
        self.install_hook()
148
150
        b = self.make_branch('.')
149
 
        if isinstance(b, RemoteBranch):
 
151
        if isinstance(b, remote.RemoteBranch):
150
152
            # RemoteBranch creation:
151
 
            if (self.transport_readonly_server ==
152
 
                server.ReadonlySmartTCPServer_for_testing_v2_only):
 
153
            if (self.transport_readonly_server
 
154
                == test_server.ReadonlySmartTCPServer_for_testing_v2_only):
153
155
                # Older servers:
154
156
                self.assertEqual(3, len(self.hook_calls))
155
157
                # creates the branch via the VFS (for older servers)
170
172
    def test_open(self):
171
173
        branch_url = self.make_branch('.').bzrdir.root_transport.base
172
174
        self.install_hook()
173
 
        b = Branch.open(branch_url)
174
 
        if isinstance(b, RemoteBranch):
 
175
        b = _mod_branch.Branch.open(branch_url)
 
176
        if isinstance(b, remote.RemoteBranch):
175
177
            self.assertEqual(3, len(self.hook_calls))
176
178
            # open_branchV2 RPC
177
179
            self.assertRealBranch(self.hook_calls[0])
185
187
    def assertRealBranch(self, b):
186
188
        # Branches opened on the server don't have comparable URLs, so we just
187
189
        # assert that it is not a RemoteBranch.
188
 
        self.assertIsInstance(b, Branch)
189
 
        self.assertFalse(isinstance(b, RemoteBranch))
 
190
        self.assertIsInstance(b, _mod_branch.Branch)
 
191
        self.assertFalse(isinstance(b, remote.RemoteBranch))
190
192
 
191
193
 
192
194
class TestPreChangeBranchTip(ChangeBranchTipTestCase):
203
205
        def assertBranchAtRevision1(params):
204
206
            self.assertEquals(
205
207
                (1, 'revid-one'), params.branch.last_revision_info())
206
 
        Branch.hooks.install_named_hook(
 
208
        _mod_branch.Branch.hooks.install_named_hook(
207
209
            'pre_change_branch_tip', assertBranchAtRevision1, None)
208
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
210
        branch.set_last_revision_info(0, revision.NULL_REVISION)
209
211
 
210
212
    def test_hook_failure_prevents_change(self):
211
213
        """If a hook raises an exception, the change does not take effect."""
215
217
            pass
216
218
        def hook_that_raises(params):
217
219
            raise PearShapedError()
218
 
        Branch.hooks.install_named_hook(
 
220
        _mod_branch.Branch.hooks.install_named_hook(
219
221
            'pre_change_branch_tip', hook_that_raises, None)
220
222
        hook_failed_exc = self.assertRaises(
221
 
            PearShapedError, branch.set_last_revision_info, 0, NULL_REVISION)
 
223
            PearShapedError,
 
224
            branch.set_last_revision_info, 0, revision.NULL_REVISION)
222
225
        # The revision info is unchanged.
223
226
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
224
227
 
225
228
    def test_empty_history(self):
226
229
        branch = self.make_branch('source')
227
230
        hook_calls = self.install_logging_hook('pre')
228
 
        branch.set_last_revision_info(0, NULL_REVISION)
229
 
        expected_params = ChangeBranchTipParams(
230
 
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
 
231
        branch.set_last_revision_info(0, revision.NULL_REVISION)
 
232
        expected_params = _mod_branch.ChangeBranchTipParams(
 
233
            branch, 0, 0, revision.NULL_REVISION, revision.NULL_REVISION)
231
234
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
232
235
 
233
236
    def test_nonempty_history(self):
238
241
            'one-\xc2\xb5', 'two-\xc2\xb5')
239
242
        hook_calls = self.install_logging_hook('pre')
240
243
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
241
 
        expected_params = ChangeBranchTipParams(
 
244
        expected_params = _mod_branch.ChangeBranchTipParams(
242
245
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
243
246
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
244
247
 
246
249
        branch = self.make_branch('source')
247
250
        def assertBranchIsLocked(params):
248
251
            self.assertTrue(params.branch.is_locked())
249
 
        Branch.hooks.install_named_hook(
 
252
        _mod_branch.Branch.hooks.install_named_hook(
250
253
            'pre_change_branch_tip', assertBranchIsLocked, None)
251
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
254
        branch.set_last_revision_info(0, revision.NULL_REVISION)
252
255
 
253
256
    def test_calls_all_hooks_no_errors(self):
254
257
        """If multiple hooks are registered, all are called (if none raise
258
261
        hook_calls_1 = self.install_logging_hook('pre')
259
262
        hook_calls_2 = self.install_logging_hook('pre')
260
263
        self.assertIsNot(hook_calls_1, hook_calls_2)
261
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
264
        branch.set_last_revision_info(0, revision.NULL_REVISION)
262
265
        # Both hooks are called.
263
 
        if isinstance(branch, RemoteBranch):
 
266
        if isinstance(branch, remote.RemoteBranch):
264
267
            count = 2
265
268
        else:
266
269
            count = 1
275
278
        branch = self.make_branch_with_revision_ids(
276
279
            'one-\xc2\xb5', 'two-\xc2\xb5')
277
280
        def hook_that_rejects(params):
278
 
            raise TipChangeRejected('rejection message')
279
 
        Branch.hooks.install_named_hook(
 
281
            raise errors.TipChangeRejected('rejection message')
 
282
        _mod_branch.Branch.hooks.install_named_hook(
280
283
            'pre_change_branch_tip', hook_that_rejects, None)
281
284
        self.assertRaises(
282
 
            TipChangeRejected, branch.set_last_revision_info, 0, NULL_REVISION)
 
285
            errors.TipChangeRejected,
 
286
            branch.set_last_revision_info, 0, revision.NULL_REVISION)
283
287
        # The revision info is unchanged.
284
288
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
285
289
 
297
301
        branch = self.make_branch_with_revision_ids('revid-one')
298
302
        def assertBranchAtRevision1(params):
299
303
            self.assertEquals(
300
 
                (0, NULL_REVISION), params.branch.last_revision_info())
301
 
        Branch.hooks.install_named_hook(
 
304
                (0, revision.NULL_REVISION), params.branch.last_revision_info())
 
305
        _mod_branch.Branch.hooks.install_named_hook(
302
306
            'post_change_branch_tip', assertBranchAtRevision1, None)
303
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
307
        branch.set_last_revision_info(0, revision.NULL_REVISION)
304
308
 
305
309
    def test_empty_history(self):
306
310
        branch = self.make_branch('source')
307
311
        hook_calls = self.install_logging_hook('post')
308
 
        branch.set_last_revision_info(0, NULL_REVISION)
309
 
        expected_params = ChangeBranchTipParams(
310
 
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
 
312
        branch.set_last_revision_info(0, revision.NULL_REVISION)
 
313
        expected_params = _mod_branch.ChangeBranchTipParams(
 
314
            branch, 0, 0, revision.NULL_REVISION, revision.NULL_REVISION)
311
315
        self.assertHookCalls(expected_params, branch, hook_calls)
312
316
 
313
317
    def test_nonempty_history(self):
318
322
            'one-\xc2\xb5', 'two-\xc2\xb5')
319
323
        hook_calls = self.install_logging_hook('post')
320
324
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
321
 
        expected_params = ChangeBranchTipParams(
 
325
        expected_params = _mod_branch.ChangeBranchTipParams(
322
326
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
323
327
        self.assertHookCalls(expected_params, branch, hook_calls)
324
328
 
327
331
        branch = self.make_branch('source')
328
332
        def assertBranchIsLocked(params):
329
333
            self.assertTrue(params.branch.is_locked())
330
 
        Branch.hooks.install_named_hook(
 
334
        _mod_branch.Branch.hooks.install_named_hook(
331
335
            'post_change_branch_tip', assertBranchIsLocked, None)
332
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
336
        branch.set_last_revision_info(0, revision.NULL_REVISION)
333
337
 
334
338
    def test_calls_all_hooks_no_errors(self):
335
339
        """If multiple hooks are registered, all are called (if none raise
339
343
        hook_calls_1 = self.install_logging_hook('post')
340
344
        hook_calls_2 = self.install_logging_hook('post')
341
345
        self.assertIsNot(hook_calls_1, hook_calls_2)
342
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
346
        branch.set_last_revision_info(0, revision.NULL_REVISION)
343
347
        # Both hooks are called.
344
 
        if isinstance(branch, RemoteBranch):
 
348
        if isinstance(branch, remote.RemoteBranch):
345
349
            count = 2
346
350
        else:
347
351
            count = 1
373
377
        """
374
378
        # Check for the number of invocations expected. One invocation is
375
379
        # local, one is remote (if the branch is remote).
376
 
        if smart_enabled and isinstance(branch, RemoteBranch):
 
380
        if smart_enabled and isinstance(branch, remote.RemoteBranch):
377
381
            length = 2
378
382
        else:
379
383
            length = 1
387
391
 
388
392
    def test_set_last_revision_info(self):
389
393
        branch = self.make_branch('')
390
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
394
        branch.set_last_revision_info(0, revision.NULL_REVISION)
391
395
        self.assertPreAndPostHooksWereInvoked(branch, True)
392
396
 
393
397
    def test_generate_revision_history(self):
394
398
        branch = self.make_branch('')
395
 
        branch.generate_revision_history(NULL_REVISION)
 
399
        branch.generate_revision_history(revision.NULL_REVISION)
396
400
        # NB: for HPSS protocols < v3, the server does not invoke branch tip
397
401
        # change events on generate_revision_history, as the change is done
398
402
        # directly by the client over the VFS.