← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Robert Collins
  • Date: 2009-02-13 00:52:18 UTC
  • mto: This revision was merged to the branch mainline in revision 4012.
  • Revision ID: robertc@robertcollins.net-20090213005218-yxrgiq7j1du2vldm
Fix RemoteBranch to be used correctly in tests using bzr+ssh, to fire off Branch hooks correctly, and improve the branch_implementations tests to check that making a branch gets the right format under test.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/branch_implementations/test_hooks.py
23
23
from bzrlib.tests import TestCaseWithMemoryTransport
24
24
 
25
25
 
26
 
class TestSetRevisionHistoryHook(TestCaseWithMemoryTransport):
 
26
class ChangeBranchTipTestCase(TestCaseWithMemoryTransport):
 
27
    """Base TestCase for testing pre/post_change_branch_tip hooks."""
 
28
 
 
29
    def install_logging_hook(self, prefix):
 
30
        """Add a hook that logs calls made to it.
 
31
 
 
32
        :returns: the list that the calls will be appended to.
 
33
        """
 
34
        hook_calls = []
 
35
        Branch.hooks.install_named_hook(
 
36
            prefix + '_change_branch_tip', hook_calls.append, None)
 
37
        return hook_calls
 
38
 
 
39
    def make_branch_with_revision_ids(self, *revision_ids):
 
40
        """Makes a branch with the given commits."""
 
41
        tree = self.make_branch_and_memory_tree('source')
 
42
        tree.lock_write()
 
43
        tree.add('')
 
44
        for revision_id in revision_ids:
 
45
            tree.commit(u'Message of ' + revision_id.decode('utf8'),
 
46
                        rev_id=revision_id)
 
47
        tree.unlock()
 
48
        branch = tree.branch
 
49
        return branch
 
50
 
 
51
    def assertHookCalls(self, expected_params, branch, hook_calls=None,
 
52
        pre=False):
 
53
        if hook_calls is None:
 
54
            hook_calls = self.hook_calls
 
55
        if isinstance(branch, RemoteBranch):
 
56
            # For a remote branch, both the server and the client will raise
 
57
            # this hook, and we see both in the test environment. The remote
 
58
            # instance comes in between the clients - the client doe pre, the
 
59
            # server does pre, the server does post, the client does post.
 
60
            if pre:
 
61
                offset = 0
 
62
            else:
 
63
                offset = 1
 
64
            self.assertEqual(expected_params, hook_calls[offset])
 
65
            self.assertEqual(2, len(hook_calls))
 
66
        else:
 
67
            self.assertEqual([expected_params], hook_calls)
 
68
 
 
69
 
 
70
class TestSetRevisionHistoryHook(ChangeBranchTipTestCase):
27
71
 
28
72
    def setUp(self):
29
73
        self.hook_calls = []
42
86
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
43
87
                                        None)
44
88
        branch.set_revision_history([])
45
 
        self.assertEqual(self.hook_calls,
46
 
            [('set_rh', branch, [], True)])
 
89
        expected_params = ('set_rh', branch, [], True)
 
90
        self.assertHookCalls(expected_params, branch)
47
91
 
48
92
    def test_set_rh_nonempty_history(self):
49
93
        tree = self.make_branch_and_memory_tree('source')
58
102
        # some branches require that their history be set to a revision in the
59
103
        # repository
60
104
        branch.set_revision_history(['f\xc2\xb5'])
61
 
        self.assertEqual(self.hook_calls,
62
 
            [('set_rh', branch, ['f\xc2\xb5'], True)])
 
105
        expected_params =('set_rh', branch, ['f\xc2\xb5'], True)
 
106
        self.assertHookCalls(expected_params, branch)
63
107
 
64
108
    def test_set_rh_branch_is_locked(self):
65
109
        branch = self.make_branch('source')
66
110
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
67
111
                                        None)
68
112
        branch.set_revision_history([])
69
 
        self.assertEqual(self.hook_calls,
70
 
            [('set_rh', branch, [], True)])
 
113
        expected_params = ('set_rh', branch, [], True)
 
114
        self.assertHookCalls(expected_params, branch)
71
115
 
72
116
    def test_set_rh_calls_all_hooks_no_errors(self):
73
117
        branch = self.make_branch('source')
76
120
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
77
121
                                        None)
78
122
        branch.set_revision_history([])
79
 
        self.assertEqual(self.hook_calls,
80
 
            [('set_rh', branch, [], True),
81
 
             ('set_rh', branch, [], True),
82
 
            ])
83
 
 
84
 
 
85
 
class ChangeBranchTipTestCase(TestCaseWithMemoryTransport):
86
 
    """Base TestCase for testing pre/post_change_branch_tip hooks."""
87
 
 
88
 
    def install_logging_hook(self, prefix):
89
 
        """Add a hook that logs calls made to it.
90
 
        
91
 
        :returns: the list that the calls will be appended to.
92
 
        """
93
 
        hook_calls = []
