← 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: Andrew Bennetts
  • Date: 2010-02-12 04:33:05 UTC
  • mfrom: (5031 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5032.
  • Revision ID: andrew.bennetts@canonical.com-20100212043305-ujdbsdoviql2t7i3
Merge lp:bzr

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_branch/test_hooks.py
1
 
# Copyright (C) 2007 Canonical Ltd
 
1
# Copyright (C) 2007-2010 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
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
 
        """If a hook raises an exception, the change does not take effect.
212
 
 
213
 
        Also, a HookFailed exception will be raised.
214
 
        """
 
213
        """If a hook raises an exception, the change does not take effect."""
215
214
        branch = self.make_branch_with_revision_ids(
216
215
            'one-\xc2\xb5', 'two-\xc2\xb5')
217
216
        class PearShapedError(Exception):
218
217
            pass
219
218
        def hook_that_raises(params):
220
219
            raise PearShapedError()
221
 
        Branch.hooks.install_named_hook(
 
220
        _mod_branch.Branch.hooks.install_named_hook(
222
221
            'pre_change_branch_tip', hook_that_raises, None)
223
222
        hook_failed_exc = self.assertRaises(
224
 
            HookFailed, branch.set_last_revision_info, 0, NULL_REVISION)
225
 
        self.assertIsInstance(hook_failed_exc.exc_value, PearShapedError)
 
223
            PearShapedError,
 
224
            branch.set_last_revision_info, 0, revision.NULL_REVISION)
226
225
        # The revision info is unchanged.
227
226
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
228
227
 
229
228
    def test_empty_history(self):
230
229
        branch = self.make_branch('source')
231
230
        hook_calls = self.install_logging_hook('pre')
232
 
        branch.set_last_revision_info(0, NULL_REVISION)
233
 
        expected_params = ChangeBranchTipParams(
234
 
            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)
235
234
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
236
235
 
237
236
    def test_nonempty_history(self):
242
241
            'one-\xc2\xb5', 'two-\xc2\xb5')
243
242
        hook_calls = self.install_logging_hook('pre')
