130
127
my_format_registry = self.make_format_registry()
131
128
self.assertEqual('Format registered lazily',
132
129
my_format_registry.get_help('lazy'))
133
self.assertEqual('Format using knits',
130
self.assertEqual('Format using knits',
134
131
my_format_registry.get_help('knit'))
135
self.assertEqual('Format using knits',
132
self.assertEqual('Format using knits',
136
133
my_format_registry.get_help('default'))
137
134
self.assertEqual('Pre-0.8 format. Slower and does not support'
138
' checkouts or shared repositories',
135
' checkouts or shared repositories',
139
136
my_format_registry.get_help('weave'))
141
138
def test_help_topic(self):
142
139
topics = help_topics.HelpTopicRegistry()
143
registry = self.make_format_registry()
144
topics.register('current-formats', registry.help_topic,
146
topics.register('other-formats', registry.help_topic,
148
new = topics.get_detail('current-formats')
149
rest = topics.get_detail('other-formats')
140
topics.register('formats', self.make_format_registry().help_topic,
142
topic = topics.get_detail('formats')
143
new, rest = topic.split('Experimental formats')
150
144
experimental, deprecated = rest.split('Deprecated formats')
151
self.assertContainsRe(new, 'formats-help')
152
self.assertContainsRe(new,
145
self.assertContainsRe(new, 'These formats can be used')
146
self.assertContainsRe(new,
153
147
':knit:\n \(native\) \(default\) Format using knits\n')
154
self.assertContainsRe(experimental,
148
self.assertContainsRe(experimental,
155
149
':branch6:\n \(native\) Experimental successor to knit')
156
self.assertContainsRe(deprecated,
150
self.assertContainsRe(deprecated,
157
151
':lazy:\n \(native\) Format registered lazily\n')
158
152
self.assertNotContainsRe(new, 'hidden')
433
418
"""The default acquisition policy should create a standalone branch."""
434
419
my_bzrdir = self.make_bzrdir('.')
435
420
repo_policy = my_bzrdir.determine_repository_policy()
436
repo, is_new = repo_policy.acquire_repository()
421
repo = repo_policy.acquire_repository()
437
422
self.assertEqual(repo.bzrdir.root_transport.base,
438
423
my_bzrdir.root_transport.base)
439
424
self.assertFalse(repo.is_shared())
441
def test_determine_stacking_policy(self):
442
parent_bzrdir = self.make_bzrdir('.')
443
child_bzrdir = self.make_bzrdir('child')
444
parent_bzrdir.get_config().set_default_stack_on('http://example.org')
445
repo_policy = child_bzrdir.determine_repository_policy()
446
self.assertEqual('http://example.org', repo_policy._stack_on)
448
def test_determine_stacking_policy_relative(self):
449
parent_bzrdir = self.make_bzrdir('.')
450
child_bzrdir = self.make_bzrdir('child')
451
parent_bzrdir.get_config().set_default_stack_on('child2')
452
repo_policy = child_bzrdir.determine_repository_policy()
453
self.assertEqual('child2', repo_policy._stack_on)
454
self.assertEqual(parent_bzrdir.root_transport.base,
455
repo_policy._stack_on_pwd)
457
def prepare_default_stacking(self, child_format='1.6'):
458
parent_bzrdir = self.make_bzrdir('.')
459
child_branch = self.make_branch('child', format=child_format)
460
parent_bzrdir.get_config().set_default_stack_on(child_branch.base)
461
new_child_transport = parent_bzrdir.transport.clone('child2')
462
return child_branch, new_child_transport
464
def test_clone_on_transport_obeys_stacking_policy(self):
465
child_branch, new_child_transport = self.prepare_default_stacking()
466
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
467
self.assertEqual(child_branch.base,
468
new_child.open_branch().get_stacked_on_url())
470
def test_default_stacking_with_stackable_branch_unstackable_repo(self):
471
# Make stackable source branch with an unstackable repo format.
472
source_bzrdir = self.make_bzrdir('source')
473
pack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
474
source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
476
# Make a directory with a default stacking policy
477
parent_bzrdir = self.make_bzrdir('parent')
478
stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
479
parent_bzrdir.get_config().set_default_stack_on(stacked_on.base)
480
# Clone source into directory
481
target = source_bzrdir.clone(self.get_url('parent/target'))
483
def test_sprout_obeys_stacking_policy(self):
484
child_branch, new_child_transport = self.prepare_default_stacking()
485
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
486
self.assertEqual(child_branch.base,
487
new_child.open_branch().get_stacked_on_url())
489
def test_clone_ignores_policy_for_unsupported_formats(self):
490
child_branch, new_child_transport = self.prepare_default_stacking(
491
child_format='pack-0.92')
492
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
493
self.assertRaises(errors.UnstackableBranchFormat,
494
new_child.open_branch().get_stacked_on_url)
496
def test_sprout_ignores_policy_for_unsupported_formats(self):
497
child_branch, new_child_transport = self.prepare_default_stacking(
498
child_format='pack-0.92')
499
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
500
self.assertRaises(errors.UnstackableBranchFormat,
501
new_child.open_branch().get_stacked_on_url)
503
def test_sprout_upgrades_format_if_stacked_specified(self):
504
child_branch, new_child_transport = self.prepare_default_stacking(
505
child_format='pack-0.92')
506
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
508
self.assertEqual(child_branch.bzrdir.root_transport.base,
509
new_child.open_branch().get_stacked_on_url())
510
repo = new_child.open_repository()
511
self.assertTrue(repo._format.supports_external_lookups)
512
self.assertFalse(repo.supports_rich_root())
514
def test_clone_on_transport_upgrades_format_if_stacked_on_specified(self):
515
child_branch, new_child_transport = self.prepare_default_stacking(
516
child_format='pack-0.92')
517
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport,
518
stacked_on=child_branch.bzrdir.root_transport.base)
519
self.assertEqual(child_branch.bzrdir.root_transport.base,
520
new_child.open_branch().get_stacked_on_url())
521
repo = new_child.open_repository()
522
self.assertTrue(repo._format.supports_external_lookups)
523
self.assertFalse(repo.supports_rich_root())
525
def test_sprout_upgrades_to_rich_root_format_if_needed(self):
526
child_branch, new_child_transport = self.prepare_default_stacking(
527
child_format='rich-root-pack')
528
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
530
repo = new_child.open_repository()
531
self.assertTrue(repo._format.supports_external_lookups)
532
self.assertTrue(repo.supports_rich_root())
534
def test_add_fallback_repo_handles_absolute_urls(self):
535
stack_on = self.make_branch('stack_on', format='1.6')
536
repo = self.make_repository('repo', format='1.6')
537
policy = bzrdir.UseExistingRepository(repo, stack_on.base)
538
policy._add_fallback(repo)
540
def test_add_fallback_repo_handles_relative_urls(self):
541
stack_on = self.make_branch('stack_on', format='1.6')
542
repo = self.make_repository('repo', format='1.6')
543
policy = bzrdir.UseExistingRepository(repo, '.', stack_on.base)
544
policy._add_fallback(repo)
546
def test_configure_relative_branch_stacking_url(self):
547
stack_on = self.make_branch('stack_on', format='1.6')
548
stacked = self.make_branch('stack_on/stacked', format='1.6')
549
policy = bzrdir.UseExistingRepository(stacked.repository,
551
policy.configure_branch(stacked)
552
self.assertEqual('..', stacked.get_stacked_on_url())
554
def test_relative_branch_stacking_to_absolute(self):
555
stack_on = self.make_branch('stack_on', format='1.6')
556
stacked = self.make_branch('stack_on/stacked', format='1.6')
557
policy = bzrdir.UseExistingRepository(stacked.repository,
558
'.', self.get_readonly_url('stack_on'))
559
policy.configure_branch(stacked)
560
self.assertEqual(self.get_readonly_url('stack_on'),
561
stacked.get_stacked_on_url())
564
427
class ChrootedTests(TestCaseWithTransport):
565
428
"""A support class that provides readonly urls outside the local namespace.
900
758
def test_needs_conversion_different_working_tree(self):
901
759
# meta1dirs need an conversion if any element is not the default.
902
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
903
tree = self.make_branch_and_tree('tree', format='knit')
904
self.assertTrue(tree.bzrdir.needs_format_conversion(
907
def test_initialize_on_format_uses_smart_transport(self):
908
self.setup_smart_server_with_call_log()
909
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
910
transport = self.get_transport('target')
911
transport.ensure_base()
912
self.reset_smart_call_log()
913
instance = new_format.initialize_on_transport(transport)
914
self.assertIsInstance(instance, remote.RemoteBzrDir)
915
rpc_count = len(self.hpss_calls)
916
# This figure represent the amount of work to perform this use case. It
917
# is entirely ok to reduce this number if a test fails due to rpc_count
918
# being too low. If rpc_count increases, more network roundtrips have
919
# become necessary for this use case. Please do not adjust this number
920
# upwards without agreement from bzr's network support maintainers.
921
self.assertEqual(2, rpc_count)
760
old_format = bzrdir.BzrDirFormat.get_default_format()
762
new_default = bzrdir.format_registry.make_bzrdir('dirstate')
763
bzrdir.BzrDirFormat._set_default_format(new_default)
765
tree = self.make_branch_and_tree('tree', format='knit')
766
self.assertTrue(tree.bzrdir.needs_format_conversion())
768
bzrdir.BzrDirFormat._set_default_format(old_format)
924
771
class TestFormat5(TestCaseWithTransport):
925
772
"""Tests specific to the version 5 bzrdir format."""
927
774
def test_same_lockfiles_between_tree_repo_branch(self):
928
# this checks that only a single lockfiles instance is created
775
# this checks that only a single lockfiles instance is created
929
776
# for format 5 objects
930
777
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
931
778
def check_dir_components_use_same_lock(dir):
938
785
# and if we open it normally.
939
786
dir = bzrdir.BzrDir.open(self.get_url())
940
787
check_dir_components_use_same_lock(dir)
942
789
def test_can_convert(self):
943
790
# format 5 dirs are convertable
944
791
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
945
792
self.assertTrue(dir.can_convert_format())
947
794
def test_needs_conversion(self):
948
# format 5 dirs need a conversion if they are not the default,
950
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
951
# don't need to convert it to itself
952
self.assertFalse(dir.needs_format_conversion(bzrdir.BzrDirFormat5()))
953
# do need to convert it to the current default
954
self.assertTrue(dir.needs_format_conversion(
955
bzrdir.BzrDirFormat.get_default_format()))
795
# format 5 dirs need a conversion if they are not the default.
796
# and they start of not the default.
797
old_format = bzrdir.BzrDirFormat.get_default_format()
798
bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirFormat5())
800
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
801
self.assertFalse(dir.needs_format_conversion())
803
bzrdir.BzrDirFormat._set_default_format(old_format)
804
self.assertTrue(dir.needs_format_conversion())
958
807
class TestFormat6(TestCaseWithTransport):
959
808
"""Tests specific to the version 6 bzrdir format."""
961
810
def test_same_lockfiles_between_tree_repo_branch(self):
962
# this checks that only a single lockfiles instance is created
811
# this checks that only a single lockfiles instance is created
963
812
# for format 6 objects
964
813
dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
965
814
def check_dir_components_use_same_lock(dir):
1099
952
workingtree.WorkingTreeFormat3)
1102
class TestHTTPRedirections(object):
1103
"""Test redirection between two http servers.
955
class TestHTTPRedirectionLoop(object):
956
"""Test redirection loop between two http servers.
1105
958
This MUST be used by daughter classes that also inherit from
1106
959
TestCaseWithTwoWebservers.
1108
961
We can't inherit directly from TestCaseWithTwoWebservers or the
1109
962
test framework will try to create an instance which cannot
1110
run, its implementation being incomplete.
963
run, its implementation being incomplete.
966
# Should be defined by daughter classes to ensure redirection
967
# still use the same transport implementation (not currently
968
# enforced as it's a bit tricky to get right (see the FIXME
969
# in BzrDir.open_from_transport for the unique use case so
1113
973
def create_transport_readonly_server(self):
1114
return http_utils.HTTPServerRedirecting()
974
return HTTPServerRedirecting()
1116
976
def create_transport_secondary_server(self):
1117
return http_utils.HTTPServerRedirecting()
977
return HTTPServerRedirecting()
1119
979
def setUp(self):
1120
super(TestHTTPRedirections, self).setUp()
980
# Both servers redirect to each server creating a loop
981
super(TestHTTPRedirectionLoop, self).setUp()
1121
982
# The redirections will point to the new server
1122
983
self.new_server = self.get_readonly_server()
1123
984
# The requests to the old server will be redirected
1124
985
self.old_server = self.get_secondary_server()
1125
986
# Configure the redirections
1126
987
self.old_server.redirect_to(self.new_server.host, self.new_server.port)
988
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
990
def _qualified_url(self, host, port):
991
return 'http+%s://%s:%s' % (self._qualifier, host, port)
1128
993
def test_loop(self):
1129
# Both servers redirect to each other creating a loop
1130
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
1131
994
# Starting from either server should loop
1132
old_url = self._qualified_url(self.old_server.host,
995
old_url = self._qualified_url(self.old_server.host,
1133
996
self.old_server.port)
1134
997
oldt = self._transport(old_url)
1135
998
self.assertRaises(errors.NotBranchError,
1136
999
bzrdir.BzrDir.open_from_transport, oldt)
1137
new_url = self._qualified_url(self.new_server.host,
1000
new_url = self._qualified_url(self.new_server.host,
1138
1001
self.new_server.port)
1139
1002
newt = self._transport(new_url)
1140
1003
self.assertRaises(errors.NotBranchError,
1141
1004
bzrdir.BzrDir.open_from_transport, newt)
1143
def test_qualifier_preserved(self):
1144
wt = self.make_branch_and_tree('branch')
1145
old_url = self._qualified_url(self.old_server.host,
1146
self.old_server.port)
1147
start = self._transport(old_url).clone('branch')
1148
bdir = bzrdir.BzrDir.open_from_transport(start)
1149
# Redirection should preserve the qualifier, hence the transport class
1151
self.assertIsInstance(bdir.root_transport, type(start))
1154
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1155
http_utils.TestCaseWithTwoWebservers):
1007
class TestHTTPRedirections_urllib(TestHTTPRedirectionLoop,
1008
TestCaseWithTwoWebservers):
1156
1009
"""Tests redirections for urllib implementation"""
1011
_qualifier = 'urllib'
1158
1012
_transport = HttpTransport_urllib
1160
def _qualified_url(self, host, port):
1161
result = 'http+urllib://%s:%s' % (host, port)
1162
self.permit_url(result)
1167
1016
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1168
TestHTTPRedirections,
1169
http_utils.TestCaseWithTwoWebservers):
1017
TestHTTPRedirectionLoop,
1018
TestCaseWithTwoWebservers):
1170
1019
"""Tests redirections for pycurl implementation"""
1172
def _qualified_url(self, host, port):
1173
result = 'http+pycurl://%s:%s' % (host, port)
1174
self.permit_url(result)
1178
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1179
http_utils.TestCaseWithTwoWebservers):
1180
"""Tests redirections for the nosmart decorator"""
1182
_transport = NoSmartTransportDecorator
1184
def _qualified_url(self, host, port):
1185
result = 'nosmart+http://%s:%s' % (host, port)
1186
self.permit_url(result)
1190
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1191
http_utils.TestCaseWithTwoWebservers):
1192
"""Tests redirections for readonly decoratror"""
1194
_transport = ReadonlyTransportDecorator
1196
def _qualified_url(self, host, port):
1197
result = 'readonly+http://%s:%s' % (host, port)
1198
self.permit_url(result)
1021
_qualifier = 'pycurl'
1202
1024
class TestDotBzrHidden(TestCaseWithTransport):
1226
1048
b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
1227
1049
self.build_tree(['a'])
1228
1050
self.assertEquals(['a'], self.get_ls())
1231
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
1232
"""Test BzrDirFormat implementation for TestBzrDirSprout."""
1234
def _open(self, transport):
1235
return _TestBzrDir(transport, self)
1238
class _TestBzrDir(bzrdir.BzrDirMeta1):
1239
"""Test BzrDir implementation for TestBzrDirSprout.
1241
When created a _TestBzrDir already has repository and a branch. The branch
1242
is a test double as well.
1245
def __init__(self, *args, **kwargs):
1246
super(_TestBzrDir, self).__init__(*args, **kwargs)
1247
self.test_branch = _TestBranch()
1248
self.test_branch.repository = self.create_repository()
1250
def open_branch(self, unsupported=False):
1251
return self.test_branch
1253
def cloning_metadir(self, require_stacking=False):
1254
return _TestBzrDirFormat()
1257
class _TestBranchFormat(bzrlib.branch.BranchFormat):
1258
"""Test Branch format for TestBzrDirSprout."""
1261
class _TestBranch(bzrlib.branch.Branch):
1262
"""Test Branch implementation for TestBzrDirSprout."""
1264
def __init__(self, *args, **kwargs):
1265
self._format = _TestBranchFormat()
1266
super(_TestBranch, self).__init__(*args, **kwargs)
1270
def sprout(self, *args, **kwargs):
1271
self.calls.append('sprout')
1272
return _TestBranch()
1274
def copy_content_into(self, destination, revision_id=None):
1275
self.calls.append('copy_content_into')
1277
def get_parent(self):
1280
def set_parent(self, parent):
1281
self._parent = parent
1284
class TestBzrDirSprout(TestCaseWithMemoryTransport):
1286
def test_sprout_uses_branch_sprout(self):
1287
"""BzrDir.sprout calls Branch.sprout.
1289
Usually, BzrDir.sprout should delegate to the branch's sprout method
1290
for part of the work. This allows the source branch to control the
1291
choice of format for the new branch.
1293
There are exceptions, but this tests avoids them:
1294
- if there's no branch in the source bzrdir,
1295
- or if the stacking has been requested and the format needs to be
1296
overridden to satisfy that.
1298
# Make an instrumented bzrdir.
1299
t = self.get_transport('source')
1301
source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
1302
# The instrumented bzrdir has a test_branch attribute that logs calls
1303
# made to the branch contained in that bzrdir. Initially the test
1304
# branch exists but no calls have been made to it.
1305
self.assertEqual([], source_bzrdir.test_branch.calls)
1308
target_url = self.get_url('target')
1309
result = source_bzrdir.sprout(target_url, recurse='no')
1311
# The bzrdir called the branch's sprout method.
1312
self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
1314
def test_sprout_parent(self):
1315
grandparent_tree = self.make_branch('grandparent')
1316
parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
1317
branch_tree = parent.bzrdir.sprout('branch').open_branch()
1318
self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
1321
class TestBzrDirHooks(TestCaseWithMemoryTransport):
1323
def test_pre_open_called(self):
1325
bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
1326
transport = self.get_transport('foo')
1327
url = transport.base
1328
self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
1329
self.assertEqual([transport.base], [t.base for t in calls])
1331
def test_pre_open_actual_exceptions_raised(self):
1333
def fail_once(transport):
1336
raise errors.BzrError("fail")
1337
bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
1338
transport = self.get_transport('foo')
1339
url = transport.base
1340
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1341
self.assertEqual('fail', err._preformatted_string)
1343
def test_post_repo_init(self):
1344
from bzrlib.bzrdir import RepoInitHookParams
1346
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1348
self.make_repository('foo')
1349
self.assertLength(1, calls)
1351
self.assertIsInstance(params, RepoInitHookParams)
1352
self.assertTrue(hasattr(params, 'bzrdir'))
1353
self.assertTrue(hasattr(params, 'repository'))
added
removed
bzrlib/tests/test_bzrdir.py
