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
17
17
"""Tests that branch classes implement hook callouts correctly."""
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
26
class TestSetRevisionHistoryHook(TestCaseWithMemoryTransport):
20
branch as _mod_branch,
26
from bzrlib.tests import test_server
28
class ChangeBranchTipTestCase(tests.TestCaseWithMemoryTransport):
29
"""Base TestCase for testing pre/post_change_branch_tip hooks."""
31
def install_logging_hook(self, prefix):
32
"""Add a hook that logs calls made to it.
34
:returns: the list that the calls will be appended to.
37
_mod_branch.Branch.hooks.install_named_hook(
38
prefix + '_change_branch_tip', hook_calls.append, None)
41
def make_branch_with_revision_ids(self, *revision_ids):
42
"""Makes a branch with the given commits."""
43
tree = self.make_branch_and_memory_tree('source')
46
for revision_id in revision_ids:
47
tree.commit(u'Message of ' + revision_id.decode('utf8'),
53
def assertHookCalls(self, expected_params, branch, hook_calls=None,
55
if hook_calls is None:
56
hook_calls = self.hook_calls
57
if isinstance(branch, remote.RemoteBranch):
58
# For a remote branch, both the server and the client will raise
59
# this hook, and we see both in the test environment. The remote
60
# instance comes in between the clients - the client doe pre, the
61
# server does pre, the server does post, the client does post.
66
self.assertEqual(expected_params, hook_calls[offset])
67
self.assertEqual(2, len(hook_calls))
69
self.assertEqual([expected_params], hook_calls)
72
class TestSetRevisionHistoryHook(ChangeBranchTipTestCase):
29
75
self.hook_calls = []
30
TestCaseWithMemoryTransport.setUp(self)
76
super(TestSetRevisionHistoryHook, self).setUp()
32
78
def capture_set_rh_hook(self, branch, rev_history):
33
79
"""Capture post set-rh hook calls to self.hook_calls.
35
81
The call is logged, as is some state of the branch.
37
83
self.hook_calls.append(
53
99
tree.commit('empty commit', rev_id='foo')
55
101
branch = tree.branch
56
Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
102
_mod_branch.Branch.hooks.install_named_hook(
103
'set_rh', self.capture_set_rh_hook, None)
58
104
# some branches require that their history be set to a revision in the
60
106
branch.set_revision_history(['f\xc2\xb5'])
61
self.assertEqual(self.hook_calls,
62
[('set_rh', branch, ['f\xc2\xb5'], True)])
107
expected_params =('set_rh', branch, ['f\xc2\xb5'], True)
108
self.assertHookCalls(expected_params, branch)
64
110
def test_set_rh_branch_is_locked(self):
65
111
branch = self.make_branch('source')
66
Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
112
_mod_branch.Branch.hooks.install_named_hook(
113
'set_rh', self.capture_set_rh_hook, None)
68
114
branch.set_revision_history([])
69
self.assertEqual(self.hook_calls,
70
[('set_rh', branch, [], True)])
115
expected_params = ('set_rh', branch, [], True)
116
self.assertHookCalls(expected_params, branch)
72
118
def test_set_rh_calls_all_hooks_no_errors(self):
73
119
branch = self.make_branch('source')
74
Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
76
Branch.hooks.install_named_hook('set_rh', self.capture_set_rh_hook,
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)
78
124
branch.set_revision_history([])
79
self.assertEqual(self.hook_calls,
80
[('set_rh', branch, [], True),
81
('set_rh', branch, [], True),
85
class ChangeBranchTipTestCase(TestCaseWithMemoryTransport):
86
"""Base TestCase for testing pre/post_change_branch_tip hooks."""
88
def install_logging_hook(self, prefix):
89
"""Add a hook that logs calls made to it.
91
:returns: the list that the calls will be appended to.
94
Branch.hooks.install_named_hook(
95
prefix + '_change_branch_tip', hook_calls.append, None)
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')
103
for revision_id in revision_ids:
104
tree.commit(u'Message of ' + revision_id.decode('utf8'),
111
class TestOpen(TestCaseWithMemoryTransport):
125
expected_calls = [('set_rh', branch, [], True),
126
('set_rh', branch, [], True),
128
if isinstance(branch, remote.RemoteBranch):
129
# For a remote branch, both the server and the client will raise
130
# set_rh, and the server will do so first because that is where
131
# the change takes place.
132
self.assertEqual(expected_calls, self.hook_calls[2:])
133
self.assertEqual(4, len(self.hook_calls))
135
self.assertEqual(expected_calls, self.hook_calls)
138
class TestOpen(tests.TestCaseWithMemoryTransport):
113
140
def capture_hook(self, branch):
114
141
self.hook_calls.append(branch)
116
143
def install_hook(self):
117
144
self.hook_calls = []
118
Branch.hooks.install_named_hook('open', self.capture_hook, None)
145
_mod_branch.Branch.hooks.install_named_hook(
146
'open', self.capture_hook, None)
120
148
def test_create(self):
121
149
self.install_hook()
122
150
b = self.make_branch('.')
123
self.assertEqual([b], self.hook_calls)
151
if isinstance(b, remote.RemoteBranch):
152
# RemoteBranch creation:
153
if (self.transport_readonly_server
154
== test_server.ReadonlySmartTCPServer_for_testing_v2_only):
156
self.assertEqual(3, len(self.hook_calls))
157
# creates the branch via the VFS (for older servers)
158
self.assertEqual(b._real_branch, self.hook_calls[0])
159
# creates a RemoteBranch object
160
self.assertEqual(b, self.hook_calls[1])
161
# get_stacked_on_url RPC
162
self.assertRealBranch(self.hook_calls[2])
164
self.assertEqual(2, len(self.hook_calls))
166
self.assertRealBranch(self.hook_calls[0])
167
# create RemoteBranch locally
168
self.assertEqual(b, self.hook_calls[1])
170
self.assertEqual([b], self.hook_calls)
125
172
def test_open(self):
126
173
branch_url = self.make_branch('.').bzrdir.root_transport.base
127
174
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))
175
b = _mod_branch.Branch.open(branch_url)
176
if isinstance(b, remote.RemoteBranch):
177
self.assertEqual(3, len(self.hook_calls))
179
self.assertRealBranch(self.hook_calls[0])
180
# create RemoteBranch locally
181
self.assertEqual(b, self.hook_calls[1])
182
# get_stacked_on_url RPC
183
self.assertRealBranch(self.hook_calls[2])
140
185
self.assertEqual([b], self.hook_calls)
187
def assertRealBranch(self, b):
188
# Branches opened on the server don't have comparable URLs, so we just
189
# assert that it is not a RemoteBranch.
190
self.assertIsInstance(b, _mod_branch.Branch)
191
self.assertFalse(isinstance(b, remote.RemoteBranch))
143
194
class TestPreChangeBranchTip(ChangeBranchTipTestCase):
144
195
"""Tests for pre_change_branch_tip hook.
146
197
Most of these tests are very similar to the tests in
147
198
TestPostChangeBranchTip.
154
205
def assertBranchAtRevision1(params):
155
206
self.assertEquals(
156
207
(1, 'revid-one'), params.branch.last_revision_info())
157
Branch.hooks.install_named_hook(
208
_mod_branch.Branch.hooks.install_named_hook(
158
209
'pre_change_branch_tip', assertBranchAtRevision1, None)
159
branch.set_last_revision_info(0, NULL_REVISION)
210
branch.set_last_revision_info(0, revision.NULL_REVISION)
161
212
def test_hook_failure_prevents_change(self):
162
"""If a hook raises an exception, the change does not take effect.
164
Also, a HookFailed exception will be raised.
213
"""If a hook raises an exception, the change does not take effect."""
166
214
branch = self.make_branch_with_revision_ids(
167
215
'one-\xc2\xb5', 'two-\xc2\xb5')
168
216
class PearShapedError(Exception):
170
218
def hook_that_raises(params):
171
219
raise PearShapedError()
172
Branch.hooks.install_named_hook(
220
_mod_branch.Branch.hooks.install_named_hook(
173
221
'pre_change_branch_tip', hook_that_raises, None)
174
222
hook_failed_exc = self.assertRaises(
175
HookFailed, branch.set_last_revision_info, 0, NULL_REVISION)
176
self.assertIsInstance(hook_failed_exc.exc_value, PearShapedError)
224
branch.set_last_revision_info, 0, revision.NULL_REVISION)
177
225
# The revision info is unchanged.
178
226
self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
180
228
def test_empty_history(self):
181
229
branch = self.make_branch('source')
182
230
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)
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)
234
self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
188
236
def test_nonempty_history(self):
189
237
# some branches require that their history be set to a revision in the
193
241
'one-\xc2\xb5', 'two-\xc2\xb5')
194
242
hook_calls = self.install_logging_hook('pre')
195
243
branch.set_last_revision_info(1, 'one-\xc2\xb5')
196
expected_params = ChangeBranchTipParams(
244
expected_params = _mod_branch.ChangeBranchTipParams(
197
245
branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
198
self.assertEqual([expected_params], hook_calls)
246
self.assertHookCalls(expected_params, branch, hook_calls, pre=True)
200
248
def test_branch_is_locked(self):
201
249
branch = self.make_branch('source')
202
250
def assertBranchIsLocked(params):
203
251
self.assertTrue(params.branch.is_locked())
204
Branch.hooks.install_named_hook(
252
_mod_branch.Branch.hooks.install_named_hook(
205
253
'pre_change_branch_tip', assertBranchIsLocked, None)
206
branch.set_last_revision_info(0, NULL_REVISION)
254
branch.set_last_revision_info(0, revision.NULL_REVISION)
208
256
def test_calls_all_hooks_no_errors(self):
209
257
"""If multiple hooks are registered, all are called (if none raise
213
261
hook_calls_1 = self.install_logging_hook('pre')
214
262
hook_calls_2 = self.install_logging_hook('pre')
215
263
self.assertIsNot(hook_calls_1, hook_calls_2)
216
branch.set_last_revision_info(0, NULL_REVISION)
264
branch.set_last_revision_info(0, revision.NULL_REVISION)
217
265
# Both hooks are called.
218
self.assertEqual(len(hook_calls_1), 1)
219
self.assertEqual(len(hook_calls_2), 1)
266
if isinstance(branch, remote.RemoteBranch):
270
self.assertEqual(len(hook_calls_1), count)
271
self.assertEqual(len(hook_calls_2), count)
221
273
def test_explicit_reject_by_hook(self):
222
274
"""If a hook raises TipChangeRejected, the change does not take effect.
224
276
TipChangeRejected exceptions are propagated, not wrapped in HookFailed.
226
278
branch = self.make_branch_with_revision_ids(
227
279
'one-\xc2\xb5', 'two-\xc2\xb5')
228
280
def hook_that_rejects(params):
229
raise TipChangeRejected('rejection message')
230
Branch.hooks.install_named_hook(
281
raise errors.TipChangeRejected('rejection message')
282
_mod_branch.Branch.hooks.install_named_hook(
231
283
'pre_change_branch_tip', hook_that_rejects, None)
232
284
self.assertRaises(
233
TipChangeRejected, branch.set_last_revision_info, 0, NULL_REVISION)
285
errors.TipChangeRejected,
286
branch.set_last_revision_info, 0, revision.NULL_REVISION)
234
287
# The revision info is unchanged.
235
288
self.assertEqual((2, 'two-\xc2\xb5'), branch.last_revision_info())
238
291
class TestPostChangeBranchTip(ChangeBranchTipTestCase):
239
292
"""Tests for post_change_branch_tip hook.
248
301
branch = self.make_branch_with_revision_ids('revid-one')
249
302
def assertBranchAtRevision1(params):
250
303
self.assertEquals(
251
(0, NULL_REVISION), params.branch.last_revision_info())
252
Branch.hooks.install_named_hook(
304
(0, revision.NULL_REVISION), params.branch.last_revision_info())
305
_mod_branch.Branch.hooks.install_named_hook(
253
306
'post_change_branch_tip', assertBranchAtRevision1, None)
254
branch.set_last_revision_info(0, NULL_REVISION)
307
branch.set_last_revision_info(0, revision.NULL_REVISION)
256
309
def test_empty_history(self):
257
310
branch = self.make_branch('source')
258
311
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)
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
self.assertHookCalls(expected_params, branch, hook_calls)
264
317
def test_nonempty_history(self):
265
318
# some branches require that their history be set to a revision in the
269
322
'one-\xc2\xb5', 'two-\xc2\xb5')
270
323
hook_calls = self.install_logging_hook('post')
271
324
branch.set_last_revision_info(1, 'one-\xc2\xb5')
272
expected_params = ChangeBranchTipParams(
325
expected_params = _mod_branch.ChangeBranchTipParams(
273
326
branch, 2, 1, 'two-\xc2\xb5', 'one-\xc2\xb5')
274
self.assertEqual([expected_params], hook_calls)
327
self.assertHookCalls(expected_params, branch, hook_calls)
276
329
def test_branch_is_locked(self):
277
330
"""The branch passed to the hook is locked."""
278
331
branch = self.make_branch('source')
279
332
def assertBranchIsLocked(params):
280
333
self.assertTrue(params.branch.is_locked())
281
Branch.hooks.install_named_hook(
334
_mod_branch.Branch.hooks.install_named_hook(
282
335
'post_change_branch_tip', assertBranchIsLocked, None)
283
branch.set_last_revision_info(0, NULL_REVISION)
336
branch.set_last_revision_info(0, revision.NULL_REVISION)
285
338
def test_calls_all_hooks_no_errors(self):
286
339
"""If multiple hooks are registered, all are called (if none raise
312
369
def resetHookCalls(self):
313
370
del self.pre_hook_calls[:], self.post_hook_calls[:]
315
def assertPreAndPostHooksWereInvoked(self):
316
# Check for len == 1, because the hooks should only be be invoked once
318
self.assertEqual(1, len(self.pre_hook_calls))
319
self.assertEqual(1, len(self.post_hook_calls))
372
def assertPreAndPostHooksWereInvoked(self, branch, smart_enabled):
373
"""assert that both pre and post hooks were called
375
:param smart_enabled: The method invoked is one that should be
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, remote.RemoteBranch):
384
self.assertEqual(length, len(self.pre_hook_calls))
385
self.assertEqual(length, len(self.post_hook_calls))
321
387
def test_set_revision_history(self):
322
388
branch = self.make_branch('')
323
389
branch.set_revision_history([])
324
self.assertPreAndPostHooksWereInvoked()
390
self.assertPreAndPostHooksWereInvoked(branch, True)
326
392
def test_set_last_revision_info(self):
327
393
branch = self.make_branch('')
328
branch.set_last_revision_info(0, NULL_REVISION)
329
self.assertPreAndPostHooksWereInvoked()
394
branch.set_last_revision_info(0, revision.NULL_REVISION)
395
self.assertPreAndPostHooksWereInvoked(branch, True)
331
397
def test_generate_revision_history(self):
332
398
branch = self.make_branch('')
333
branch.generate_revision_history(NULL_REVISION)
334
self.assertPreAndPostHooksWereInvoked()
399
branch.generate_revision_history(revision.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)
336
405
def test_pull(self):
337
406
source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
338
407
self.resetHookCalls()
339
408
destination_branch = self.make_branch('destination')
340
409
destination_branch.pull(source_branch)
341
self.assertPreAndPostHooksWereInvoked()
410
self.assertPreAndPostHooksWereInvoked(destination_branch, False)
343
412
def test_push(self):
344
413
source_branch = self.make_branch_with_revision_ids('rev-1', 'rev-2')
345
414
self.resetHookCalls()
346
415
destination_branch = self.make_branch('destination')
347
416
source_branch.push(destination_branch)
348
self.assertPreAndPostHooksWereInvoked()
417
self.assertPreAndPostHooksWereInvoked(destination_branch, True)