244
243
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
245
 
        expected_params = ChangeBranchTipParams(
 
244
        expected_params = _mod_branch.ChangeBranchTipParams(
246
245
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
247
246
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
248
247
 
250
249
        branch = self.make_branch('source')
251
250
        def assertBranchIsLocked(params):
252
251
            self.assertTrue(params.branch.is_locked())
253
 
        Branch.hooks.install_named_hook(
 
252
        _mod_branch.Branch.hooks.install_named_hook(
254
253
            'pre_change_branch_tip', assertBranchIsLocked, None)
255
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
254
        branch.set_last_revision_info(0, revision.NULL_REVISION)
256
255
 
257
256
    def test_calls_all_hooks_no_errors(self):
258
257
        """If multiple hooks are registered, all are called (if none raise
262
261
        hook_calls_1 = self.install_logging_hook('pre')
263
262
        hook_calls_2 = self.install_logging_hook('pre')
264
263
        self.assertIsNot(hook_calls_1, hook_calls_2)
265
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
264
        branch.set_last_revision_info(0, revision.NULL_REVISION)
266
265
        # Both hooks are called.
267
 
        if isinstance(branch, RemoteBranch):
 
266
        if isinstance(branch, remote.RemoteBranch):
268
267
            count = 2
269
268
        else:
270
269
            count = 1
279
278
        branch = self.make_branch_with_revision_ids(
280
279
            'one-\xc2\xb5', 'two-\xc2\xb5')
281
280
        def hook_that_rejects(params):
282
 
            raise TipChangeRejected('rejection message')
283
 
        Branch.hooks.install_named_hook(
 
281
            raise errors.TipChangeRejected('rejection message')
 
282
        _mod_branch.Branch.hooks.install_named_hook(
284
283
            'pre_change_branch_tip', hook_that_rejects, None)
285
284
        self.assertRaises(
286
 
            TipChangeRejected, branch.set_last_revision_info, 0, NULL_REVISION)
 
285
            errors.TipChangeRejected,
 
286
            branch.set_last_revision_info, 0, revision.NULL_REVISION)
287
287
        # The revision info is unchanged.
288
288
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
289
289
 
301
301
        branch = self.make_branch_with_revision_ids('revid-one')
302
302
        def assertBranchAtRevision1(params):
303
303
            self.assertEquals(
304
 
                (0, NULL_REVISION), params.branch.last_revision_info())
305
 
        Branch.hooks.install_named_hook(
 
304
                (0, revision.NULL_REVISION), params.branch.last_revision_info())
 
305
        _mod_branch.Branch.hooks.install_named_hook(
306
306
            'post_change_branch_tip', assertBranchAtRevision1, None)
307
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
307
        branch.set_last_revision_info(0, revision.NULL_REVISION)
308
308
 
309
309
    def test_empty_history(self):
310
310
        branch = self.make_branch('source')
311
311
        hook_calls = self.install_logging_hook('post')
312
 
        branch.set_last_revision_info(0, NULL_REVISION)
313
 
        expected_params = ChangeBranchTipParams(
314
 
            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)
315
315
        self.assertHookCalls(expected_params, branch, hook_calls)
316
316
 
317
317
    def test_nonempty_history(self):
322
322
            'one-\xc2\xb5', 'two-\xc2\xb5')
323
323
        hook_calls = self.install_logging_hook('post')
324
324
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
325
 
        expected_params = ChangeBranchTipParams(
 
325
        expected_params = _mod_branch.ChangeBranchTipParams(
326
326
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
327
327
        self.assertHookCalls(expected_params, branch, hook_calls)
328
328
 
331
331
        branch = self.make_branch('source')
332
332
        def assertBranchIsLocked(params):
333
333
            self.assertTrue(params.branch.is_locked())
334
 
        Branch.hooks.install_named_hook(
 
334
        _mod_branch.Branch.hooks.install_named_hook(
335
335
            'post_change_branch_tip', assertBranchIsLocked, None)
336
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
336
        branch.set_last_revision_info(0, revision.NULL_REVISION)
337
337
 
338
338
    def test_calls_all_hooks_no_errors(self):
339
339
        """If multiple hooks are registered, all are called (if none raise
343
343
        hook_calls_1 = self.install_logging_hook('post')
344
344
        hook_calls_2 = self.install_logging_hook('post')
345
345
        self.assertIsNot(hook_calls_1, hook_calls_2)
346
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
346
        branch.set_last_revision_info(0, revision.NULL_REVISION)
347
347
        # Both hooks are called.
348
 
        if isinstance(branch, RemoteBranch):
 
348
        if isinstance(branch, remote.RemoteBranch):
349
349
            count = 2
350
350
        else:
351
351
            count = 1
377
377
        """
378
378
        # Check for the number of invocations expected. One invocation is
379
379
        # local, one is remote (if the branch is remote).
380
 
        if smart_enabled and isinstance(branch, RemoteBranch):
 
380
        if smart_enabled and isinstance(branch, remote.RemoteBranch):
381
381
            length = 2
382
382
        else:
383
383
            length = 1
391
391
 
392
392
    def test_set_last_revision_info(self):
393
393
        branch = self.make_branch('')
394
 
        branch.set_last_revision_info(0, NULL_REVISION)
 
394
        branch.set_last_revision_info(0, revision.NULL_REVISION)
395
395
        self.assertPreAndPostHooksWereInvoked(branch, True)
396
396
 
397
397
    def test_generate_revision_history(self):
398
398
        branch = self.make_branch('')
399
 
        branch.generate_revision_history(NULL_REVISION)
 
399
        branch.generate_revision_history(revision.NULL_REVISION)
400
400
        # NB: for HPSS protocols < v3, the server does not invoke branch tip
401
401
        # change events on generate_revision_history, as the change is done
402
402
        # directly by the client over the VFS.