/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/branch_implementations/test_hooks.py

merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/branch_implementations/test_hooks.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
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
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests that branch classes implement hook callouts correctly."""
18
18
 
19
 
from bzrlib.branch import Branch
 
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
20
24
from bzrlib.tests import TestCaseWithMemoryTransport
21
25
 
22
26
 
23
 
class TestSetRevisionHistoryHook(TestCaseWithMemoryTransport):
 
27
class ChangeBranchTipTestCase(TestCaseWithMemoryTransport):
 
28
    """Base TestCase for testing pre/post_change_branch_tip hooks."""
 
29
 
 
30
    def install_logging_hook(self, prefix):
 
31
        """Add a hook that logs calls made to it.
 
32
 
 
33
        :returns: the list that the calls will be appended to.
 
34
        """
 
35
        hook_calls = []
 
36
        Branch.hooks.install_named_hook(
 
37
            prefix + '_change_branch_tip', hook_calls.append, None)
 
38
        return hook_calls
 
39
 
 
40
    def make_branch_with_revision_ids(self, *revision_ids):
 
41
        """Makes a branch with the given commits."""
 
42
        tree = self.make_branch_and_memory_tree('source')
 
43
        tree.lock_write()
 
44
        tree.add('')
 
45
        for revision_id in revision_ids:
 
46
            tree.commit(u'Message of ' + revision_id.decode('utf8'),
 
47
                        rev_id=revision_id)
 
48
        tree.unlock()
 
49
        branch = tree.branch
 
50
        return branch
 
51
 
 
52
    def assertHookCalls(self, expected_params, branch, hook_calls=None,
 
53
        pre=False):
 
54
        if hook_calls is None:
 
55
            hook_calls = self.hook_calls
 
56
        if isinstance(branch, RemoteBranch):
 
57
            # For a remote branch, both the server and the client will raise
 
58
            # this hook, and we see both in the test environment. The remote
 
59
            # instance comes in between the clients - the client doe pre, the
 
60
            # server does pre, the server does post, the client does post.
 
61
            if pre:
 
62
                offset = 0
 
63
            else:
 
64
                offset = 1
 
65
            self.assertEqual(expected_params, hook_calls[offset])
 
66
            self.assertEqual(2, len(hook_calls))
 
67
        else:
 
68
            self.assertEqual([expected_params], hook_calls)
 
69
 
 
70
 
 
71
class TestSetRevisionHistoryHook(ChangeBranchTipTestCase):
24
72
 
25
73
    def setUp(self):
26
74
        self.hook_calls = []
28
76
 
29
77
    def capture_set_rh_hook(self, branch, rev_history):
30
78
        """Capture post set-rh hook calls to self.hook_calls.
31
 
        
 
79
 
32
80
        The call is logged, as is some state of the branch.
