17
17
"""Tests for the Branch facility that are not interface tests.
19
For interface tests see tests/per_branch/*.py.
19
For interface tests see `tests/per_branch/*.py`.
21
21
For concrete class tests see this file, and for meta-branch tests
22
22
also see this file.
25
from cStringIO import StringIO
28
27
branch as _mod_branch,
36
branch as _mod_bzrbranch,
39
from ..bzr.fullhistory import (
43
from ..sixish import (
48
class TestErrors(tests.TestCase):
50
def test_unstackable_branch_format(self):
53
error = _mod_branch.UnstackableBranchFormat(format, url)
55
"The branch '/foo'(foo) is not a stackable format. "
56
"You will need to upgrade the branch to permit branch stacking.",
39
60
class TestDefaultFormat(tests.TestCase):
41
62
def test_default_format(self):
42
63
# update this if you change the default branch format
43
self.assertIsInstance(_mod_branch.BranchFormat.get_default_format(),
44
_mod_branch.BzrBranchFormat7)
64
self.assertIsInstance(_mod_branch.format_registry.get_default(),
65
_mod_bzrbranch.BzrBranchFormat7)
46
67
def test_default_format_is_same_as_bzrdir_default(self):
47
68
# XXX: it might be nice if there was only one place the default was
48
69
# set, but at the moment that's not true -- mbp 20070814 --
49
70
# https://bugs.launchpad.net/bzr/+bug/132376
51
_mod_branch.BranchFormat.get_default_format(),
72
_mod_branch.format_registry.get_default(),
52
73
bzrdir.BzrDirFormat.get_default_format().get_branch_format())
54
75
def test_get_set_default_format(self):
55
76
# 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())
77
old_format = _mod_branch.format_registry.get_default()
78
_mod_branch.format_registry.set_default(
59
81
# the default branch format is used by the meta dir format
60
82
# which is not the default bzrdir format at this point
86
108
self.assertIsDirectory('.bzr/branch/lock/held', t)
88
110
def test_set_push_location(self):
89
from bzrlib.config import (locations_config_filename,
90
ensure_config_dir_exists)
91
ensure_config_dir_exists()
92
fn = locations_config_filename()
93
# write correct newlines to locations.conf
94
# by default ConfigObj uses native line-endings for new files
95
# but uses already existing line-endings if file is not empty
98
f.write('# comment\n')
111
conf = config.LocationConfig.from_string('# comment\n', '.', save=True)
102
113
branch = self.make_branch('.', format='knit')
103
114
branch.set_push_location('foo')
104
115
local_path = urlutils.local_path_from_url(branch.base[:-1])
105
self.assertFileEqual("# comment\n"
107
"push_location = foo\n"
108
"push_location:policy = norecurse\n" % local_path,
116
self.assertFileEqual(b"# comment\n"
118
b"push_location = foo\n"
119
b"push_location:policy = norecurse\n" % local_path.encode(
121
bedding.locations_config_path())
111
123
# TODO RBC 20051029 test getting a push location from a branch in a
112
124
# recursive section - that is, it appends the branch name.
115
class SampleBranchFormat(_mod_branch.BranchFormat):
127
class SampleBranchFormat(_mod_bzrbranch.BranchFormatMetadir):
116
128
"""A sample format
118
130
this format is initializable, unsupported to aid in testing the
119
131
open and open_downlevel routines.
122
def get_format_string(self):
135
def get_format_string(cls):
123
136
"""See BzrBranchFormat.get_format_string()."""
124
return "Sample branch format."
137
return b"Sample branch format."
126
def initialize(self, a_bzrdir, name=None):
139
def initialize(self, a_controldir, name=None, repository=None,
140
append_revisions_only=None):
127
141
"""Format 4 branches cannot be created."""
128
t = a_bzrdir.get_branch_transport(self, name=name)
142
t = a_controldir.get_branch_transport(self, name=name)
129
143
t.put_bytes('format', self.get_format_string())
130
144
return 'A branch'
132
146
def is_supported(self):
135
def open(self, transport, name=None, _found=False, ignore_fallbacks=False):
149
def open(self, transport, name=None, _found=False, ignore_fallbacks=False,
150
possible_transports=None):
136
151
return "opened branch."
154
# Demonstrating how lazy loading is often implemented:
155
# A constant string is created.
156
SampleSupportedBranchFormatString = b"Sample supported branch format."
158
# And the format class can then reference the constant to avoid skew.
161
class SampleSupportedBranchFormat(_mod_bzrbranch.BranchFormatMetadir):
162
"""A sample supported format."""
165
def get_format_string(cls):
166
"""See BzrBranchFormat.get_format_string()."""
167
return SampleSupportedBranchFormatString
169
def initialize(self, a_controldir, name=None, append_revisions_only=None):
170
t = a_controldir.get_branch_transport(self, name=name)
171
t.put_bytes('format', self.get_format_string())
174
def open(self, transport, name=None, _found=False, ignore_fallbacks=False,
175
possible_transports=None):
176
return "opened supported branch."
179
class SampleExtraBranchFormat(_mod_branch.BranchFormat):
180
"""A sample format that is not usable in a metadir."""
182
def get_format_string(self):
183
# This format is not usable in a metadir.
186
def network_name(self):
187
# Network name always has to be provided.
190
def initialize(self, a_controldir, name=None):
191
raise NotImplementedError(self.initialize)
193
def open(self, transport, name=None, _found=False, ignore_fallbacks=False,
194
possible_transports=None):
195
raise NotImplementedError(self.open)
139
198
class TestBzrBranchFormat(tests.TestCaseWithTransport):
140
199
"""Tests for the BzrBranchFormat facility."""
144
203
# create a branch with a few known format objects.
145
204
# this is not quite the same as
146
205
self.build_tree(["foo/", "bar/"])
147
207
def check_format(format, url):
148
dir = format._matchingbzrdir.initialize(url)
208
dir = format._matchingcontroldir.initialize(url)
149
209
dir.create_repository()
150
210
format.initialize(dir)
151
found_format = _mod_branch.BranchFormat.find_format(dir)
152
self.failUnless(isinstance(found_format, format.__class__))
153
check_format(_mod_branch.BzrBranchFormat5(), "bar")
211
found_format = _mod_bzrbranch.BranchFormatMetadir.find_format(dir)
212
self.assertIsInstance(found_format, format.__class__)
213
check_format(BzrBranchFormat5(), "bar")
215
def test_from_string(self):
216
self.assertIsInstance(
217
SampleBranchFormat.from_string(b"Sample branch format."),
219
self.assertRaises(AssertionError,
220
SampleBranchFormat.from_string, b"Different branch format.")
155
222
def test_find_format_not_branch(self):
156
223
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
157
224
self.assertRaises(errors.NotBranchError,
158
_mod_branch.BranchFormat.find_format,
225
_mod_bzrbranch.BranchFormatMetadir.find_format,
161
228
def test_find_format_unknown_format(self):
162
229
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
163
230
SampleBranchFormat().initialize(dir)
164
231
self.assertRaises(errors.UnknownFormatError,
165
_mod_branch.BranchFormat.find_format,
232
_mod_bzrbranch.BranchFormatMetadir.find_format,
235
def test_find_format_with_features(self):
236
tree = self.make_branch_and_tree('.', format='2a')
237
tree.branch.update_feature_flags({b"name": b"optional"})
238
found_format = _mod_bzrbranch.BranchFormatMetadir.find_format(
240
self.assertIsInstance(found_format, _mod_bzrbranch.BranchFormatMetadir)
241
self.assertEqual(found_format.features.get(b"name"), b"optional")
242
tree.branch.update_feature_flags({b"name": None})
243
branch = _mod_branch.Branch.open('.')
244
self.assertEqual(branch._format.features, {})
247
class TestBranchFormatRegistry(tests.TestCase):
250
super(TestBranchFormatRegistry, self).setUp()
251
self.registry = _mod_branch.BranchFormatRegistry()
253
def test_default(self):
254
self.assertIs(None, self.registry.get_default())
255
format = SampleBranchFormat()
256
self.registry.set_default(format)
257
self.assertEqual(format, self.registry.get_default())
168
259
def test_register_unregister_format(self):
169
260
format = SampleBranchFormat()
171
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
173
format.initialize(dir)
174
# register a format for it.
175
_mod_branch.BranchFormat.register_format(format)
176
# which branch.Open will refuse (not supported)
177
self.assertRaises(errors.UnsupportedFormatError,
178
_mod_branch.Branch.open, self.get_url())
179
self.make_branch_and_tree('foo')
180
# but open_downlevel will work
183
bzrdir.BzrDir.open(self.get_url()).open_branch(unsupported=True))
184
# unregister the format
185
_mod_branch.BranchFormat.unregister_format(format)
186
self.make_branch_and_tree('bar')
261
self.registry.register(format)
262
self.assertEqual(format,
263
self.registry.get(b"Sample branch format."))
264
self.registry.remove(format)
265
self.assertRaises(KeyError, self.registry.get,
266
b"Sample branch format.")
268
def test_get_all(self):
269
format = SampleBranchFormat()
270
self.assertEqual([], self.registry._get_all())
271
self.registry.register(format)
272
self.assertEqual([format], self.registry._get_all())
274
def test_register_extra(self):
275
format = SampleExtraBranchFormat()
276
self.assertEqual([], self.registry._get_all())
277
self.registry.register_extra(format)
278
self.assertEqual([format], self.registry._get_all())
280
def test_register_extra_lazy(self):
281
self.assertEqual([], self.registry._get_all())
282
self.registry.register_extra_lazy("breezy.tests.test_branch",
283
"SampleExtraBranchFormat")
284
formats = self.registry._get_all()
285
self.assertEqual(1, len(formats))
286
self.assertIsInstance(formats[0], SampleExtraBranchFormat)
189
289
class TestBranch67(object):
211
311
def test_layout(self):
212
312
branch = self.make_branch('a', format=self.get_format_name())
213
self.failUnlessExists('a/.bzr/branch/last-revision')
214
self.failIfExists('a/.bzr/branch/revision-history')
215
self.failIfExists('a/.bzr/branch/references')
313
self.assertPathExists('a/.bzr/branch/last-revision')
314
self.assertPathDoesNotExist('a/.bzr/branch/revision-history')
315
self.assertPathDoesNotExist('a/.bzr/branch/references')
217
317
def test_config(self):
218
318
"""Ensure that all configuration data is stored in the branch"""
219
319
branch = self.make_branch('a', format=self.get_format_name())
220
branch.set_parent('http://bazaar-vcs.org')
221
self.failIfExists('a/.bzr/branch/parent')
222
self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
223
branch.set_push_location('sftp://bazaar-vcs.org')
224
config = branch.get_config()._get_branch_data_config()
225
self.assertEqual('sftp://bazaar-vcs.org',
226
config.get_user_option('push_location'))
227
branch.set_bound_location('ftp://bazaar-vcs.org')
228
self.failIfExists('a/.bzr/branch/bound')
229
self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
231
def test_set_revision_history(self):
232
builder = self.make_branch_builder('.', format=self.get_format_name())
233
builder.build_snapshot('foo', None,
234
[('add', ('', None, 'directory', None))],
236
builder.build_snapshot('bar', None, [], message='bar')
237
branch = builder.get_branch()
239
self.addCleanup(branch.unlock)
240
branch.set_revision_history(['foo', 'bar'])
241
branch.set_revision_history(['foo'])
242
self.assertRaises(errors.NotLefthandHistory,
243
branch.set_revision_history, ['bar'])
320
branch.set_parent('http://example.com')
321
self.assertPathDoesNotExist('a/.bzr/branch/parent')
322
self.assertEqual('http://example.com', branch.get_parent())
323
branch.set_push_location('sftp://example.com')
324
conf = branch.get_config_stack()
325
self.assertEqual('sftp://example.com', conf.get('push_location'))
326
branch.set_bound_location('ftp://example.com')
327
self.assertPathDoesNotExist('a/.bzr/branch/bound')
328
self.assertEqual('ftp://example.com', branch.get_bound_location())
245
330
def do_checkout_test(self, lightweight=False):
246
331
tree = self.make_branch_and_tree('source',
247
format=self.get_format_name_subtree())
332
format=self.get_format_name_subtree())
248
333
subtree = self.make_branch_and_tree('source/subtree',
249
format=self.get_format_name_subtree())
334
format=self.get_format_name_subtree())
250
335
subsubtree = self.make_branch_and_tree('source/subtree/subsubtree',
251
format=self.get_format_name_subtree())
336
format=self.get_format_name_subtree())
252
337
self.build_tree(['source/subtree/file',
253
338
'source/subtree/subsubtree/file'])
254
339
subsubtree.add('file')
306
375
def test_set_stacked_on_url_errors(self):
307
376
branch = self.make_branch('a', format=self.get_format_name())
308
self.assertRaises(errors.UnstackableBranchFormat,
309
branch.set_stacked_on_url, None)
377
self.assertRaises(_mod_branch.UnstackableBranchFormat,
378
branch.set_stacked_on_url, None)
311
380
def test_default_stacked_location(self):
312
381
branch = self.make_branch('a', format=self.get_format_name())
313
self.assertRaises(errors.UnstackableBranchFormat, branch.get_stacked_on_url)
382
self.assertRaises(_mod_branch.UnstackableBranchFormat,
383
branch.get_stacked_on_url)
316
386
class TestBranch7(TestBranch67, tests.TestCaseWithTransport):
318
388
def get_class(self):
319
return _mod_branch.BzrBranch7
389
return _mod_bzrbranch.BzrBranch7
321
391
def get_format_name(self):
327
397
def test_set_stacked_on_url_unstackable_repo(self):
328
398
repo = self.make_repository('a', format='dirstate-tags')
329
control = repo.bzrdir
330
branch = _mod_branch.BzrBranchFormat7().initialize(control)
399
control = repo.controldir
400
branch = _mod_bzrbranch.BzrBranchFormat7().initialize(control)
331
401
target = self.make_branch('b')
332
402
self.assertRaises(errors.UnstackableRepositoryFormat,
333
branch.set_stacked_on_url, target.base)
403
branch.set_stacked_on_url, target.base)
335
405
def test_clone_stacked_on_unstackable_repo(self):
336
406
repo = self.make_repository('a', format='dirstate-tags')
337
control = repo.bzrdir
338
branch = _mod_branch.BzrBranchFormat7().initialize(control)
407
control = repo.controldir
408
branch = _mod_bzrbranch.BzrBranchFormat7().initialize(control)
339
409
# Calling clone should not raise UnstackableRepositoryFormat.
340
410
cloned_bzrdir = control.clone('cloned')
368
438
def make_branch(self, location, format=None):
369
439
if format is None:
370
format = bzrdir.format_registry.make_bzrdir('1.9')
371
format.set_branch_format(_mod_branch.BzrBranchFormat8())
440
format = controldir.format_registry.make_controldir('1.9')
441
format.set_branch_format(_mod_bzrbranch.BzrBranchFormat8())
372
442
return tests.TestCaseWithTransport.make_branch(
373
443
self, location, format=format)
375
445
def create_branch_with_reference(self):
376
446
branch = self.make_branch('branch')
377
branch._set_all_reference_info({'file-id': ('path', 'location')})
447
branch._set_all_reference_info({'path': ('location', b'file-id')})
381
451
def instrument_branch(branch, gets):
382
452
old_get = branch._transport.get
383
454
def get(*args, **kwargs):
384
455
gets.append((args, kwargs))
385
456
return old_get(*args, **kwargs)
391
462
branch.lock_read()
392
463
self.addCleanup(branch.unlock)
393
464
self.instrument_branch(branch, gets)
394
branch.get_reference_info('file-id')
395
branch.get_reference_info('file-id')
465
branch.get_reference_info('path')
466
branch.get_reference_info('path')
396
467
self.assertEqual(1, len(gets))
398
469
def test_reference_info_caching_read_unlocked(self):
400
471
branch = self.create_branch_with_reference()
401
472
self.instrument_branch(branch, gets)
402
branch.get_reference_info('file-id')
403
branch.get_reference_info('file-id')
473
branch.get_reference_info('path')
474
branch.get_reference_info('path')
404
475
self.assertEqual(2, len(gets))
406
477
def test_reference_info_caching_write_locked(self):
409
480
branch.lock_write()
410
481
self.instrument_branch(branch, gets)
411
482
self.addCleanup(branch.unlock)
412
branch._set_all_reference_info({'file-id': ('path2', 'location2')})
413
path, location = branch.get_reference_info('file-id')
483
branch._set_all_reference_info({'path2': ('location2', b'file-id')})
484
location, file_id = branch.get_reference_info('path2')
414
485
self.assertEqual(0, len(gets))
415
self.assertEqual('path2', path)
486
self.assertEqual(b'file-id', file_id)
416
487
self.assertEqual('location2', location)
418
489
def test_reference_info_caches_cleared(self):
419
490
branch = self.make_branch('branch')
421
branch.set_reference_info('file-id', 'path2', 'location2')
491
with branch.lock_write():
492
branch.set_reference_info('path2', 'location2', b'file-id')
423
493
doppelganger = _mod_branch.Branch.open('branch')
424
doppelganger.set_reference_info('file-id', 'path3', 'location3')
425
self.assertEqual(('path3', 'location3'),
426
branch.get_reference_info('file-id'))
494
doppelganger.set_reference_info('path3', 'location3', b'file-id')
495
self.assertEqual(('location3', b'file-id'),
496
branch.get_reference_info('path3'))
498
def _recordParentMapCalls(self, repo):
499
self._parent_map_calls = []
500
orig_get_parent_map = repo.revisions.get_parent_map
502
def get_parent_map(q):
504
self._parent_map_calls.extend([e[0] for e in q])
505
return orig_get_parent_map(q)
506
repo.revisions.get_parent_map = get_parent_map
428
509
class TestBranchReference(tests.TestCaseWithTransport):
429
510
"""Tests for the branch reference facility."""
431
512
def test_create_open_reference(self):
432
513
bzrdirformat = bzrdir.BzrDirMetaFormat1()
433
t = transport.get_transport(self.get_url('.'))
514
t = self.get_transport()
435
516
dir = bzrdirformat.initialize(self.get_url('repo'))
436
517
dir.create_repository()
437
518
target_branch = dir.create_branch()
438
519
t.mkdir('branch')
439
520
branch_dir = bzrdirformat.initialize(self.get_url('branch'))
440
made_branch = _mod_branch.BranchReferenceFormat().initialize(
521
made_branch = _mod_bzrbranch.BranchReferenceFormat().initialize(
441
522
branch_dir, target_branch=target_branch)
442
523
self.assertEqual(made_branch.base, target_branch.base)
443
524
opened_branch = branch_dir.open_branch()
444
525
self.assertEqual(opened_branch.base, target_branch.base)
446
527
def test_get_reference(self):
447
"""For a BranchReference, get_reference should reutrn the location."""
528
"""For a BranchReference, get_reference should return the location."""
448
529
branch = self.make_branch('target')
449
530
checkout = branch.create_checkout('checkout', lightweight=True)
450
reference_url = branch.bzrdir.root_transport.abspath('') + '/'
531
reference_url = branch.controldir.root_transport.abspath('') + '/'
451
532
# if the api for create_checkout changes to return different checkout types
452
533
# then this file read will fail.
453
self.assertFileEqual(reference_url, 'checkout/.bzr/branch/location')
534
self.assertFileEqual(reference_url.encode('utf-8'),
535
'checkout/.bzr/branch/location')
454
536
self.assertEqual(reference_url,
455
_mod_branch.BranchReferenceFormat().get_reference(checkout.bzrdir))
537
_mod_bzrbranch.BranchReferenceFormat().get_reference(checkout.controldir))
458
540
class TestHooks(tests.TestCaseWithTransport):
483
565
def test_post_branch_init_hook(self):
485
567
_mod_branch.Branch.hooks.install_named_hook('post_branch_init',
487
569
self.assertLength(0, calls)
488
570
branch = self.make_branch('a')
489
571
self.assertLength(1, calls)
490
572
params = calls[0]
491
573
self.assertIsInstance(params, _mod_branch.BranchInitHookParams)
492
self.assertTrue(hasattr(params, 'bzrdir'))
574
self.assertTrue(hasattr(params, 'controldir'))
493
575
self.assertTrue(hasattr(params, 'branch'))
577
def test_post_branch_init_hook_repr(self):
579
_mod_branch.Branch.hooks.install_named_hook('post_branch_init',
580
lambda params: param_reprs.append(repr(params)), None)
581
branch = self.make_branch('a')
582
self.assertLength(1, param_reprs)
583
param_repr = param_reprs[0]
584
self.assertStartsWith(param_repr, '<BranchInitHookParams of ')
495
586
def test_post_switch_hook(self):
496
from bzrlib import switch
587
from .. import switch
498
589
_mod_branch.Branch.hooks.install_named_hook('post_switch',
500
591
tree = self.make_branch_and_tree('branch-1')
501
592
self.build_tree(['branch-1/file-1'])
502
593
tree.add('file-1')
503
594
tree.commit('rev1')
504
to_branch = tree.bzrdir.sprout('branch-2').open_branch()
595
to_branch = tree.controldir.sprout('branch-2').open_branch()
505
596
self.build_tree(['branch-1/file-2'])
506
597
tree.add('file-2')
507
598
tree.remove('file-1')
508
599
tree.commit('rev2')
509
600
checkout = tree.branch.create_checkout('checkout')
510
601
self.assertLength(0, calls)
511
switch.switch(checkout.bzrdir, to_branch)
602
switch.switch(checkout.controldir, to_branch)
512
603
self.assertLength(1, calls)
513
604
params = calls[0]
514
605
self.assertIsInstance(params, _mod_branch.SwitchHookParams)
522
613
super(TestBranchOptions, self).setUp()
523
614
self.branch = self.make_branch('.')
524
self.config = self.branch.get_config()
615
self.config_stack = self.branch.get_config_stack()
526
617
def check_append_revisions_only(self, expected_value, value=None):
527
618
"""Set append_revisions_only in config and check its interpretation."""
528
619
if value is not None:
529
self.config.set_user_option('append_revisions_only', value)
620
self.config_stack.set('append_revisions_only', value)
530
621
self.assertEqual(expected_value,
531
self.branch._get_append_revisions_only())
622
self.branch.get_append_revisions_only())
533
624
def test_valid_append_revisions_only(self):
534
self.assertEquals(None,
535
self.config.get_user_option('append_revisions_only'))
625
self.assertEqual(None,
626
self.config_stack.get('append_revisions_only'))
536
627
self.check_append_revisions_only(None)
537
628
self.check_append_revisions_only(False, 'False')
538
629
self.check_append_revisions_only(True, 'True')
544
635
def test_invalid_append_revisions_only(self):
545
636
"""Ensure warning is noted on invalid settings"""
546
637
self.warnings = []
547
639
def warning(*args):
548
640
self.warnings.append(args[0] % args[1:])
549
641
self.overrideAttr(trace, 'warning', warning)
550
642
self.check_append_revisions_only(None, 'not-a-bool')
551
643
self.assertLength(1, self.warnings)
552
644
self.assertEqual(
553
'Value "not-a-bool" is not a boolean for "append_revisions_only"',
645
'Value "not-a-bool" is not valid for "append_revisions_only"',
554
646
self.warnings[0])
648
def test_use_fresh_values(self):
649
copy = _mod_branch.Branch.open(self.branch.base)
652
copy.get_config_stack().set('foo', 'bar')
655
self.assertFalse(self.branch.is_locked())
656
# Since the branch is locked, the option value won't be saved on disk
657
# so trying to access the config of locked branch via another older
658
# non-locked branch object pointing to the same branch is not supported
659
self.assertEqual(None, self.branch.get_config_stack().get('foo'))
660
# Using a newly created branch object works as expected
661
fresh = _mod_branch.Branch.open(self.branch.base)
662
self.assertEqual('bar', fresh.get_config_stack().get('foo'))
664
def test_set_from_config_get_from_config_stack(self):
665
self.branch.lock_write()
666
self.addCleanup(self.branch.unlock)
667
self.branch.get_config().set_user_option('foo', 'bar')
668
result = self.branch.get_config_stack().get('foo')
669
# https://bugs.launchpad.net/bzr/+bug/948344
670
self.expectFailure('BranchStack uses cache after set_user_option',
671
self.assertEqual, 'bar', result)
673
def test_set_from_config_stack_get_from_config(self):
674
self.branch.lock_write()
675
self.addCleanup(self.branch.unlock)
676
self.branch.get_config_stack().set('foo', 'bar')
677
# Since the branch is locked, the option value won't be saved on disk
678
# so mixing get() and get_user_option() is broken by design.
679
self.assertEqual(None,
680
self.branch.get_config().get_user_option('foo'))
682
def test_set_delays_write_when_branch_is_locked(self):
683
self.branch.lock_write()
684
self.addCleanup(self.branch.unlock)
685
self.branch.get_config_stack().set('foo', 'bar')
686
copy = _mod_branch.Branch.open(self.branch.base)
687
result = copy.get_config_stack().get('foo')
688
# Accessing from a different branch object is like accessing from a
689
# different process: the option has not been saved yet and the new
690
# value cannot be seen.
691
self.assertIs(None, result)
557
694
class TestPullResult(tests.TestCase):
559
def test_pull_result_to_int(self):
560
# to support old code, the pull result can be used as an int
561
r = _mod_branch.PullResult()
564
# this usage of results is not recommended for new code (because it
565
# doesn't describe very well what happened), but for api stability
566
# it's still supported
567
a = "%d revisions pulled" % r
568
self.assertEqual(a, "10 revisions pulled")
570
696
def test_report_changed(self):
571
697
r = _mod_branch.PullResult()
572
r.old_revid = "old-revid"
698
r.old_revid = b"old-revid"
574
r.new_revid = "new-revid"
700
r.new_revid = b"new-revid"
580
706
def test_report_unchanged(self):
581
707
r = _mod_branch.PullResult()
582
r.old_revid = "same-revid"
583
r.new_revid = "same-revid"
708
r.old_revid = b"same-revid"
709
r.new_revid = b"same-revid"
586
self.assertEqual("No revisions to pull.\n", f.getvalue())
589
class _StubLockable(object):
590
"""Helper for TestRunWithWriteLockedTarget."""
592
def __init__(self, calls, unlock_exc=None):
594
self.unlock_exc = unlock_exc
596
def lock_write(self):
597
self.calls.append('lock_write')
600
self.calls.append('unlock')
601
if self.unlock_exc is not None:
602
raise self.unlock_exc
605
class _ErrorFromCallable(Exception):
606
"""Helper for TestRunWithWriteLockedTarget."""
609
class _ErrorFromUnlock(Exception):
610
"""Helper for TestRunWithWriteLockedTarget."""
613
class TestRunWithWriteLockedTarget(tests.TestCase):
614
"""Tests for _run_with_write_locked_target."""
617
tests.TestCase.setUp(self)
620
def func_that_returns_ok(self):
621
self._calls.append('func called')
624
def func_that_raises(self):
625
self._calls.append('func called')
626
raise _ErrorFromCallable()
628
def test_success_unlocks(self):
629
lockable = _StubLockable(self._calls)
630
result = _mod_branch._run_with_write_locked_target(
631
lockable, self.func_that_returns_ok)
632
self.assertEqual('ok', result)
633
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
635
def test_exception_unlocks_and_propagates(self):
636
lockable = _StubLockable(self._calls)
637
self.assertRaises(_ErrorFromCallable,
638
_mod_branch._run_with_write_locked_target,
639
lockable, self.func_that_raises)
640
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
642
def test_callable_succeeds_but_error_during_unlock(self):
643
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
644
self.assertRaises(_ErrorFromUnlock,
645
_mod_branch._run_with_write_locked_target,
646
lockable, self.func_that_returns_ok)
647
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
649
def test_error_during_unlock_does_not_mask_original_error(self):
650
lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
651
self.assertRaises(_ErrorFromCallable,
652
_mod_branch._run_with_write_locked_target,
653
lockable, self.func_that_raises)
654
self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
712
self.assertEqual("No revisions or tags to pull.\n", f.getvalue())
added
removed
breezy/tests/test_branch.py