94
 
        Branch.hooks.install_named_hook(
95
 
            prefix + '_change_branch_tip', hook_calls.append, None)
96
 
        return hook_calls
97
 
 
98
 
    def make_branch_with_revision_ids(self, *revision_ids):
99
 
        """Makes a branch with the given commits."""
100
 
        tree = self.make_branch_and_memory_tree('source')
101
 
        tree.lock_write()
102
 
        tree.add('')
103
 
        for revision_id in revision_ids:
104
 
            tree.commit(u'Message of ' + revision_id.decode('utf8'),
105
 
                        rev_id=revision_id)
106
 
        tree.unlock()
107
 
        branch = tree.branch
108
 
        return branch
 
123
        expected_calls = [('set_rh', branch, [], True),
 
124
            ('set_rh', branch, [], True),
 
125
            ]
 
126
        if isinstance(branch, RemoteBranch):
 
127
            # For a remote branch, both the server and the client will raise
 
128
            # set_rh, and the server will do so first because that is where
 
129
            # the change takes place.
 
130
            self.assertEqual(expected_calls, self.hook_calls[2:])
 
131
            self.assertEqual(4, len(self.hook_calls))
 
132
        else:
 
133
            self.assertEqual(expected_calls, self.hook_calls)
109
134
 
110
135
 
111
136
class TestOpen(TestCaseWithMemoryTransport):
120
145
    def test_create(self):
121
146
        self.install_hook()
122
147
        b = self.make_branch('.')
123
 
        self.assertEqual([b], self.hook_calls)
 
148
        if isinstance(b, RemoteBranch):
 
149
            # RemoteBranch creation:
 
150
            # - creates the branch via the VFS
 
151
            # - does a branch open (by creating a RemoteBranch object)
 
152
            # - this has the same behaviour as simple branch opening, with an
 
153
            # additional VFS open at the front.
 
154
            self.assertEqual(b._real_branch, self.hook_calls[0])
 
155
            self.assertOpenedRemoteBranch(self.hook_calls[1:], b)
 
156
        else:
 
157
            self.assertEqual([b], self.hook_calls)
124
158
 
125
159
    def test_open(self):
126
160
        branch_url = self.make_branch('.').bzrdir.root_transport.base
127
161
        self.install_hook()
128
162
        b = Branch.open(branch_url)
129
163
        if isinstance(b, RemoteBranch):
130
 
            # RemoteBranch open always opens the backing branch to get stacking
131
 
            # details. As that is done remotely we can't see the branch object
132
 
            # nor even compare base url's etc. So we just assert that the first
133
 
            # branch returned is the RemoteBranch, and that the second is a
134
 
            # Branch but not a RemoteBranch.
135
 
            self.assertEqual(2, len(self.hook_calls))
136
 
            self.assertEqual(b, self.hook_calls[0])
137
 
            self.assertIsInstance(self.hook_calls[1], Branch)
138
 
            self.assertFalse(isinstance(self.hook_calls[1], RemoteBranch))
 
164
            self.assertOpenedRemoteBranch(self.hook_calls, b)
139
165
        else:
140
166
            self.assertEqual([b], self.hook_calls)
141
167
 
 
168
    def assertOpenedRemoteBranch(self, hook_calls, b):
 
169
        """Assert that the expected calls were recorded for opening 'b'."""
 
170
        # RemoteBranch open always opens the backing branch to get stacking
 
171
        # details. As that is done remotely we can't see the branch object
 
172
        # nor even compare base url's etc. So we just assert that the first
 
173
        # branch returned is the RemoteBranch, and that the second is a
 
174
        # Branch but not a RemoteBranch.
 
175
        self.assertEqual(2, len(hook_calls))
 
176
        self.assertEqual(b, hook_calls[0])
 
177
        self.assertIsInstance(hook_calls[1], Branch)
 
178
        self.assertFalse(isinstance(hook_calls[1], RemoteBranch))
 
179
 
142
180
 
143
181
class TestPreChangeBranchTip(ChangeBranchTipTestCase):
144
182
    """Tests for pre_change_branch_tip hook.
183
221
        branch.set_last_revision_info(0, NULL_REVISION)
184
222
        expected_params = ChangeBranchTipParams(
185
223
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
186
 
        self.assertEqual([expected_params], hook_calls)
 
224
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
187
225
 
188
226
    def test_nonempty_history(self):
189
227
        # some branches require that their history be set to a revision in the
195
233
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
196
234
        expected_params = ChangeBranchTipParams(
197
235
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
198
 
        self.assertEqual([expected_params], hook_calls)
 
236
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
199
237
 
200
238
    def test_branch_is_locked(self):
201
239
        branch = self.make_branch('source')
215
253
        self.assertIsNot(hook_calls_1, hook_calls_2)
216
254
        branch.set_last_revision_info(0, NULL_REVISION)
217
255
        # Both hooks are called.
218
 
        self.assertEqual(len(hook_calls_1), 1)
219
 
        self.assertEqual(len(hook_calls_2), 1)
 
256
        if isinstance(branch, RemoteBranch):
 
257
            count = 2
 
258
        else:
 
259
            count = 1
 
260
        self.assertEqual(len(hook_calls_1), count)
 
261
        self.assertEqual(len(hook_calls_2), count)
220
262
 
221
263
    def test_explicit_reject_by_hook(self):
222
264
        """If a hook raises TipChangeRejected, the change does not take effect.