33
81
        """
34
82
        self.hook_calls.append(
36
84
 
37
85
    def test_set_rh_empty_history(self):
38
86
        branch = self.make_branch('source')
39
 
        Branch.hooks.install_hook('set_rh', self.capture_set_rh_hook)
 
87
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
88
                                        None)
40
89
        branch.set_revision_history([])
41
 
        self.assertEqual(self.hook_calls,
42
 
            [('set_rh', branch, [], True)])
 
90
        expected_params = ('set_rh', branch, [], True)
 
91
        self.assertHookCalls(expected_params, branch)
43
92
 
44
93
    def test_set_rh_nonempty_history(self):
45
94
        tree = self.make_branch_and_memory_tree('source')
46
95
        tree.lock_write()
47
96
        tree.add('')
 
97
        tree.commit('another commit', rev_id='f\xc2\xb5')
48
98
        tree.commit('empty commit', rev_id='foo')
49
99
        tree.unlock()
50
100
        branch = tree.branch
51
 
        Branch.hooks.install_hook('set_rh', self.capture_set_rh_hook)
 
101
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
102
                                        None)
 
103
        # some branches require that their history be set to a revision in the
 
104
        # repository
52
105
        branch.set_revision_history(['f\xc2\xb5'])
53
 
        self.assertEqual(self.hook_calls,
54
 
            [('set_rh', branch, ['f\xc2\xb5'], True)])
 
106
        expected_params =('set_rh', branch, ['f\xc2\xb5'], True)
 
107
        self.assertHookCalls(expected_params, branch)
55
108
 
56
109
    def test_set_rh_branch_is_locked(self):
57
110
        branch = self.make_branch('source')
58
 
        Branch.hooks.install_hook('set_rh', self.capture_set_rh_hook)
 
111
        Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
 
112
                                        None)
59
113
        branch.set_revision_history([])
60
 
        self.assertEqual(self.hook_calls,
61
 
            [('set_rh', branch, [], True)])
 
114
        expected_params = ('set_rh', branch, [], True)
 
115
        self.assertHookCalls(expected_params, branch)
62
116
 
63
117
    def test_set_rh_calls_all_hooks_no_errors(self):
64
118
        branch = self.make_branch('source')
65
 
        Branch.hooks.install_hook('set_rh', self.capture_set_rh_hook)
66
 
        Branch.hooks.install_hook('set_rh', self.capture_set_rh_hook)
67
 
        branch.set_revision_history([])
68
 
        self.assertEqual(self.hook_calls,
69
 
            [('set_rh', branch, [], True),
70
 
             ('set_rh', branch, [], True),
71
 
            ])
 
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)
 
123
        branch.set_revision_history([])
 
124
        expected_calls = [('set_rh', branch, [], True),
 
125
            ('set_rh', branch, [], True),
 
126
            ]
 
127
        if isinstance(branch, RemoteBranch):
 
128
            # For a remote branch, both the server and the client will raise
 
129
            # set_rh, and the server will do so first because that is where
 
130
            # the change takes place.
 
131
            self.assertEqual(expected_calls, self.hook_calls[2:])
 
132
            self.assertEqual(4, len(self.hook_calls))
 
133
        else:
 
134
            self.assertEqual(expected_calls, self.hook_calls)
 
135
 
 
136
 
 
137
class TestOpen(TestCaseWithMemoryTransport):
 
138
 
 
139
    def capture_hook(self, branch):
 
140
        self.hook_calls.append(branch)
 
141
 
 
142
    def install_hook(self):
 
143
        self.hook_calls = []
 
144
        Branch.hooks.install_named_hook('open', self.capture_hook, None)
 
145
 
 
146
    def test_create(self):
 
147
        self.install_hook()
 
148
        b = self.make_branch('.')
 
149
        if isinstance(b, RemoteBranch):
 
150
            # RemoteBranch creation:
 
151
            if (self.transport_readonly_server ==
 
152
                server.ReadonlySmartTCPServer_for_testing_v2_only):
 
153
                # Older servers:
 
154
                self.assertEqual(3, len(self.hook_calls))
 
155
                # creates the branch via the VFS (for older servers)
 
156
                self.assertEqual(b._real_branch, self.hook_calls[0])
 
157
                # creates a RemoteBranch object
 
158
                self.assertEqual(b, self.hook_calls[1])
 
159
                # get_stacked_on_url RPC
 
160
                self.assertRealBranch(self.hook_calls[2])
 
161
            else:
 
162
                self.assertEqual(2, len(self.hook_calls))
 
163
                # create_branch RPC
 
164
                self.assertRealBranch(self.hook_calls[0])
 
165
                # create RemoteBranch locally
 
166
                self.assertEqual(b, self.hook_calls[1])
 
167
        else:
 
168
            self.assertEqual([b], self.hook_calls)
 
169
 
 
170
    def test_open(self):
 
171
        branch_url = self.make_branch('.').bzrdir.root_transport.base
 
172
        self.install_hook()
 
173
        b = Branch.open(branch_url)
 
174
        if isinstance(b, RemoteBranch):
 
175
            self.assertEqual(3, len(self.hook_calls))
 
176
            # open_branchV2 RPC
 
177
            self.assertRealBranch(self.hook_calls[0])
 
178
            # create RemoteBranch locally
 
179
            self.assertEqual(b, self.hook_calls[1])
 
180
            # get_stacked_on_url RPC
 
181
            self.assertRealBranch(self.hook_calls[2])
 
182
        else:
 
183
            self.assertEqual([b], self.hook_calls)
 
184
 
 
185
    def assertRealBranch(self, b):
 
186
        # Branches opened on the server don't have comparable URLs, so we just
 
187
        # assert that it is not a RemoteBranch.
 
188
        self.assertIsInstance(b, Branch)
 
189
        self.assertFalse(isinstance(b, RemoteBranch))
 
190
 
 
191
 
 
192
class TestPreChangeBranchTip(ChangeBranchTipTestCase):
 
193
    """Tests for pre_change_branch_tip hook.
 
194
 
 
195
    Most of these tests are very similar to the tests in
 
196
    TestPostChangeBranchTip.
 
197
    """
 
198
 
 
199
    def test_hook_runs_before_change(self):
 
200
        """The hook runs *before* the branch's last_revision_info has changed.
 
201
        """
 
202
        branch = self.make_branch_with_revision_ids('revid-one')
 
203
        def assertBranchAtRevision1(params):
 
204
            self.assertEquals(
 
205
                (1, 'revid-one'), params.branch.last_revision_info())
 
206
        Branch.hooks.install_named_hook(
 
207
            'pre_change_branch_tip', assertBranchAtRevision1, None)
 
208
        branch.set_last_revision_info(0, NULL_REVISION)
 
209
 
 
210
    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
        """
 
