~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Aaron Bentley
  • Date: 2008-10-16 21:27:10 UTC
  • mfrom: (0.15.26 unshelve)
  • mto: (0.16.74 shelf-ui)
  • mto: This revision was merged to the branch mainline in revision 3820.
  • Revision ID: aaron@aaronbentley.com-20081016212710-h9av3nhk76dvmsv5
Merge with unshelve

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2007 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Tests that branch classes implement hook callouts correctly."""
 
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.tests import TestCaseWithMemoryTransport
 
24
 
 
25
 
 
26
class TestSetRevisionHistoryHook(TestCaseWithMemoryTransport):
 
27
 
 
28
    def setUp(self):
 
29
        self.hook_calls = []
 
30
        TestCaseWithMemoryTransport.setUp(self)
 
31
 
 
32
    def capture_set_rh_hook(self, branch, rev_history):
 
33
        """Capture post set-rh hook calls to self.hook_calls.
 
34
        
 
35
        The call is logged, as is some state of the branch.
 
36
        """
 
37
        self.hook_calls.append(
 
38
            ('set_rh', branch, rev_history, branch.is_locked()))
 
39
 
 
40
    def test_set_rh_empty_history(self):
 
41
        branch = self.make_branch('source')
 
42
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
43
                                        None)
 
44
        branch.set_revision_history([])
 
45
        self.assertEqual(self.hook_calls,
 
46
            [('set_rh', branch, [], True)])
 
47
 
 
48
    def test_set_rh_nonempty_history(self):
 
49
        tree = self.make_branch_and_memory_tree('source')
 
50
        tree.lock_write()
 
51
        tree.add('')
 
52
        tree.commit('another commit', rev_id='f\xc2\xb5')
 
53
        tree.commit('empty commit', rev_id='foo')
 
54
        tree.unlock()
 
55
        branch = tree.branch
 
56
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
57
                                        None)
 
58
        # some branches require that their history be set to a revision in the
 
59
        # repository
 
60
        branch.set_revision_history(['f\xc2\xb5'])
 
61
        self.assertEqual(self.hook_calls,
 
62
            [('set_rh', branch, ['f\xc2\xb5'], True)])
 
63
 
 
64
    def test_set_rh_branch_is_locked(self):
 
65
        branch = self.make_branch('source')
 
66
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
67
                                        None)
 
68
        branch.set_revision_history([])
 
69
        self.assertEqual(self.hook_calls,
 
70
            [('set_rh', branch, [], True)])
 
71
 
 
72
    def test_set_rh_calls_all_hooks_no_errors(self):
 
73
        branch = self.make_branch('source')
 
74
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
75
                                        None)
 
76
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
77
                                        None)
 
78
        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
 
109
 
 
110
 
 
111
class TestOpen(TestCaseWithMemoryTransport):
 
112
 
 
113
    def capture_hook(self, branch):
 
114
        self.hook_calls.append(branch)
 
115
 
 
116
    def install_hook(self):
 
117
        self.hook_calls = []
 
118
        Branch.hooks.install_named_hook('open', self.capture_hook, None)
 
119
 
 
120
    def test_create(self):
 
121
        self.install_hook()
 
122
        b = self.make_branch('.')
 
123
        self.assertEqual([b], self.hook_calls)
 
124
 
 
125
    def test_open(self):
 
126
        branch_url = self.make_branch('.').bzrdir.root_transport.base
 
127
        self.install_hook()
 
128
        b = Branch.open(branch_url)
 
129
        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))
 
139
        else:
 
140
            self.assertEqual([b], self.hook_calls)
 
141
 
 
142
 
 
143
class TestPreChangeBranchTip(ChangeBranchTipTestCase):
 
144
    """Tests for pre_change_branch_tip hook.
 
145
    
 
146
    Most of these tests are very similar to the tests in
 
147
    TestPostChangeBranchTip.
 
148
    """
 
149
 
 
150
    def test_hook_runs_before_change(self):
 
151
        """The hook runs *before* the branch's last_revision_info has changed.
 
152
        """
 
153
        branch = self.make_branch_with_revision_ids('revid-one')
 
154
        def assertBranchAtRevision1(params):
 
155
            self.assertEquals(
 
156
                (1, 'revid-one'), params.branch.last_revision_info())
 
157
        Branch.hooks.install_named_hook(
 
158
            'pre_change_branch_tip', assertBranchAtRevision1, None)
 
159
        branch.set_last_revision_info(0, NULL_REVISION)
 
160
 
 
161
    def test_hook_failure_prevents_change(self):
 
162
        """If a hook raises an exception, the change does not take effect.
 