259
301
        branch.set_last_revision_info(0, NULL_REVISION)
260
302
        expected_params = ChangeBranchTipParams(
261
303
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
262
 
        self.assertEqual([expected_params], hook_calls)
 
304
        self.assertHookCalls(expected_params, branch, hook_calls)
263
305
 
264
306
    def test_nonempty_history(self):
265
307
        # some branches require that their history be set to a revision in the
271
313
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
272
314
        expected_params = ChangeBranchTipParams(
273
315
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
274
 
        self.assertEqual([expected_params], hook_calls)
 
316
        self.assertHookCalls(expected_params, branch, hook_calls)
275
317
 
276
318
    def test_branch_is_locked(self):
277
319
        """The branch passed to the hook is locked."""
292
334
        self.assertIsNot(hook_calls_1, hook_calls_2)
293
335
        branch.set_last_revision_info(0, NULL_REVISION)
294
336
        # Both hooks are called.
295
 
        self.assertEqual(len(hook_calls_1), 1)
296
 
        self.assertEqual(len(hook_calls_2), 1)
 
337
        if isinstance(branch, RemoteBranch):
 
338
            count = 2
 
339
        else:
 
340
            count = 1
 
341
        self.assertEqual(len(hook_calls_1), count)
 
342
        self.assertEqual(len(hook_calls_2), count)
297
343
 
298
344
 
299
345
class TestAllMethodsThatChangeTipWillRunHooks(ChangeBranchTipTestCase):
312
358
    def resetHookCalls(self):
313
359
        del self.pre_hook_calls[:], self.post_hook_calls[:]
314
360
 
315
 
    def assertPreAndPostHooksWereInvoked(self):
316
 
        # Check for len == 1, because the hooks should only be be invoked once
317
 
        # by an operation.
318
 
        self.assertEqual(1, len(self.pre_hook_calls))
319
 
        self.assertEqual(1, len(self.post_hook_calls))
 
361
    def assertPreAndPostHooksWereInvoked(self, branch, smart_enabled):
 
362
        """assert that both pre and post hooks were called
 
363
 
 
364
        :param smart_enabled: The method invoked is one that should be
 
365
            smart server ready.
 
366
        """
 
367
        # Check for the number of invocations expected. One invocation is
 
368
        # local, one is remote (if the branch is remote).
 
369
        if smart_enabled and isinstance(branch, RemoteBranch):
 
370
            length = 2
 
371
        else:
 
372
            length = 1
 
373
        self.assertEqual(length, len(self.pre_hook_calls))
 
374
        self.assertEqual(length, len(self.post_hook_calls))
320
375
 
321
376
    def test_set_revision_history(self):
322
377
        branch = self.make_branch('')
323
378
        branch.set_revision_history([])
324
 
        self.assertPreAndPostHooksWereInvoked()
 
379
        self.assertPreAndPostHooksWereInvoked(branch, True)
325
380
 
326
381
    def test_set_last_revision_info(self):
327
382
        branch = self.make_branch('')
328
383
        branch.set_last_revision_info(0, NULL_REVISION)
329
 
        self.assertPreAndPostHooksWereInvoked()
 
384
        self.assertPreAndPostHooksWereInvoked(branch, True)
330
385
 
331
386
    def test_generate_revision_history(self):
332
387
        branch = self.make_branch('')
333
388
        branch.generate_revision_history(NULL_REVISION)
334
 
        self.assertPreAndPostHooksWereInvoked()
 
389
        # NB: for HPSS protocols < v3, the server does not invoke branch tip
 
390
        # change events on generate_revision_history, as the change is done
 
391
        # directly by the client over the VFS.
 
392
        self.assertPreAndPostHooksWereInvoked(branch, True)
335
393
 
336
394
    def test_pull(self):
337
395
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
338
396
        self.resetHookCalls()
339
397
        destination_branch = self.make_branch('destination')
340
398
        destination_branch.pull(source_branch)
341
 
        self.assertPreAndPostHooksWereInvoked()
 
399
        self.assertPreAndPostHooksWereInvoked(destination_branch, False)
342
400
 
343
401
    def test_push(self):
344
402
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
345
403
        self.resetHookCalls()
346
404
        destination_branch = self.make_branch('destination')
347
405
        source_branch.push(destination_branch)
348
 
        self.assertPreAndPostHooksWereInvoked()
349
 
 
350
 
        
 
406
        self.assertPreAndPostHooksWereInvoked(destination_branch, True)