215
        branch = self.make_branch_with_revision_ids(
 
216
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
217
        class PearShapedError(Exception):
 
218
            pass
 
219
        def hook_that_raises(params):
 
220
            raise PearShapedError()
 
221
        Branch.hooks.install_named_hook(
 
222
            'pre_change_branch_tip', hook_that_raises, None)
 
223
        hook_failed_exc = self.assertRaises(
 
224
            HookFailed, branch.set_last_revision_info, 0, NULL_REVISION)
 
225
        self.assertIsInstance(hook_failed_exc.exc_value, PearShapedError)
 
226
        # The revision info is unchanged.
 
227
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
 
228
 
 
229
    def test_empty_history(self):
 
230
        branch = self.make_branch('source')
 
231
        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)
 
235
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
 
236
 
 
237
    def test_nonempty_history(self):
 
238
        # some branches require that their history be set to a revision in the
 
239
        # repository, so we need to make a branch with non-empty history for
 
240
        # this test.
 
241
        branch = self.make_branch_with_revision_ids(
 
242
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
243
        hook_calls = self.install_logging_hook('pre')
 
244
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
 
245
        expected_params = ChangeBranchTipParams(
 
246
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
 
247
        self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
 
248
 
 
249
    def test_branch_is_locked(self):
 
250
        branch = self.make_branch('source')
 
251
        def assertBranchIsLocked(params):
 
252
            self.assertTrue(params.branch.is_locked())
 
253
        Branch.hooks.install_named_hook(
 
254
            'pre_change_branch_tip', assertBranchIsLocked, None)
 
255
        branch.set_last_revision_info(0, NULL_REVISION)
 
256
 
 
257
    def test_calls_all_hooks_no_errors(self):
 
258
        """If multiple hooks are registered, all are called (if none raise
 
259
        errors).
 
260
        """
 
261
        branch = self.make_branch('source')
 
262
        hook_calls_1 = self.install_logging_hook('pre')
 
263
        hook_calls_2 = self.install_logging_hook('pre')
 
264
        self.assertIsNot(hook_calls_1, hook_calls_2)
 
265
        branch.set_last_revision_info(0, NULL_REVISION)
 
266
        # Both hooks are called.
 
267
        if isinstance(branch, RemoteBranch):
 
268
            count = 2
 
269
        else:
 
270
            count = 1
 
271
        self.assertEqual(len(hook_calls_1), count)
 
272
        self.assertEqual(len(hook_calls_2), count)
 
273
 
 
274
    def test_explicit_reject_by_hook(self):
 
275
        """If a hook raises TipChangeRejected, the change does not take effect.
 
276
 
 
277
        TipChangeRejected exceptions are propagated, not wrapped in HookFailed.
 
278
        """
 
279
        branch = self.make_branch_with_revision_ids(
 
280
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
281
        def hook_that_rejects(params):
 
282
            raise TipChangeRejected('rejection message')
 
283
        Branch.hooks.install_named_hook(
 
284
            'pre_change_branch_tip', hook_that_rejects, None)
 
285
        self.assertRaises(
 
286
            TipChangeRejected, branch.set_last_revision_info, 0, NULL_REVISION)
 
287
        # The revision info is unchanged.
 
288
        self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
 
289
 
 
290
 
 
291
class TestPostChangeBranchTip(ChangeBranchTipTestCase):
 
292
    """Tests for post_change_branch_tip hook.
 
293
 
 
294
    Most of these tests are very similar to the tests in
 
295
    TestPostChangeBranchTip.
 
296
    """
 
297
 
 
298
    def test_hook_runs_after_change(self):
 
299
        """The hook runs *after* the branch's last_revision_info has changed.
 
300
        """
 
301
        branch = self.make_branch_with_revision_ids('revid-one')
 
302
        def assertBranchAtRevision1(params):
 
303
            self.assertEquals(
 
304
                (0, NULL_REVISION), params.branch.last_revision_info())
 
305
        Branch.hooks.install_named_hook(
 
306
            'post_change_branch_tip', assertBranchAtRevision1, None)
 
307
        branch.set_last_revision_info(0, NULL_REVISION)
 
308
 
 
309
    def test_empty_history(self):
 
310
        branch = self.make_branch('source')
 
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)
 
315
        self.assertHookCalls(expected_params, branch, hook_calls)
 
316
 
 
317
    def test_nonempty_history(self):
 
318
        # some branches require that their history be set to a revision in the
 
319
        # repository, so we need to make a branch with non-empty history for
 
320
        # this test.
 
321
        branch = self.make_branch_with_revision_ids(
 
322
            'one-\xc2\xb5', 'two-\xc2\xb5')
 
323
        hook_calls = self.install_logging_hook('post')
 
324
        branch.set_last_revision_info(1, 'one-\xc2\xb5')
 
325
        expected_params = ChangeBranchTipParams(
 
326
            branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
 
327
        self.assertHookCalls(expected_params, branch, hook_calls)
 
328
 
 
329
    def test_branch_is_locked(self):
 
330
        """The branch passed to the hook is locked."""
 
331
        branch = self.make_branch('source')
 
332
        def assertBranchIsLocked(params):
 
333
            self.assertTrue(params.branch.is_locked())
 
334
        Branch.hooks.install_named_hook(
 
335
            'post_change_branch_tip', assertBranchIsLocked, None)
 
336
        branch.set_last_revision_info(0, NULL_REVISION)
 
337
 
 
338
    def test_calls_all_hooks_no_errors(self):
 
339
        """If multiple hooks are registered, all are called (if none raise
 
340
        errors).
 
341
        """
 
342
        branch = self.make_branch('source')
 
343
        hook_calls_1 = self.install_logging_hook('post')
 
344
        hook_calls_2 = self.install_logging_hook('post')
 
345
        self.assertIsNot(hook_calls_1, hook_calls_2)
 
346
        branch.set_last_revision_info(0, NULL_REVISION)
 
347
        # Both hooks are called.
 
348
        if isinstance(branch, RemoteBranch):
 
349
            count = 2
 
350
        else:
 
351
            count = 1
 
352
        self.assertEqual(len(hook_calls_1), count)
 
353
        self.assertEqual(len(hook_calls_2), count)
 
354
 
 
355
 
 
356
class TestAllMethodsThatChangeTipWillRunHooks(ChangeBranchTipTestCase):
 
357
    """Every method of Branch that changes a branch tip will invoke the
 
358
    pre/post_change_branch_tip hooks.
 
359
    """
 
360
 
 
361
    def setUp(self):
 
362
        ChangeBranchTipTestCase.setUp(self)
 
363
        self.installPreAndPostHooks()
 
364
 
 
365
    def installPreAndPostHooks(self):
 
366
        self.pre_hook_calls = self.install_logging_hook('pre')
 
367
        self.post_hook_calls = self.install_logging_hook('post')
 
368
 
 
369
    def resetHookCalls(self):
 
370
        del self.pre_hook_calls[:], self.post_hook_calls[:]
 
371
 
 
372
    def assertPreAndPostHooksWereInvoked(self, branch, smart_enabled):
 
373
        """assert that both pre and post hooks were called
 
374
 
 
375
        :param smart_enabled: The method invoked is one that should be
 
376
            smart server ready.
 
377
        """
 
378
        # Check for the number of invocations expected. One invocation is
 
379
        # local, one is remote (if the branch is remote).
 
380
        if smart_enabled and isinstance(branch, RemoteBranch):
 
381
            length = 2
 
382
        else:
 
383
            length = 1
 
384
        self.assertEqual(length, len(self.pre_hook_calls))
 
385
        self.assertEqual(length, len(self.post_hook_calls))
 
386
 
 
387
    def test_set_revision_history(self):
 
388
        branch = self.make_branch('')
 
389
        branch.set_revision_history([])
 
390
        self.assertPreAndPostHooksWereInvoked(branch, True)
 
391
 
 
392
    def test_set_last_revision_info(self):
 
393
        branch = self.make_branch('')
 
394
        branch.set_last_revision_info(0, NULL_REVISION)
 
395
        self.assertPreAndPostHooksWereInvoked(branch, True)
 
396
 
 
397
    def test_generate_revision_history(self):
 
398
        branch = self.make_branch('')
 
399
        branch.generate_revision_history(NULL_REVISION)
 
400
        # NB: for HPSS protocols < v3, the server does not invoke branch tip
 
401
        # change events on generate_revision_history, as the change is done
 
402
        # directly by the client over the VFS.
 
403
        self.assertPreAndPostHooksWereInvoked(branch, True)
 
404
 
 
405
    def test_pull(self):
 
406
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
 
407
        self.resetHookCalls()
 
408
        destination_branch = self.make_branch('destination')
 
409
        destination_branch.pull(source_branch)
 
410
        self.assertPreAndPostHooksWereInvoked(destination_branch, False)
 
411
 
 
412
    def test_push(self):
 
413
        source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
 
414
        self.resetHookCalls()
 
415
        destination_branch = self.make_branch('destination')
 
416
        source_branch.push(destination_branch)
 
417
        self.assertPreAndPostHooksWereInvoked(destination_branch, True)