190
203
self.make_branch_and_tree('bar')
193
class TestBranch6(TestCaseWithTransport):
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)
195
218
def test_creation(self):
196
219
format = BzrDirMetaFormat1()
197
220
format.set_branch_format(_mod_branch.BzrBranchFormat6())
198
221
branch = self.make_branch('a', format=format)
199
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
200
branch = self.make_branch('b', format='dirstate-tags')
201
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
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())
202
225
branch = _mod_branch.Branch.open('a')
203
self.assertIsInstance(branch, _mod_branch.BzrBranch6)
226
self.assertIsInstance(branch, self.get_class())
205
228
def test_layout(self):
206
branch = self.make_branch('a', format='dirstate-tags')
229
branch = self.make_branch('a', format=self.get_format_name())
207
230
self.failUnlessExists('a/.bzr/branch/last-revision')
208
231
self.failIfExists('a/.bzr/branch/revision-history')
232
self.failIfExists('a/.bzr/branch/references')
210
234
def test_config(self):
211
235
"""Ensure that all configuration data is stored in the branch"""
212
branch = self.make_branch('a', format='dirstate-tags')
236
branch = self.make_branch('a', format=self.get_format_name())
213
237
branch.set_parent('http://bazaar-vcs.org')
214
238
self.failIfExists('a/.bzr/branch/parent')
215
239
self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
222
246
self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
224
248
def test_set_revision_history(self):
225
tree = self.make_branch_and_memory_tree('.',
226
format='dirstate-tags')
230
tree.commit('foo', rev_id='foo')
231
tree.commit('bar', rev_id='bar')
232
tree.branch.set_revision_history(['foo', 'bar'])
233
tree.branch.set_revision_history(['foo'])
234
self.assertRaises(errors.NotLefthandHistory,
235
tree.branch.set_revision_history, ['bar'])
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'])
239
262
def do_checkout_test(self, lightweight=False):
240
tree = self.make_branch_and_tree('source', format='dirstate-with-subtree')
263
tree = self.make_branch_and_tree('source',
264
format=self.get_format_name_subtree())
241
265
subtree = self.make_branch_and_tree('source/subtree',
242
format='dirstate-with-subtree')
266
format=self.get_format_name_subtree())
243
267
subsubtree = self.make_branch_and_tree('source/subtree/subsubtree',
244
format='dirstate-with-subtree')
268
format=self.get_format_name_subtree())
245
269
self.build_tree(['source/subtree/file',
246
270
'source/subtree/subsubtree/file'])
247
271
subsubtree.add('file')
284
308
self.assertEqual(warnings[0], 'Value "new" is masked by "old" from '
285
309
'locations.conf')
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'))
287
444
class TestBranchReference(TestCaseWithTransport):
288
445
"""Tests for the branch reference facility."""
342
503
# it's still supported
343
504
a = "%d revisions pulled" % r
344
505
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)