163
        
 
164
        Also, a HookFailed exception will be raised.
 
165
        """
 
166
        branch = self.make_branch_with_revision_ids(
 
167
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
168
        class PearShapedError(Exception):
 
169
            pass
 
170
        def hook_that_raises(params):
 
171
            raise PearShapedError()
 
172
        Branch.hooks.install_named_hook(
 
173
            'pre_change_branch_tip', hook_that_raises, None)
 
174
        hook_failed_exc = self.assertRaises(
 
175
            HookFailed, branch.set_last_revision_info, 0, NULL_REVISION)
 
176
        self.assertIsInstance(hook_failed_exc.exc_value, PearShapedError)
 
177
        # The revision info is unchanged.
 
178
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
 
179
        
 
180
    def test_empty_history(self):
 
181
        branch = self.make_branch('source')
 
182
        hook_calls = self.install_logging_hook('pre')
 
183
        branch.set_last_revision_info(0, NULL_REVISION)
 
184
        expected_params = ChangeBranchTipParams(
 
185
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
 
186
        self.assertEqual([expected_params], hook_calls)
 
187
 
 
188
    def test_nonempty_history(self):
 
189
        # some branches require that their history be set to a revision in the
 
190
        # repository, so we need to make a branch with non-empty history for
 
191
        # this test.
 
192
        branch = self.make_branch_with_revision_ids(
 
193
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
194
        hook_calls = self.install_logging_hook('pre')
 
195
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
 
196
        expected_params = ChangeBranchTipParams(
 
197
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
 
198
        self.assertEqual([expected_params], hook_calls)
 
199
 
 
200
    def test_branch_is_locked(self):
 
201
        branch = self.make_branch('source')
 
202
        def assertBranchIsLocked(params):
 
203
            self.assertTrue(params.branch.is_locked())
 
204
        Branch.hooks.install_named_hook(
 
205
            'pre_change_branch_tip', assertBranchIsLocked, None)
 
206
        branch.set_last_revision_info(0, NULL_REVISION)
 
207
 
 
208
    def test_calls_all_hooks_no_errors(self):
 
209
        """If multiple hooks are registered, all are called (if none raise
 
210
        errors).
 
211
        """
 
212
        branch = self.make_branch('source')
 
213
        hook_calls_1 = self.install_logging_hook('pre')
 
214
        hook_calls_2 = self.install_logging_hook('pre')
 
215
        self.assertIsNot(hook_calls_1, hook_calls_2)
 
216
        branch.set_last_revision_info(0, NULL_REVISION)
 
217
        # Both hooks are called.
 
218
        self.assertEqual(len(hook_calls_1), 1)
 
219
        self.assertEqual(len(hook_calls_2), 1)
 
220
 
 
221
    def test_explicit_reject_by_hook(self):
 
222
        """If a hook raises TipChangeRejected, the change does not take effect.
 
223
        
 
224
        TipChangeRejected exceptions are propagated, not wrapped in HookFailed.
 
225
        """
 
226
        branch = self.make_branch_with_revision_ids(
 
227
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
228
        def hook_that_rejects(params):
 
229
            raise TipChangeRejected('rejection message')
 
230
        Branch.hooks.install_named_hook(
 
231
            'pre_change_branch_tip', hook_that_rejects, None)
 
232
        self.assertRaises(
 
233
            TipChangeRejected, branch.set_last_revision_info, 0, NULL_REVISION)
 
234
        # The revision info is unchanged.
 
235
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
 
236
        
 
237
 
 
238
class TestPostChangeBranchTip(ChangeBranchTipTestCase):
 
239
    """Tests for post_change_branch_tip hook.
 
240
 
 
241
    Most of these tests are very similar to the tests in
 
242
    TestPostChangeBranchTip.
 
243
    """
 
244
 
 
245
    def test_hook_runs_after_change(self):
 
246
        """The hook runs *after* the branch's last_revision_info has changed.
 
