48
52
# set, but at the moment that's not true -- mbp 20070814 --
49
53
# https://bugs.launchpad.net/bzr/+bug/132376
51
_mod_branch.BranchFormat.get_default_format(),
55
_mod_branch.format_registry.get_default(),
52
56
bzrdir.BzrDirFormat.get_default_format().get_branch_format())
54
58
def test_get_set_default_format(self):
55
59
# set the format and then set it back again
56
old_format = _mod_branch.BranchFormat.get_default_format()
57
_mod_branch.BranchFormat.set_default_format(SampleBranchFormat())
60
old_format = _mod_branch.format_registry.get_default()
61
_mod_branch.format_registry.set_default(SampleBranchFormat())
59
63
# the default branch format is used by the meta dir format
60
64
# which is not the default bzrdir format at this point
101
105
# recursive section - that is, it appends the branch name.
104
class SampleBranchFormat(_mod_branch.BranchFormat):
108
class SampleBranchFormat(_mod_branch.BranchFormatMetadir):
105
109
"""A sample format
107
111
this format is initializable, unsupported to aid in testing the
108
112
open and open_downlevel routines.
111
def get_format_string(self):
116
def get_format_string(cls):
112
117
"""See BzrBranchFormat.get_format_string()."""
113
118
return "Sample branch format."
115
def initialize(self, a_bzrdir, name=None):
120
def initialize(self, a_bzrdir, name=None, repository=None,
121
append_revisions_only=None):
116
122
"""Format 4 branches cannot be created."""
117
123
t = a_bzrdir.get_branch_transport(self, name=name)
118
124
t.put_bytes('format', self.get_format_string())
130
137
SampleSupportedBranchFormatString = "Sample supported branch format."
132
139
# And the format class can then reference the constant to avoid skew.
133
class SampleSupportedBranchFormat(_mod_branch.BranchFormat):
140
class SampleSupportedBranchFormat(_mod_branch.BranchFormatMetadir):
134
141
"""A sample supported format."""
136
def get_format_string(self):
144
def get_format_string(cls):
137
145
"""See BzrBranchFormat.get_format_string()."""
138
146
return SampleSupportedBranchFormatString
140
def initialize(self, a_bzrdir, name=None):
148
def initialize(self, a_bzrdir, name=None, append_revisions_only=None):
141
149
t = a_bzrdir.get_branch_transport(self, name=name)
142
150
t.put_bytes('format', self.get_format_string())
143
151
return 'A branch'
145
def open(self, transport, name=None, _found=False, ignore_fallbacks=False):
153
def open(self, transport, name=None, _found=False, ignore_fallbacks=False,
154
possible_transports=None):
146
155
return "opened supported branch."
158
class SampleExtraBranchFormat(_mod_branch.BranchFormat):
159
"""A sample format that is not usable in a metadir."""
161
def get_format_string(self):
162
# This format is not usable in a metadir.
165
def network_name(self):
166
# Network name always has to be provided.
169
def initialize(self, a_bzrdir, name=None):
170
raise NotImplementedError(self.initialize)
172
def open(self, transport, name=None, _found=False, ignore_fallbacks=False,
173
possible_transports=None):
174
raise NotImplementedError(self.open)
149
177
class TestBzrBranchFormat(tests.TestCaseWithTransport):
150
178
"""Tests for the BzrBranchFormat facility."""
168
196
factory = _mod_branch.MetaDirBranchFormatFactory(
169
197
SampleSupportedBranchFormatString,
170
198
"bzrlib.tests.test_branch", "SampleSupportedBranchFormat")
171
_mod_branch.BranchFormat.register_format(factory)
172
self.addCleanup(_mod_branch.BranchFormat.unregister_format, factory)
199
_mod_branch.format_registry.register(factory)
200
self.addCleanup(_mod_branch.format_registry.remove, factory)
173
201
b = _mod_branch.Branch.open(self.get_url())
174
202
self.assertEqual(b, "opened supported branch.")
204
def test_from_string(self):
205
self.assertIsInstance(
206
SampleBranchFormat.from_string("Sample branch format."),
208
self.assertRaises(AssertionError,
209
SampleBranchFormat.from_string, "Different branch format.")
176
211
def test_find_format_not_branch(self):
177
212
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
178
213
self.assertRaises(errors.NotBranchError,
179
_mod_branch.BranchFormat.find_format,
214
_mod_branch.BranchFormatMetadir.find_format,
182
217
def test_find_format_unknown_format(self):
183
218
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
184
219
SampleBranchFormat().initialize(dir)
185
220
self.assertRaises(errors.UnknownFormatError,
186
_mod_branch.BranchFormat.find_format,
221
_mod_branch.BranchFormatMetadir.find_format,
224
def test_find_format_with_features(self):
225
tree = self.make_branch_and_tree('.', format='2a')
226
tree.branch.update_feature_flags({"name": "optional"})
227
found_format = _mod_branch.BranchFormatMetadir.find_format(tree.bzrdir)
228
self.assertIsInstance(found_format, _mod_branch.BranchFormatMetadir)
229
self.assertEquals(found_format.features.get("name"), "optional")
230
tree.branch.update_feature_flags({"name": None})
231
branch = _mod_branch.Branch.open('.')
232
self.assertEquals(branch._format.features, {})
235
class TestBranchFormatRegistry(tests.TestCase):
238
super(TestBranchFormatRegistry, self).setUp()
239
self.registry = _mod_branch.BranchFormatRegistry()
241
def test_default(self):
242
self.assertIs(None, self.registry.get_default())
243
format = SampleBranchFormat()
244
self.registry.set_default(format)
245
self.assertEquals(format, self.registry.get_default())
189
247
def test_register_unregister_format(self):
190
248
format = SampleBranchFormat()
192
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
194
format.initialize(dir)
195
# register a format for it.
196
_mod_branch.BranchFormat.register_format(format)
197
# which branch.Open will refuse (not supported)
198
self.assertRaises(errors.UnsupportedFormatError,
199
_mod_branch.Branch.open, self.get_url())
200
self.make_branch_and_tree('foo')
201
# but open_downlevel will work
204
bzrdir.BzrDir.open(self.get_url()).open_branch(unsupported=True))
205
# unregister the format
206
_mod_branch.BranchFormat.unregister_format(format)
207
self.make_branch_and_tree('bar')
249
self.registry.register(format)
250
self.assertEquals(format,
251
self.registry.get("Sample branch format."))
252
self.registry.remove(format)
253
self.assertRaises(KeyError, self.registry.get,
254
"Sample branch format.")
256
def test_get_all(self):
257
format = SampleBranchFormat()
258
self.assertEquals([], self.registry._get_all())
259
self.registry.register(format)
260
self.assertEquals([format], self.registry._get_all())
262
def test_register_extra(self):
263
format = SampleExtraBranchFormat()
264
self.assertEquals([], self.registry._get_all())
265
self.registry.register_extra(format)
266
self.assertEquals([format], self.registry._get_all())
268
def test_register_extra_lazy(self):
269
self.assertEquals([], self.registry._get_all())
270
self.registry.register_extra_lazy("bzrlib.tests.test_branch",
271
"SampleExtraBranchFormat")
272
formats = self.registry._get_all()
273
self.assertEquals(1, len(formats))
274
self.assertIsInstance(formats[0], SampleExtraBranchFormat)
210
277
#Used by TestMetaDirBranchFormatFactory
260
327
def test_layout(self):
261
328
branch = self.make_branch('a', format=self.get_format_name())
262
self.failUnlessExists('a/.bzr/branch/last-revision')
263
self.failIfExists('a/.bzr/branch/revision-history')
264
self.failIfExists('a/.bzr/branch/references')
329
self.assertPathExists('a/.bzr/branch/last-revision')
330
self.assertPathDoesNotExist('a/.bzr/branch/revision-history')
331
self.assertPathDoesNotExist('a/.bzr/branch/references')
266
333
def test_config(self):
267
334
"""Ensure that all configuration data is stored in the branch"""
268
335
branch = self.make_branch('a', format=self.get_format_name())
269
branch.set_parent('http://bazaar-vcs.org')
270
self.failIfExists('a/.bzr/branch/parent')
271
self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
272
branch.set_push_location('sftp://bazaar-vcs.org')
273
config = branch.get_config()._get_branch_data_config()
274
self.assertEqual('sftp://bazaar-vcs.org',
275
config.get_user_option('push_location'))
276
branch.set_bound_location('ftp://bazaar-vcs.org')
277
self.failIfExists('a/.bzr/branch/bound')
278
self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
280
def test_set_revision_history(self):
281
builder = self.make_branch_builder('.', format=self.get_format_name())
282
builder.build_snapshot('foo', None,
283
[('add', ('', None, 'directory', None))],
285
builder.build_snapshot('bar', None, [], message='bar')
286
branch = builder.get_branch()
288
self.addCleanup(branch.unlock)
289
branch.set_revision_history(['foo', 'bar'])
290
branch.set_revision_history(['foo'])
291
self.assertRaises(errors.NotLefthandHistory,
292
branch.set_revision_history, ['bar'])
336
branch.set_parent('http://example.com')
337
self.assertPathDoesNotExist('a/.bzr/branch/parent')
338
self.assertEqual('http://example.com', branch.get_parent())
339
branch.set_push_location('sftp://example.com')
340
conf = branch.get_config_stack()
341
self.assertEqual('sftp://example.com', conf.get('push_location'))
342
branch.set_bound_location('ftp://example.com')
343
self.assertPathDoesNotExist('a/.bzr/branch/bound')
344
self.assertEqual('ftp://example.com', branch.get_bound_location())
294
346
def do_checkout_test(self, lightweight=False):
295
347
tree = self.make_branch_and_tree('source',
474
510
self.assertEqual(('path3', 'location3'),
475
511
branch.get_reference_info('file-id'))
513
def _recordParentMapCalls(self, repo):
514
self._parent_map_calls = []
515
orig_get_parent_map = repo.revisions.get_parent_map
516
def get_parent_map(q):
518
self._parent_map_calls.extend([e[0] for e in q])
519
return orig_get_parent_map(q)
520
repo.revisions.get_parent_map = get_parent_map
477
523
class TestBranchReference(tests.TestCaseWithTransport):
478
524
"""Tests for the branch reference facility."""
480
526
def test_create_open_reference(self):
481
527
bzrdirformat = bzrdir.BzrDirMetaFormat1()
482
t = transport.get_transport(self.get_url('.'))
528
t = self.get_transport()
484
530
dir = bzrdirformat.initialize(self.get_url('repo'))
485
531
dir.create_repository()
580
625
super(TestBranchOptions, self).setUp()
581
626
self.branch = self.make_branch('.')
582
self.config = self.branch.get_config()
627
self.config_stack = self.branch.get_config_stack()
584
629
def check_append_revisions_only(self, expected_value, value=None):
585
630
"""Set append_revisions_only in config and check its interpretation."""
586
631
if value is not None:
587
self.config.set_user_option('append_revisions_only', value)
632
self.config_stack.set('append_revisions_only', value)
588
633
self.assertEqual(expected_value,
589
self.branch._get_append_revisions_only())
634
self.branch.get_append_revisions_only())
591
636
def test_valid_append_revisions_only(self):
592
637
self.assertEquals(None,
593
self.config.get_user_option('append_revisions_only'))
638
self.config_stack.get('append_revisions_only'))
594
639
self.check_append_revisions_only(None)
595
640
self.check_append_revisions_only(False, 'False')
596
641
self.check_append_revisions_only(True, 'True')
608
653
self.check_append_revisions_only(None, 'not-a-bool')
609
654
self.assertLength(1, self.warnings)
610
655
self.assertEqual(
611
'Value "not-a-bool" is not a boolean for "append_revisions_only"',
656
'Value "not-a-bool" is not valid for "append_revisions_only"',
612
657
self.warnings[0])
659
def test_use_fresh_values(self):
660
copy = _mod_branch.Branch.open(self.branch.base)
663
copy.get_config_stack().set('foo', 'bar')
666
self.assertFalse(self.branch.is_locked())
667
# Since the branch is locked, the option value won't be saved on disk
668
# so trying to access the config of locked branch via another older
669
# non-locked branch object pointing to the same branch is not supported
670
self.assertEqual(None, self.branch.get_config_stack().get('foo'))
671
# Using a newly created branch object works as expected
672
fresh = _mod_branch.Branch.open(self.branch.base)
673
self.assertEqual('bar', fresh.get_config_stack().get('foo'))
675
def test_set_from_config_get_from_config_stack(self):
676
self.branch.lock_write()
677
self.addCleanup(self.branch.unlock)
678
self.branch.get_config().set_user_option('foo', 'bar')
679
result = self.branch.get_config_stack().get('foo')
680
# https://bugs.launchpad.net/bzr/+bug/948344
681
self.expectFailure('BranchStack uses cache after set_user_option',
682
self.assertEqual, 'bar', result)
684
def test_set_from_config_stack_get_from_config(self):
685
self.branch.lock_write()
686
self.addCleanup(self.branch.unlock)
687
self.branch.get_config_stack().set('foo', 'bar')
688
# Since the branch is locked, the option value won't be saved on disk
689
# so mixing get() and get_user_option() is broken by design.
690
self.assertEqual(None,
691
self.branch.get_config().get_user_option('foo'))
693
def test_set_delays_write_when_branch_is_locked(self):
694
self.branch.lock_write()
695
self.addCleanup(self.branch.unlock)
696
self.branch.get_config_stack().set('foo', 'bar')
697
copy = _mod_branch.Branch.open(self.branch.base)
698
result = copy.get_config_stack().get('foo')
699
# Accessing from a different branch object is like accessing from a
700
# different process: the option has not been saved yet and the new
701
# value cannot be seen.
702
self.assertIs(None, result)
615
705
class TestPullResult(tests.TestCase):
617
def test_pull_result_to_int(self):
618
# to support old code, the pull result can be used as an int
619
r = _mod_branch.PullResult()
622
# this usage of results is not recommended for new code (because it
623
# doesn't describe very well what happened), but for api stability
624
# it's still supported
625
a = "%d revisions pulled" % r
626
self.assertEqual(a, "10 revisions pulled")
628
707
def test_report_changed(self):
629
708
r = _mod_branch.PullResult()
630
709
r.old_revid = "old-revid"
641
721
r.new_revid = "same-revid"
644
self.assertEqual("No revisions to pull.\n", f.getvalue())
647
class _StubLockable(object):
648
"""Helper for TestRunWithWriteLockedTarget."""
650
def __init__(self, calls, unlock_exc=None):
652
self.unlock_exc = unlock_exc
654
def lock_write(self):
655
self.calls.append('lock_write')
658
self.calls.append('unlock')
659
if self.unlock_exc is not None:
660
raise self.unlock_exc
663
class _ErrorFromCallable(Exception):
664
"""Helper for TestRunWithWriteLockedTarget."""
667
class _ErrorFromUnlock(Exception):
668
"""Helper for TestRunWithWriteLockedTarget."""
671
class TestRunWithWriteLockedTarget(tests.TestCase):
672
"""Tests for _run_with_write_locked_target."""
675
tests.TestCase.setUp(self)
678
def func_that_returns_ok(self):
679
self._calls.append('func called')
682
def func_that_raises(self):
683
self._calls.append('func called')
684
raise _ErrorFromCallable()
686
def test_success_unlocks(self):
687
lockable = _StubLockable(self._calls)
688
result = _mod_branch._run_with_write_locked_target(
689
lockable, self.func_that_returns_ok)
690
self.assertEqual('ok', result)
691
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
693
def test_exception_unlocks_and_propagates(self):
694
lockable = _StubLockable(self._calls)
695
self.assertRaises(_ErrorFromCallable,
696
_mod_branch._run_with_write_locked_target,
697
lockable, self.func_that_raises)
698
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
700
def test_callable_succeeds_but_error_during_unlock(self):
701
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
702
self.assertRaises(_ErrorFromUnlock,
703
_mod_branch._run_with_write_locked_target,
704
lockable, self.func_that_returns_ok)
705
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
707
def test_error_during_unlock_does_not_mask_original_error(self):
708
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
709
self.assertRaises(_ErrorFromCallable,
710
_mod_branch._run_with_write_locked_target,
711
lockable, self.func_that_raises)
712
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
724
self.assertEqual("No revisions or tags to pull.\n", f.getvalue())