202
175
BranchFormat.unregister_format(format)
203
176
self.make_branch_and_tree('bar')
206
class TestBranch67(object):
207
"""Common tests for both branch 6 and 7 which are mostly the same."""
209
def get_format_name(self):
210
raise NotImplementedError(self.get_format_name)
212
def get_format_name_subtree(self):
213
raise NotImplementedError(self.get_format_name)
216
raise NotImplementedError(self.get_class)
178
def test_checkout_format(self):
179
branch = self.make_repository('repository', shared=True)
180
branch = self.make_branch('repository/branch',
182
tree = branch.create_checkout('checkout')
183
self.assertIs(tree.branch.__class__, _mod_branch.BzrBranch5)
186
class TestBranch6(TestCaseWithTransport):
218
188
def test_creation(self):
219
189
format = BzrDirMetaFormat1()
220
190
format.set_branch_format(_mod_branch.BzrBranchFormat6())
221
191
branch = self.make_branch('a', format=format)
222
self.assertIsInstance(branch, self.get_class())
223
branch = self.make_branch('b', format=self.get_format_name())
224
self.assertIsInstance(branch, self.get_class())
192
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
193
branch = self.make_branch('b', format='dirstate-with-subtree')
194
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
225
195
branch = _mod_branch.Branch.open('a')
226
self.assertIsInstance(branch, self.get_class())
196
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
228
198
def test_layout(self):
229
branch = self.make_branch('a', format=self.get_format_name())
199
branch = self.make_branch('a', format='dirstate-with-subtree')
230
200
self.failUnlessExists('a/.bzr/branch/last-revision')
231
201
self.failIfExists('a/.bzr/branch/revision-history')
232
self.failIfExists('a/.bzr/branch/references')
234
203
def test_config(self):
235
204
"""Ensure that all configuration data is stored in the branch"""
236
branch = self.make_branch('a', format=self.get_format_name())
205
branch = self.make_branch('a', format='dirstate-with-subtree')
237
206
branch.set_parent('http://bazaar-vcs.org')
238
207
self.failIfExists('a/.bzr/branch/parent')
239
208
self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
246
215
self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
248
217
def test_set_revision_history(self):
249
builder = self.make_branch_builder('.', format=self.get_format_name())
250
builder.build_snapshot('foo', None,
251
[('add', ('', None, 'directory', None))],
253
builder.build_snapshot('bar', None, [], message='bar')
254
branch = builder.get_branch()
256
self.addCleanup(branch.unlock)
257
branch.set_revision_history(['foo', 'bar'])
258
branch.set_revision_history(['foo'])
259
self.assertRaises(errors.NotLefthandHistory,
260
branch.set_revision_history, ['bar'])
218
tree = self.make_branch_and_memory_tree('.',
219
format='dirstate-with-subtree')
223
tree.commit('foo', rev_id='foo')
224
tree.commit('bar', rev_id='bar')
225
tree.branch.set_revision_history(['foo', 'bar'])
226
tree.branch.set_revision_history(['foo'])
227
self.assertRaises(errors.NotLefthandHistory,
228
tree.branch.set_revision_history, ['bar'])
232
def test_append_revision(self):
233
tree = self.make_branch_and_tree('branch1',
234
format='dirstate-with-subtree')
237
tree.commit('foo', rev_id='foo')
238
tree.commit('bar', rev_id='bar')
239
tree.commit('baz', rev_id='baz')
240
tree.set_last_revision('bar')
241
tree.branch.set_last_revision_info(2, 'bar')
242
tree.commit('qux', rev_id='qux')
243
tree.add_parent_tree_id('baz')
244
tree.commit('qux', rev_id='quxx')
245
tree.branch.set_last_revision_info(0, 'null:')
246
self.assertRaises(errors.NotLeftParentDescendant,
247
tree.branch.append_revision, 'bar')
248
tree.branch.append_revision('foo')
249
self.assertRaises(errors.NotLeftParentDescendant,
250
tree.branch.append_revision, 'baz')
251
tree.branch.append_revision('bar')
252
tree.branch.append_revision('baz')
253
self.assertRaises(errors.NotLeftParentDescendant,
254
tree.branch.append_revision, 'quxx')
262
258
def do_checkout_test(self, lightweight=False):
263
tree = self.make_branch_and_tree('source',
264
format=self.get_format_name_subtree())
259
tree = self.make_branch_and_tree('source', format='dirstate-with-subtree')
265
260
subtree = self.make_branch_and_tree('source/subtree',
266
format=self.get_format_name_subtree())
261
format='dirstate-with-subtree')
267
262
subsubtree = self.make_branch_and_tree('source/subtree/subsubtree',
268
format=self.get_format_name_subtree())
263
format='dirstate-with-subtree')
269
264
self.build_tree(['source/subtree/file',
270
265
'source/subtree/subsubtree/file'])
271
266
subsubtree.add('file')
287
282
self.assertEndsWith(subbranch.base, 'target/subtree/subsubtree/')
289
285
def test_checkout_with_references(self):
290
286
self.do_checkout_test()
292
288
def test_light_checkout_with_references(self):
293
289
self.do_checkout_test(lightweight=True)
295
def test_set_push(self):
296
branch = self.make_branch('source', format=self.get_format_name())
297
branch.get_config().set_user_option('push_location', 'old',
298
store=config.STORE_LOCATION)
301
warnings.append(args[0] % args[1:])
302
_warning = trace.warning
303
trace.warning = warning
305
branch.set_push_location('new')
307
trace.warning = _warning
308
self.assertEqual(warnings[0], 'Value "new" is masked by "old" from '
312
class TestBranch6(TestBranch67, TestCaseWithTransport):
315
return _mod_branch.BzrBranch6
317
def get_format_name(self):
318
return "dirstate-tags"
320
def get_format_name_subtree(self):
321
return "dirstate-with-subtree"
323
def test_set_stacked_on_url_errors(self):
324
branch = self.make_branch('a', format=self.get_format_name())
325
self.assertRaises(errors.UnstackableBranchFormat,
326
branch.set_stacked_on_url, None)
328
def test_default_stacked_location(self):
329
branch = self.make_branch('a', format=self.get_format_name())
330
self.assertRaises(errors.UnstackableBranchFormat, branch.get_stacked_on_url)
333
class TestBranch7(TestBranch67, TestCaseWithTransport):
336
return _mod_branch.BzrBranch7
338
def get_format_name(self):
341
def get_format_name_subtree(self):
342
return "development-subtree"
344
def test_set_stacked_on_url_unstackable_repo(self):
345
repo = self.make_repository('a', format='dirstate-tags')
346
control = repo.bzrdir
347
branch = _mod_branch.BzrBranchFormat7().initialize(control)
348
target = self.make_branch('b')
349
self.assertRaises(errors.UnstackableRepositoryFormat,
350
branch.set_stacked_on_url, target.base)
352
def test_clone_stacked_on_unstackable_repo(self):
353
repo = self.make_repository('a', format='dirstate-tags')
354
control = repo.bzrdir
355
branch = _mod_branch.BzrBranchFormat7().initialize(control)
356
# Calling clone should not raise UnstackableRepositoryFormat.
357
cloned_bzrdir = control.clone('cloned')
359
def _test_default_stacked_location(self):
360
branch = self.make_branch('a', format=self.get_format_name())
361
self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
363
def test_stack_and_unstack(self):
364
branch = self.make_branch('a', format=self.get_format_name())
365
target = self.make_branch_and_tree('b', format=self.get_format_name())
366
branch.set_stacked_on_url(target.branch.base)
367
self.assertEqual(target.branch.base, branch.get_stacked_on_url())
368
revid = target.commit('foo')
369
self.assertTrue(branch.repository.has_revision(revid))
370
branch.set_stacked_on_url(None)
371
self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
372
self.assertFalse(branch.repository.has_revision(revid))
374
def test_open_opens_stacked_reference(self):
375
branch = self.make_branch('a', format=self.get_format_name())
376
target = self.make_branch_and_tree('b', format=self.get_format_name())
377
branch.set_stacked_on_url(target.branch.base)
378
branch = branch.bzrdir.open_branch()
379
revid = target.commit('foo')
380
self.assertTrue(branch.repository.has_revision(revid))
383
class BzrBranch8(TestCaseWithTransport):
385
def make_branch(self, location, format=None):
387
format = bzrdir.format_registry.make_bzrdir('1.9')
388
format.set_branch_format(_mod_branch.BzrBranchFormat8())
389
return TestCaseWithTransport.make_branch(self, location, format=format)
391
def create_branch_with_reference(self):
392
branch = self.make_branch('branch')
393
branch._set_all_reference_info({'file-id': ('path', 'location')})
397
def instrument_branch(branch, gets):
398
old_get = branch._transport.get
399
def get(*args, **kwargs):
400
gets.append((args, kwargs))
401
return old_get(*args, **kwargs)
402
branch._transport.get = get
404
def test_reference_info_caching_read_locked(self):
406
branch = self.create_branch_with_reference()
408
self.addCleanup(branch.unlock)
409
self.instrument_branch(branch, gets)
410
branch.get_reference_info('file-id')
411
branch.get_reference_info('file-id')
412
self.assertEqual(1, len(gets))
414
def test_reference_info_caching_read_unlocked(self):
416
branch = self.create_branch_with_reference()
417
self.instrument_branch(branch, gets)
418
branch.get_reference_info('file-id')
419
branch.get_reference_info('file-id')
420
self.assertEqual(2, len(gets))
422
def test_reference_info_caching_write_locked(self):
424
branch = self.make_branch('branch')
426
self.instrument_branch(branch, gets)
427
self.addCleanup(branch.unlock)
428
branch._set_all_reference_info({'file-id': ('path2', 'location2')})
429
path, location = branch.get_reference_info('file-id')
430
self.assertEqual(0, len(gets))
431
self.assertEqual('path2', path)
432
self.assertEqual('location2', location)
434
def test_reference_info_caches_cleared(self):
435
branch = self.make_branch('branch')
437
branch.set_reference_info('file-id', 'path2', 'location2')
439
doppelganger = Branch.open('branch')
440
doppelganger.set_reference_info('file-id', 'path3', 'location3')
441
self.assertEqual(('path3', 'location3'),
442
branch.get_reference_info('file-id'))
444
291
class TestBranchReference(TestCaseWithTransport):
445
292
"""Tests for the branch reference facility."""
478
314
self.assertTrue("set_rh" in hooks, "set_rh not in %s" % hooks)
479
315
self.assertTrue("post_push" in hooks, "post_push not in %s" % hooks)
480
316
self.assertTrue("post_commit" in hooks, "post_commit not in %s" % hooks)
481
self.assertTrue("pre_commit" in hooks, "pre_commit not in %s" % hooks)
482
317
self.assertTrue("post_pull" in hooks, "post_pull not in %s" % hooks)
483
318
self.assertTrue("post_uncommit" in hooks, "post_uncommit not in %s" % hooks)
484
self.assertTrue("post_change_branch_tip" in hooks,
485
"post_change_branch_tip not in %s" % hooks)
487
320
def test_installed_hooks_are_BranchHooks(self):
488
321
"""The installed hooks object should be a BranchHooks."""
489
322
# the installed hooks are saved in self._preserved_hooks.
490
self.assertIsInstance(self._preserved_hooks[_mod_branch.Branch][1],
323
self.assertIsInstance(self._preserved_hooks, BranchHooks)
325
def test_install_hook_raises_unknown_hook(self):
326
"""install_hook should raise UnknownHook if a hook is unknown."""
327
hooks = BranchHooks()
328
self.assertRaises(UnknownHook, hooks.install_hook, 'silly', None)
330
def test_install_hook_appends_known_hook(self):
331
"""install_hook should append the callable for known hooks."""
332
hooks = BranchHooks()
333
hooks.install_hook('set_rh', None)
334
self.assertEqual(hooks['set_rh'], [None])
494
337
class TestPullResult(TestCase):
503
346
# it's still supported
504
347
a = "%d revisions pulled" % r
505
348
self.assertEqual(a, "10 revisions pulled")
509
class _StubLockable(object):
510
"""Helper for TestRunWithWriteLockedTarget."""
512
def __init__(self, calls, unlock_exc=None):
514
self.unlock_exc = unlock_exc
516
def lock_write(self):
517
self.calls.append('lock_write')
520
self.calls.append('unlock')
521
if self.unlock_exc is not None:
522
raise self.unlock_exc
525
class _ErrorFromCallable(Exception):
526
"""Helper for TestRunWithWriteLockedTarget."""
529
class _ErrorFromUnlock(Exception):
530
"""Helper for TestRunWithWriteLockedTarget."""
533
class TestRunWithWriteLockedTarget(TestCase):
534
"""Tests for _run_with_write_locked_target."""
540
def func_that_returns_ok(self):
541
self._calls.append('func called')
544
def func_that_raises(self):
545
self._calls.append('func called')
546
raise _ErrorFromCallable()
548
def test_success_unlocks(self):
549
lockable = _StubLockable(self._calls)
550
result = _run_with_write_locked_target(
551
lockable, self.func_that_returns_ok)
552
self.assertEqual('ok', result)
553
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
555
def test_exception_unlocks_and_propagates(self):
556
lockable = _StubLockable(self._calls)
557
self.assertRaises(_ErrorFromCallable,
558
_run_with_write_locked_target, lockable, self.func_that_raises)
559
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
561
def test_callable_succeeds_but_error_during_unlock(self):
562
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
563
self.assertRaises(_ErrorFromUnlock,
564
_run_with_write_locked_target, lockable, self.func_that_returns_ok)
565
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
567
def test_error_during_unlock_does_not_mask_original_error(self):
568
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
569
self.assertRaises(_ErrorFromCallable,
570
_run_with_write_locked_target, lockable, self.func_that_raises)
571
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)