247
        """
 
248
        branch = self.make_branch_with_revision_ids('revid-one')
 
249
        def assertBranchAtRevision1(params):
 
250
            self.assertEquals(
 
251
                (0, NULL_REVISION), params.branch.last_revision_info())
 
252
        Branch.hooks.install_named_hook(
 
253
            'post_change_branch_tip', assertBranchAtRevision1, None)
 
254
        branch.set_last_revision_info(0, NULL_REVISION)
 
255
 
 
256
    def test_empty_history(self):
 
257
        branch = self.make_branch('source')
 
258
        hook_calls = self.install_logging_hook('post')
 
259
        branch.set_last_revision_info(0, NULL_REVISION)
 
260
        expected_params = ChangeBranchTipParams(
 
261
            branch, 0, 0, NULL_REVISION, NULL_REVISION)
 
262
        self.assertEqual([expected_params], hook_calls)
 
263
 
 
264
    def test_nonempty_history(self):
 
265
        # some branches require that their history be set to a revision in the
 
266
        # repository, so we need to make a branch with non-empty history for
 
267
        # this test.
 
268
        branch = self.make_branch_with_revision_ids(
 
269
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
270
        hook_calls = self.install_logging_hook('post')
 
271
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
 
272
        expected_params = ChangeBranchTipParams(
 
273
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
 
274
        self.assertEqual([expected_params], hook_calls)
 
275
 
 
276
    def test_branch_is_locked(self):
 
277
        """The branch passed to the hook is locked."""
 
278
        branch = self.make_branch('source')
 
279
        def assertBranchIsLocked(params):
 
280
            self.assertTrue(params.branch.is_locked())
 
281
        Branch.hooks.install_named_hook(
 
282
            'post_change_branch_tip', assertBranchIsLocked, None)
 
283
        branch.set_last_revision_info(0, NULL_REVISION)
 
284
 
 
285
    def test_calls_all_hooks_no_errors(self):
 
286
        """If multiple hooks are registered, all are called (if none raise
 
287
        errors).
 
288
        """
 
289
        branch = self.make_branch('source')
 
290
        hook_calls_1 = self.install_logging_hook('post')
 
291
        hook_calls_2 = self.install_logging_hook('post')
 
292
        self.assertIsNot(hook_calls_1, hook_calls_2)
 
293
        branch.set_last_revision_info(0, NULL_REVISION)
 
294
        # Both hooks are called.
 
295
        self.assertEqual(len(hook_calls_1), 1)
 
296
        self.assertEqual(len(hook_calls_2), 1)
 
297
 
 
298
 
 
299
class TestAllMethodsThatChangeTipWillRunHooks(ChangeBranchTipTestCase):
 
300
    """Every method of Branch that changes a branch tip will invoke the
 
301
    pre/post_change_branch_tip hooks.
 
302
    """
 
303
 
 
304
    def setUp(self):
 
305
        ChangeBranchTipTestCase.setUp(self)
 
306
        self.installPreAndPostHooks()
 
307
        
 
308
    def installPreAndPostHooks(self):
 
309
        self.pre_hook_calls = self.install_logging_hook('pre')
 
310
        self.post_hook_calls = self.install_logging_hook('post')
 
311
 
 
312
    def resetHookCalls(self):
 
313
        del self.pre_hook_calls[:], self.post_hook_calls[:]
 
314
 
 
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))
 
320
 
 
321
    def test_set_revision_history(self):
 
322
        branch = self.make_branch('')
 
323
        branch.set_revision_history([])
 
324
        self.assertPreAndPostHooksWereInvoked()
 
325
 
 
326
    def test_set_last_revision_info(self):
 
327
        branch = self.make_branch('')
 
328
        branch.set_last_revision_info(0, NULL_REVISION)
 
329
        self.assertPreAndPostHooksWereInvoked()
 
330
 
 
331
    def test_generate_revision_history(self):
 
332
        branch = self.make_branch('')
 
333
        branch.generate_revision_history(NULL_REVISION)
 
334
        self.assertPreAndPostHooksWereInvoked()
 
335
 
 
336
    def test_pull(self):
 
337
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
 
338
        self.resetHookCalls()
 
339
        destination_branch = self.make_branch('destination')
 
340
        destination_branch.pull(source_branch)
 
341
        self.assertPreAndPostHooksWereInvoked()
 
342
 
 
343
    def test_push(self):
 
344
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
 
345
        self.resetHookCalls()
 
346
        destination_branch = self.make_branch('destination')
 
347
        source_branch.push(destination_branch)
 
348
        self.assertPreAndPostHooksWereInvoked()
 
349
 
 
350