1
# Copyright (C) 2006-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for the BzrDir facility and any format specific tests.
19
For interface contract tests, see tests/per_bzr_dir.
35
revision as _mod_revision,
39
transport as _mod_transport,
46
from bzrlib.errors import (
48
NoColocatedBranchSupport,
50
UnsupportedFormatError,
52
from bzrlib.tests import (
54
TestCaseWithMemoryTransport,
55
TestCaseWithTransport,
58
from bzrlib.tests import(
62
from bzrlib.tests.test_http import TestWithTransport_pycurl
63
from bzrlib.transport import (
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
68
from bzrlib.transport.nosmart import NoSmartTransportDecorator
69
from bzrlib.transport.readonly import ReadonlyTransportDecorator
70
from bzrlib.repofmt import knitrepo, knitpack_repo
73
class TestDefaultFormat(TestCase):
75
def test_get_set_default_format(self):
76
old_format = bzrdir.BzrDirFormat.get_default_format()
77
# default is BzrDirMetaFormat1
78
self.assertIsInstance(old_format, bzrdir.BzrDirMetaFormat1)
79
controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
80
# creating a bzr dir should now create an instrumented dir.
82
result = bzrdir.BzrDir.create('memory:///')
83
self.assertIsInstance(result, SampleBzrDir)
85
controldir.ControlDirFormat._set_default_format(old_format)
86
self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
89
class DeprecatedBzrDirFormat(bzrdir.BzrDirFormat):
90
"""A deprecated bzr dir format."""
93
class TestFormatRegistry(TestCase):
95
def make_format_registry(self):
96
my_format_registry = controldir.ControlDirFormatRegistry()
97
my_format_registry.register('deprecated', DeprecatedBzrDirFormat,
98
'Some format. Slower and unawesome and deprecated.',
100
my_format_registry.register_lazy('lazy', 'bzrlib.tests.test_bzrdir',
101
'DeprecatedBzrDirFormat', 'Format registered lazily',
103
bzrdir.register_metadir(my_format_registry, 'knit',
104
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
105
'Format using knits',
107
my_format_registry.set_default('knit')
108
bzrdir.register_metadir(my_format_registry,
110
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
111
'Experimental successor to knit. Use at your own risk.',
112
branch_format='bzrlib.branch.BzrBranchFormat6',
114
bzrdir.register_metadir(my_format_registry,
116
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
117
'Experimental successor to knit. Use at your own risk.',
118
branch_format='bzrlib.branch.BzrBranchFormat6', hidden=True)
119
my_format_registry.register('hiddendeprecated', DeprecatedBzrDirFormat,
120
'Old format. Slower and does not support things. ', hidden=True)
121
my_format_registry.register_lazy('hiddenlazy', 'bzrlib.tests.test_bzrdir',
122
'DeprecatedBzrDirFormat', 'Format registered lazily',
123
deprecated=True, hidden=True)
124
return my_format_registry
126
def test_format_registry(self):
127
my_format_registry = self.make_format_registry()
128
my_bzrdir = my_format_registry.make_bzrdir('lazy')
129
self.assertIsInstance(my_bzrdir, DeprecatedBzrDirFormat)
130
my_bzrdir = my_format_registry.make_bzrdir('deprecated')
131
self.assertIsInstance(my_bzrdir, DeprecatedBzrDirFormat)
132
my_bzrdir = my_format_registry.make_bzrdir('default')
133
self.assertIsInstance(my_bzrdir.repository_format,
134
knitrepo.RepositoryFormatKnit1)
135
my_bzrdir = my_format_registry.make_bzrdir('knit')
136
self.assertIsInstance(my_bzrdir.repository_format,
137
knitrepo.RepositoryFormatKnit1)
138
my_bzrdir = my_format_registry.make_bzrdir('branch6')
139
self.assertIsInstance(my_bzrdir.get_branch_format(),
140
bzrlib.branch.BzrBranchFormat6)
142
def test_get_help(self):
143
my_format_registry = self.make_format_registry()
144
self.assertEqual('Format registered lazily',
145
my_format_registry.get_help('lazy'))
146
self.assertEqual('Format using knits',
147
my_format_registry.get_help('knit'))
148
self.assertEqual('Format using knits',
149
my_format_registry.get_help('default'))
150
self.assertEqual('Some format. Slower and unawesome and deprecated.',
151
my_format_registry.get_help('deprecated'))
153
def test_help_topic(self):
154
topics = help_topics.HelpTopicRegistry()
155
registry = self.make_format_registry()
156
topics.register('current-formats', registry.help_topic,
158
topics.register('other-formats', registry.help_topic,
160
new = topics.get_detail('current-formats')
161
rest = topics.get_detail('other-formats')
162
experimental, deprecated = rest.split('Deprecated formats')
163
self.assertContainsRe(new, 'formats-help')
164
self.assertContainsRe(new,
165
':knit:\n \(native\) \(default\) Format using knits\n')
166
self.assertContainsRe(experimental,
167
':branch6:\n \(native\) Experimental successor to knit')
168
self.assertContainsRe(deprecated,
169
':lazy:\n \(native\) Format registered lazily\n')
170
self.assertNotContainsRe(new, 'hidden')
172
def test_set_default_repository(self):
173
default_factory = bzrdir.format_registry.get('default')
174
old_default = [k for k, v in bzrdir.format_registry.iteritems()
175
if v == default_factory and k != 'default'][0]
176
bzrdir.format_registry.set_default_repository('dirstate-with-subtree')
178
self.assertIs(bzrdir.format_registry.get('dirstate-with-subtree'),
179
bzrdir.format_registry.get('default'))
181
repository.format_registry.get_default().__class__,
182
knitrepo.RepositoryFormatKnit3)
184
bzrdir.format_registry.set_default_repository(old_default)
186
def test_aliases(self):
187
a_registry = controldir.ControlDirFormatRegistry()
188
a_registry.register('deprecated', DeprecatedBzrDirFormat,
189
'Old format. Slower and does not support stuff',
191
a_registry.register('deprecatedalias', DeprecatedBzrDirFormat,
192
'Old format. Slower and does not support stuff',
193
deprecated=True, alias=True)
194
self.assertEqual(frozenset(['deprecatedalias']), a_registry.aliases())
197
class SampleBranch(bzrlib.branch.Branch):
198
"""A dummy branch for guess what, dummy use."""
200
def __init__(self, dir):
204
class SampleRepository(bzrlib.repository.Repository):
207
def __init__(self, dir):
211
class SampleBzrDir(bzrdir.BzrDir):
212
"""A sample BzrDir implementation to allow testing static methods."""
214
def create_repository(self, shared=False):
215
"""See BzrDir.create_repository."""
216
return "A repository"
218
def open_repository(self):
219
"""See BzrDir.open_repository."""
220
return SampleRepository(self)
222
def create_branch(self, name=None):
223
"""See BzrDir.create_branch."""
225
raise NoColocatedBranchSupport(self)
226
return SampleBranch(self)
228
def create_workingtree(self):
229
"""See BzrDir.create_workingtree."""
233
class SampleBzrDirFormat(bzrdir.BzrDirFormat):
236
this format is initializable, unsupported to aid in testing the
237
open and open_downlevel routines.
240
def get_format_string(self):
241
"""See BzrDirFormat.get_format_string()."""
242
return "Sample .bzr dir format."
244
def initialize_on_transport(self, t):
245
"""Create a bzr dir."""
247
t.put_bytes('.bzr/branch-format', self.get_format_string())
248
return SampleBzrDir(t, self)
250
def is_supported(self):
253
def open(self, transport, _found=None):
254
return "opened branch."
257
class BzrDirFormatTest1(bzrdir.BzrDirMetaFormat1):
260
def get_format_string():
261
return "Test format 1"
264
class BzrDirFormatTest2(bzrdir.BzrDirMetaFormat1):
267
def get_format_string():
268
return "Test format 2"
271
class TestBzrDirFormat(TestCaseWithTransport):
272
"""Tests for the BzrDirFormat facility."""
274
def test_find_format(self):
275
# is the right format object found for a branch?
276
# create a branch with a few known format objects.
277
bzrdir.BzrProber.formats.register(BzrDirFormatTest1.get_format_string(),
279
self.addCleanup(bzrdir.BzrProber.formats.remove,
280
BzrDirFormatTest1.get_format_string())
281
bzrdir.BzrProber.formats.register(BzrDirFormatTest2.get_format_string(),
283
self.addCleanup(bzrdir.BzrProber.formats.remove,
284
BzrDirFormatTest2.get_format_string())
285
t = self.get_transport()
286
self.build_tree(["foo/", "bar/"], transport=t)
287
def check_format(format, url):
288
format.initialize(url)
289
t = _mod_transport.get_transport_from_path(url)
290
found_format = bzrdir.BzrDirFormat.find_format(t)
291
self.assertIsInstance(found_format, format.__class__)
292
check_format(BzrDirFormatTest1(), "foo")
293
check_format(BzrDirFormatTest2(), "bar")
295
def test_find_format_nothing_there(self):
296
self.assertRaises(NotBranchError,
297
bzrdir.BzrDirFormat.find_format,
298
_mod_transport.get_transport_from_path('.'))
300
def test_find_format_unknown_format(self):
301
t = self.get_transport()
303
t.put_bytes('.bzr/branch-format', '')
304
self.assertRaises(UnknownFormatError,
305
bzrdir.BzrDirFormat.find_format,
306
_mod_transport.get_transport_from_path('.'))
308
def test_register_unregister_format(self):
309
format = SampleBzrDirFormat()
312
format.initialize(url)
313
# register a format for it.
314
bzrdir.BzrProber.formats.register(format.get_format_string(), format)
315
# which bzrdir.Open will refuse (not supported)
316
self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open, url)
317
# which bzrdir.open_containing will refuse (not supported)
318
self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open_containing, url)
319
# but open_downlevel will work
320
t = _mod_transport.get_transport_from_url(url)
321
self.assertEqual(format.open(t), bzrdir.BzrDir.open_unsupported(url))
322
# unregister the format
323
bzrdir.BzrProber.formats.remove(format.get_format_string())
324
# now open_downlevel should fail too.
325
self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
327
def test_create_branch_and_repo_uses_default(self):
328
format = SampleBzrDirFormat()
329
branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(),
331
self.assertTrue(isinstance(branch, SampleBranch))
333
def test_create_branch_and_repo_under_shared(self):
334
# creating a branch and repo in a shared repo uses the
336
format = bzrdir.format_registry.make_bzrdir('knit')
337
self.make_repository('.', shared=True, format=format)
338
branch = bzrdir.BzrDir.create_branch_and_repo(
339
self.get_url('child'), format=format)
340
self.assertRaises(errors.NoRepositoryPresent,
341
branch.bzrdir.open_repository)
343
def test_create_branch_and_repo_under_shared_force_new(self):
344
# creating a branch and repo in a shared repo can be forced to
346
format = bzrdir.format_registry.make_bzrdir('knit')
347
self.make_repository('.', shared=True, format=format)
348
branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
351
branch.bzrdir.open_repository()
353
def test_create_standalone_working_tree(self):
354
format = SampleBzrDirFormat()
355
# note this is deliberately readonly, as this failure should
356
# occur before any writes.
357
self.assertRaises(errors.NotLocalUrl,
358
bzrdir.BzrDir.create_standalone_workingtree,
359
self.get_readonly_url(), format=format)
360
tree = bzrdir.BzrDir.create_standalone_workingtree('.',
362
self.assertEqual('A tree', tree)
364
def test_create_standalone_working_tree_under_shared_repo(self):
365
# create standalone working tree always makes a repo.
366
format = bzrdir.format_registry.make_bzrdir('knit')
367
self.make_repository('.', shared=True, format=format)
368
# note this is deliberately readonly, as this failure should
369
# occur before any writes.
370
self.assertRaises(errors.NotLocalUrl,
371
bzrdir.BzrDir.create_standalone_workingtree,
372
self.get_readonly_url('child'), format=format)
373
tree = bzrdir.BzrDir.create_standalone_workingtree('child',
375
tree.bzrdir.open_repository()
377
def test_create_branch_convenience(self):
378
# outside a repo the default convenience output is a repo+branch_tree
379
format = bzrdir.format_registry.make_bzrdir('knit')
380
branch = bzrdir.BzrDir.create_branch_convenience('.', format=format)
381
branch.bzrdir.open_workingtree()
382
branch.bzrdir.open_repository()
384
def test_create_branch_convenience_possible_transports(self):
385
"""Check that the optional 'possible_transports' is recognized"""
386
format = bzrdir.format_registry.make_bzrdir('knit')
387
t = self.get_transport()
388
branch = bzrdir.BzrDir.create_branch_convenience(
389
'.', format=format, possible_transports=[t])
390
branch.bzrdir.open_workingtree()
391
branch.bzrdir.open_repository()
393
def test_create_branch_convenience_root(self):
394
"""Creating a branch at the root of a fs should work."""
395
self.vfs_transport_factory = memory.MemoryServer
396
# outside a repo the default convenience output is a repo+branch_tree
397
format = bzrdir.format_registry.make_bzrdir('knit')
398
branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
400
self.assertRaises(errors.NoWorkingTree,
401
branch.bzrdir.open_workingtree)
402
branch.bzrdir.open_repository()
404
def test_create_branch_convenience_under_shared_repo(self):
405
# inside a repo the default convenience output is a branch+ follow the
407
format = bzrdir.format_registry.make_bzrdir('knit')
408
self.make_repository('.', shared=True, format=format)
409
branch = bzrdir.BzrDir.create_branch_convenience('child',
411
branch.bzrdir.open_workingtree()
412
self.assertRaises(errors.NoRepositoryPresent,
413
branch.bzrdir.open_repository)
415
def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
416
# inside a repo the default convenience output is a branch+ follow the
417
# repo tree policy but we can override that
418
format = bzrdir.format_registry.make_bzrdir('knit')
419
self.make_repository('.', shared=True, format=format)
420
branch = bzrdir.BzrDir.create_branch_convenience('child',
421
force_new_tree=False, format=format)
422
self.assertRaises(errors.NoWorkingTree,
423
branch.bzrdir.open_workingtree)
424
self.assertRaises(errors.NoRepositoryPresent,
425
branch.bzrdir.open_repository)
427
def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
428
# inside a repo the default convenience output is a branch+ follow the
430
format = bzrdir.format_registry.make_bzrdir('knit')
431
repo = self.make_repository('.', shared=True, format=format)
432
repo.set_make_working_trees(False)
433
branch = bzrdir.BzrDir.create_branch_convenience('child',
435
self.assertRaises(errors.NoWorkingTree,
436
branch.bzrdir.open_workingtree)
437
self.assertRaises(errors.NoRepositoryPresent,
438
branch.bzrdir.open_repository)
440
def test_create_branch_convenience_under_shared_repo_no_tree_policy_force_tree(self):
441
# inside a repo the default convenience output is a branch+ follow the
442
# repo tree policy but we can override that
443
format = bzrdir.format_registry.make_bzrdir('knit')
444
repo = self.make_repository('.', shared=True, format=format)
445
repo.set_make_working_trees(False)
446
branch = bzrdir.BzrDir.create_branch_convenience('child',
447
force_new_tree=True, format=format)
448
branch.bzrdir.open_workingtree()
449
self.assertRaises(errors.NoRepositoryPresent,
450
branch.bzrdir.open_repository)
452
def test_create_branch_convenience_under_shared_repo_force_new_repo(self):
453
# inside a repo the default convenience output is overridable to give
455
format = bzrdir.format_registry.make_bzrdir('knit')
456
self.make_repository('.', shared=True, format=format)
457
branch = bzrdir.BzrDir.create_branch_convenience('child',
458
force_new_repo=True, format=format)
459
branch.bzrdir.open_repository()
460
branch.bzrdir.open_workingtree()
463
class TestRepositoryAcquisitionPolicy(TestCaseWithTransport):
465
def test_acquire_repository_standalone(self):
466
"""The default acquisition policy should create a standalone branch."""
467
my_bzrdir = self.make_bzrdir('.')
468
repo_policy = my_bzrdir.determine_repository_policy()
469
repo, is_new = repo_policy.acquire_repository()
470
self.assertEqual(repo.bzrdir.root_transport.base,
471
my_bzrdir.root_transport.base)
472
self.assertFalse(repo.is_shared())
474
def test_determine_stacking_policy(self):
475
parent_bzrdir = self.make_bzrdir('.')
476
child_bzrdir = self.make_bzrdir('child')
477
parent_bzrdir.get_config().set_default_stack_on('http://example.org')
478
repo_policy = child_bzrdir.determine_repository_policy()
479
self.assertEqual('http://example.org', repo_policy._stack_on)
481
def test_determine_stacking_policy_relative(self):
482
parent_bzrdir = self.make_bzrdir('.')
483
child_bzrdir = self.make_bzrdir('child')
484
parent_bzrdir.get_config().set_default_stack_on('child2')
485
repo_policy = child_bzrdir.determine_repository_policy()
486
self.assertEqual('child2', repo_policy._stack_on)
487
self.assertEqual(parent_bzrdir.root_transport.base,
488
repo_policy._stack_on_pwd)
490
def prepare_default_stacking(self, child_format='1.6'):
491
parent_bzrdir = self.make_bzrdir('.')
492
child_branch = self.make_branch('child', format=child_format)
493
parent_bzrdir.get_config().set_default_stack_on(child_branch.base)
494
new_child_transport = parent_bzrdir.transport.clone('child2')
495
return child_branch, new_child_transport
497
def test_clone_on_transport_obeys_stacking_policy(self):
498
child_branch, new_child_transport = self.prepare_default_stacking()
499
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
500
self.assertEqual(child_branch.base,
501
new_child.open_branch().get_stacked_on_url())
503
def test_default_stacking_with_stackable_branch_unstackable_repo(self):
504
# Make stackable source branch with an unstackable repo format.
505
source_bzrdir = self.make_bzrdir('source')
506
knitpack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
507
source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
509
# Make a directory with a default stacking policy
510
parent_bzrdir = self.make_bzrdir('parent')
511
stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
512
parent_bzrdir.get_config().set_default_stack_on(stacked_on.base)
513
# Clone source into directory
514
target = source_bzrdir.clone(self.get_url('parent/target'))
516
def test_sprout_obeys_stacking_policy(self):
517
child_branch, new_child_transport = self.prepare_default_stacking()
518
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
519
self.assertEqual(child_branch.base,
520
new_child.open_branch().get_stacked_on_url())
522
def test_clone_ignores_policy_for_unsupported_formats(self):
523
child_branch, new_child_transport = self.prepare_default_stacking(
524
child_format='pack-0.92')
525
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
526
self.assertRaises(errors.UnstackableBranchFormat,
527
new_child.open_branch().get_stacked_on_url)
529
def test_sprout_ignores_policy_for_unsupported_formats(self):
530
child_branch, new_child_transport = self.prepare_default_stacking(
531
child_format='pack-0.92')
532
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
533
self.assertRaises(errors.UnstackableBranchFormat,
534
new_child.open_branch().get_stacked_on_url)
536
def test_sprout_upgrades_format_if_stacked_specified(self):
537
child_branch, new_child_transport = self.prepare_default_stacking(
538
child_format='pack-0.92')
539
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
541
self.assertEqual(child_branch.bzrdir.root_transport.base,
542
new_child.open_branch().get_stacked_on_url())
543
repo = new_child.open_repository()
544
self.assertTrue(repo._format.supports_external_lookups)
545
self.assertFalse(repo.supports_rich_root())
547
def test_clone_on_transport_upgrades_format_if_stacked_on_specified(self):
548
child_branch, new_child_transport = self.prepare_default_stacking(
549
child_format='pack-0.92')
550
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport,
551
stacked_on=child_branch.bzrdir.root_transport.base)
552
self.assertEqual(child_branch.bzrdir.root_transport.base,
553
new_child.open_branch().get_stacked_on_url())
554
repo = new_child.open_repository()
555
self.assertTrue(repo._format.supports_external_lookups)
556
self.assertFalse(repo.supports_rich_root())
558
def test_sprout_upgrades_to_rich_root_format_if_needed(self):
559
child_branch, new_child_transport = self.prepare_default_stacking(
560
child_format='rich-root-pack')
561
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
563
repo = new_child.open_repository()
564
self.assertTrue(repo._format.supports_external_lookups)
565
self.assertTrue(repo.supports_rich_root())
567
def test_add_fallback_repo_handles_absolute_urls(self):
568
stack_on = self.make_branch('stack_on', format='1.6')
569
repo = self.make_repository('repo', format='1.6')
570
policy = bzrdir.UseExistingRepository(repo, stack_on.base)
571
policy._add_fallback(repo)
573
def test_add_fallback_repo_handles_relative_urls(self):
574
stack_on = self.make_branch('stack_on', format='1.6')
575
repo = self.make_repository('repo', format='1.6')
576
policy = bzrdir.UseExistingRepository(repo, '.', stack_on.base)
577
policy._add_fallback(repo)
579
def test_configure_relative_branch_stacking_url(self):
580
stack_on = self.make_branch('stack_on', format='1.6')
581
stacked = self.make_branch('stack_on/stacked', format='1.6')
582
policy = bzrdir.UseExistingRepository(stacked.repository,
584
policy.configure_branch(stacked)
585
self.assertEqual('..', stacked.get_stacked_on_url())
587
def test_relative_branch_stacking_to_absolute(self):
588
stack_on = self.make_branch('stack_on', format='1.6')
589
stacked = self.make_branch('stack_on/stacked', format='1.6')
590
policy = bzrdir.UseExistingRepository(stacked.repository,
591
'.', self.get_readonly_url('stack_on'))
592
policy.configure_branch(stacked)
593
self.assertEqual(self.get_readonly_url('stack_on'),
594
stacked.get_stacked_on_url())
597
class ChrootedTests(TestCaseWithTransport):
598
"""A support class that provides readonly urls outside the local namespace.
600
This is done by checking if self.transport_server is a MemoryServer. if it
601
is then we are chrooted already, if it is not then an HttpServer is used
606
super(ChrootedTests, self).setUp()
607
if not self.vfs_transport_factory == memory.MemoryServer:
608
self.transport_readonly_server = http_server.HttpServer
610
def local_branch_path(self, branch):
611
return os.path.realpath(urlutils.local_path_from_url(branch.base))
613
def test_open_containing(self):
614
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
615
self.get_readonly_url(''))
616
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
617
self.get_readonly_url('g/p/q'))
618
control = bzrdir.BzrDir.create(self.get_url())
619
branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url(''))
620
self.assertEqual('', relpath)
621
branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
622
self.assertEqual('g/p/q', relpath)
624
def test_open_containing_tree_branch_or_repository_empty(self):
625
self.assertRaises(errors.NotBranchError,
626
bzrdir.BzrDir.open_containing_tree_branch_or_repository,
627
self.get_readonly_url(''))
629
def test_open_containing_tree_branch_or_repository_all(self):
630
self.make_branch_and_tree('topdir')
631
tree, branch, repo, relpath = \
632
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
634
self.assertEqual(os.path.realpath('topdir'),
635
os.path.realpath(tree.basedir))
636
self.assertEqual(os.path.realpath('topdir'),
637
self.local_branch_path(branch))
639
osutils.realpath(os.path.join('topdir', '.bzr', 'repository')),
640
repo.bzrdir.transport.local_abspath('repository'))
641
self.assertEqual(relpath, 'foo')
643
def test_open_containing_tree_branch_or_repository_no_tree(self):
644
self.make_branch('branch')
645
tree, branch, repo, relpath = \
646
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
648
self.assertEqual(tree, None)
649
self.assertEqual(os.path.realpath('branch'),
650
self.local_branch_path(branch))
652
osutils.realpath(os.path.join('branch', '.bzr', 'repository')),
653
repo.bzrdir.transport.local_abspath('repository'))
654
self.assertEqual(relpath, 'foo')
656
def test_open_containing_tree_branch_or_repository_repo(self):
657
self.make_repository('repo')
658
tree, branch, repo, relpath = \
659
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
661
self.assertEqual(tree, None)
662
self.assertEqual(branch, None)
664
osutils.realpath(os.path.join('repo', '.bzr', 'repository')),
665
repo.bzrdir.transport.local_abspath('repository'))
666
self.assertEqual(relpath, '')
668
def test_open_containing_tree_branch_or_repository_shared_repo(self):
669
self.make_repository('shared', shared=True)
670
bzrdir.BzrDir.create_branch_convenience('shared/branch',
671
force_new_tree=False)
672
tree, branch, repo, relpath = \
673
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
675
self.assertEqual(tree, None)
676
self.assertEqual(os.path.realpath('shared/branch'),
677
self.local_branch_path(branch))
679
osutils.realpath(os.path.join('shared', '.bzr', 'repository')),
680
repo.bzrdir.transport.local_abspath('repository'))
681
self.assertEqual(relpath, '')
683
def test_open_containing_tree_branch_or_repository_branch_subdir(self):
684
self.make_branch_and_tree('foo')
685
self.build_tree(['foo/bar/'])
686
tree, branch, repo, relpath = \
687
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
689
self.assertEqual(os.path.realpath('foo'),
690
os.path.realpath(tree.basedir))
691
self.assertEqual(os.path.realpath('foo'),
692
self.local_branch_path(branch))
694
osutils.realpath(os.path.join('foo', '.bzr', 'repository')),
695
repo.bzrdir.transport.local_abspath('repository'))
696
self.assertEqual(relpath, 'bar')
698
def test_open_containing_tree_branch_or_repository_repo_subdir(self):
699
self.make_repository('bar')
700
self.build_tree(['bar/baz/'])
701
tree, branch, repo, relpath = \
702
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
704
self.assertEqual(tree, None)
705
self.assertEqual(branch, None)
707
osutils.realpath(os.path.join('bar', '.bzr', 'repository')),
708
repo.bzrdir.transport.local_abspath('repository'))
709
self.assertEqual(relpath, 'baz')
711
def test_open_containing_from_transport(self):
712
self.assertRaises(NotBranchError,
713
bzrdir.BzrDir.open_containing_from_transport,
714
_mod_transport.get_transport_from_url(self.get_readonly_url('')))
715
self.assertRaises(NotBranchError,
716
bzrdir.BzrDir.open_containing_from_transport,
717
_mod_transport.get_transport_from_url(
718
self.get_readonly_url('g/p/q')))
719
control = bzrdir.BzrDir.create(self.get_url())
720
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
721
_mod_transport.get_transport_from_url(
722
self.get_readonly_url('')))
723
self.assertEqual('', relpath)
724
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
725
_mod_transport.get_transport_from_url(
726
self.get_readonly_url('g/p/q')))
727
self.assertEqual('g/p/q', relpath)
729
def test_open_containing_tree_or_branch(self):
730
self.make_branch_and_tree('topdir')
731
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
733
self.assertEqual(os.path.realpath('topdir'),
734
os.path.realpath(tree.basedir))
735
self.assertEqual(os.path.realpath('topdir'),
736
self.local_branch_path(branch))
737
self.assertIs(tree.bzrdir, branch.bzrdir)
738
self.assertEqual('foo', relpath)
739
# opening from non-local should not return the tree
740
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
741
self.get_readonly_url('topdir/foo'))
742
self.assertEqual(None, tree)
743
self.assertEqual('foo', relpath)
745
self.make_branch('topdir/foo')
746
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
748
self.assertIs(tree, None)
749
self.assertEqual(os.path.realpath('topdir/foo'),
750
self.local_branch_path(branch))
751
self.assertEqual('', relpath)
753
def test_open_tree_or_branch(self):
754
self.make_branch_and_tree('topdir')
755
tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir')
756
self.assertEqual(os.path.realpath('topdir'),
757
os.path.realpath(tree.basedir))
758
self.assertEqual(os.path.realpath('topdir'),
759
self.local_branch_path(branch))
760
self.assertIs(tree.bzrdir, branch.bzrdir)
761
# opening from non-local should not return the tree
762
tree, branch = bzrdir.BzrDir.open_tree_or_branch(
763
self.get_readonly_url('topdir'))
764
self.assertEqual(None, tree)
766
self.make_branch('topdir/foo')
767
tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir/foo')
768
self.assertIs(tree, None)
769
self.assertEqual(os.path.realpath('topdir/foo'),
770
self.local_branch_path(branch))
772
def test_open_from_transport(self):
773
# transport pointing at bzrdir should give a bzrdir with root transport
774
# set to the given transport
775
control = bzrdir.BzrDir.create(self.get_url())
776
t = self.get_transport()
777
opened_bzrdir = bzrdir.BzrDir.open_from_transport(t)
778
self.assertEqual(t.base, opened_bzrdir.root_transport.base)
779
self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
781
def test_open_from_transport_no_bzrdir(self):
782
t = self.get_transport()
783
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
785
def test_open_from_transport_bzrdir_in_parent(self):
786
control = bzrdir.BzrDir.create(self.get_url())
787
t = self.get_transport()
789
t = t.clone('subdir')
790
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
792
def test_sprout_recursive(self):
793
tree = self.make_branch_and_tree('tree1',
794
format='dirstate-with-subtree')
795
sub_tree = self.make_branch_and_tree('tree1/subtree',
796
format='dirstate-with-subtree')
797
sub_tree.set_root_id('subtree-root')
798
tree.add_reference(sub_tree)
799
self.build_tree(['tree1/subtree/file'])
801
tree.commit('Initial commit')
802
tree2 = tree.bzrdir.sprout('tree2').open_workingtree()
804
self.addCleanup(tree2.unlock)
805
self.assertPathExists('tree2/subtree/file')
806
self.assertEqual('tree-reference', tree2.kind('subtree-root'))
808
def test_cloning_metadir(self):
809
"""Ensure that cloning metadir is suitable"""
810
bzrdir = self.make_bzrdir('bzrdir')
811
bzrdir.cloning_metadir()
812
branch = self.make_branch('branch', format='knit')
813
format = branch.bzrdir.cloning_metadir()
814
self.assertIsInstance(format.workingtree_format,
815
workingtree_4.WorkingTreeFormat6)
817
def test_sprout_recursive_treeless(self):
818
tree = self.make_branch_and_tree('tree1',
819
format='dirstate-with-subtree')
820
sub_tree = self.make_branch_and_tree('tree1/subtree',
821
format='dirstate-with-subtree')
822
tree.add_reference(sub_tree)
823
self.build_tree(['tree1/subtree/file'])
825
tree.commit('Initial commit')
826
# The following line force the orhaning to reveal bug #634470
827
tree.branch.get_config().set_user_option(
828
'bzr.transform.orphan_policy', 'move')
829
tree.bzrdir.destroy_workingtree()
830
# FIXME: subtree/.bzr is left here which allows the test to pass (or
831
# fail :-( ) -- vila 20100909
832
repo = self.make_repository('repo', shared=True,
833
format='dirstate-with-subtree')
834
repo.set_make_working_trees(False)
835
# FIXME: we just deleted the workingtree and now we want to use it ????
836
# At a minimum, we should use tree.branch below (but this fails too
837
# currently) or stop calling this test 'treeless'. Specifically, I've
838
# turn the line below into an assertRaises when 'subtree/.bzr' is
839
# orphaned and sprout tries to access the branch there (which is left
840
# by bzrdir.BzrDirMeta1.destroy_workingtree when it ignores the
841
# [DeletingParent('Not deleting', u'subtree', None)] conflict). See bug
842
# #634470. -- vila 20100909
843
self.assertRaises(errors.NotBranchError,
844
tree.bzrdir.sprout, 'repo/tree2')
845
# self.assertPathExists('repo/tree2/subtree')
846
# self.assertPathDoesNotExist('repo/tree2/subtree/file')
848
def make_foo_bar_baz(self):
849
foo = bzrdir.BzrDir.create_branch_convenience('foo').bzrdir
850
bar = self.make_branch('foo/bar').bzrdir
851
baz = self.make_branch('baz').bzrdir
854
def test_find_bzrdirs(self):
855
foo, bar, baz = self.make_foo_bar_baz()
856
t = self.get_transport()
857
self.assertEqualBzrdirs([baz, foo, bar], bzrdir.BzrDir.find_bzrdirs(t))
859
def make_fake_permission_denied_transport(self, transport, paths):
860
"""Create a transport that raises PermissionDenied for some paths."""
863
raise errors.PermissionDenied(path)
865
path_filter_server = pathfilter.PathFilteringServer(transport, filter)
866
path_filter_server.start_server()
867
self.addCleanup(path_filter_server.stop_server)
868
path_filter_transport = pathfilter.PathFilteringTransport(
869
path_filter_server, '.')
870
return (path_filter_server, path_filter_transport)
872
def assertBranchUrlsEndWith(self, expect_url_suffix, actual_bzrdirs):
873
"""Check that each branch url ends with the given suffix."""
874
for actual_bzrdir in actual_bzrdirs:
875
self.assertEndsWith(actual_bzrdir.user_url, expect_url_suffix)
877
def test_find_bzrdirs_permission_denied(self):
878
foo, bar, baz = self.make_foo_bar_baz()
879
t = self.get_transport()
880
path_filter_server, path_filter_transport = \
881
self.make_fake_permission_denied_transport(t, ['foo'])
883
self.assertBranchUrlsEndWith('/baz/',
884
bzrdir.BzrDir.find_bzrdirs(path_filter_transport))
886
smart_transport = self.make_smart_server('.',
887
backing_server=path_filter_server)
888
self.assertBranchUrlsEndWith('/baz/',
889
bzrdir.BzrDir.find_bzrdirs(smart_transport))
891
def test_find_bzrdirs_list_current(self):
892
def list_current(transport):
893
return [s for s in transport.list_dir('') if s != 'baz']
895
foo, bar, baz = self.make_foo_bar_baz()
896
t = self.get_transport()
897
self.assertEqualBzrdirs(
899
bzrdir.BzrDir.find_bzrdirs(t, list_current=list_current))
901
def test_find_bzrdirs_evaluate(self):
902
def evaluate(bzrdir):
904
repo = bzrdir.open_repository()
905
except errors.NoRepositoryPresent:
906
return True, bzrdir.root_transport.base
908
return False, bzrdir.root_transport.base
910
foo, bar, baz = self.make_foo_bar_baz()
911
t = self.get_transport()
912
self.assertEqual([baz.root_transport.base, foo.root_transport.base],
913
list(bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate)))
915
def assertEqualBzrdirs(self, first, second):
917
second = list(second)
918
self.assertEqual(len(first), len(second))
919
for x, y in zip(first, second):
920
self.assertEqual(x.root_transport.base, y.root_transport.base)
922
def test_find_branches(self):
923
root = self.make_repository('', shared=True)
924
foo, bar, baz = self.make_foo_bar_baz()
925
qux = self.make_bzrdir('foo/qux')
926
t = self.get_transport()
927
branches = bzrdir.BzrDir.find_branches(t)
928
self.assertEqual(baz.root_transport.base, branches[0].base)
929
self.assertEqual(foo.root_transport.base, branches[1].base)
930
self.assertEqual(bar.root_transport.base, branches[2].base)
932
# ensure this works without a top-level repo
933
branches = bzrdir.BzrDir.find_branches(t.clone('foo'))
934
self.assertEqual(foo.root_transport.base, branches[0].base)
935
self.assertEqual(bar.root_transport.base, branches[1].base)
938
class TestMissingRepoBranchesSkipped(TestCaseWithMemoryTransport):
940
def test_find_bzrdirs_missing_repo(self):
941
t = self.get_transport()
942
arepo = self.make_repository('arepo', shared=True)
943
abranch_url = arepo.user_url + '/abranch'
944
abranch = bzrdir.BzrDir.create(abranch_url).create_branch()
945
t.delete_tree('arepo/.bzr')
946
self.assertRaises(errors.NoRepositoryPresent,
947
branch.Branch.open, abranch_url)
948
self.make_branch('baz')
949
for actual_bzrdir in bzrdir.BzrDir.find_branches(t):
950
self.assertEndsWith(actual_bzrdir.user_url, '/baz/')
953
class TestMeta1DirFormat(TestCaseWithTransport):
954
"""Tests specific to the meta1 dir format."""
956
def test_right_base_dirs(self):
957
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
959
branch_base = t.clone('branch').base
960
self.assertEqual(branch_base, dir.get_branch_transport(None).base)
961
self.assertEqual(branch_base,
962
dir.get_branch_transport(bzrlib.branch.BzrBranchFormat5()).base)
963
repository_base = t.clone('repository').base
964
self.assertEqual(repository_base, dir.get_repository_transport(None).base)
965
repository_format = repository.format_registry.get_default()
966
self.assertEqual(repository_base,
967
dir.get_repository_transport(repository_format).base)
968
checkout_base = t.clone('checkout').base
969
self.assertEqual(checkout_base, dir.get_workingtree_transport(None).base)
970
self.assertEqual(checkout_base,
971
dir.get_workingtree_transport(workingtree_3.WorkingTreeFormat3()).base)
973
def test_meta1dir_uses_lockdir(self):
974
"""Meta1 format uses a LockDir to guard the whole directory, not a file."""
975
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
977
self.assertIsDirectory('branch-lock', t)
979
def test_comparison(self):
980
"""Equality and inequality behave properly.
982
Metadirs should compare equal iff they have the same repo, branch and
985
mydir = bzrdir.format_registry.make_bzrdir('knit')
986
self.assertEqual(mydir, mydir)
987
self.assertFalse(mydir != mydir)
988
otherdir = bzrdir.format_registry.make_bzrdir('knit')
989
self.assertEqual(otherdir, mydir)
990
self.assertFalse(otherdir != mydir)
991
otherdir2 = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
992
self.assertNotEqual(otherdir2, mydir)
993
self.assertFalse(otherdir2 == mydir)
995
def test_needs_conversion_different_working_tree(self):
996
# meta1dirs need an conversion if any element is not the default.
997
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
998
tree = self.make_branch_and_tree('tree', format='knit')
999
self.assertTrue(tree.bzrdir.needs_format_conversion(
1002
def test_initialize_on_format_uses_smart_transport(self):
1003
self.setup_smart_server_with_call_log()
1004
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
1005
transport = self.get_transport('target')
1006
transport.ensure_base()
1007
self.reset_smart_call_log()
1008
instance = new_format.initialize_on_transport(transport)
1009
self.assertIsInstance(instance, remote.RemoteBzrDir)
1010
rpc_count = len(self.hpss_calls)
1011
# This figure represent the amount of work to perform this use case. It
1012
# is entirely ok to reduce this number if a test fails due to rpc_count
1013
# being too low. If rpc_count increases, more network roundtrips have
1014
# become necessary for this use case. Please do not adjust this number
1015
# upwards without agreement from bzr's network support maintainers.
1016
self.assertEqual(2, rpc_count)
1019
class NonLocalTests(TestCaseWithTransport):
1020
"""Tests for bzrdir static behaviour on non local paths."""
1023
super(NonLocalTests, self).setUp()
1024
self.vfs_transport_factory = memory.MemoryServer
1026
def test_create_branch_convenience(self):
1027
# outside a repo the default convenience output is a repo+branch_tree
1028
format = bzrdir.format_registry.make_bzrdir('knit')
1029
branch = bzrdir.BzrDir.create_branch_convenience(
1030
self.get_url('foo'), format=format)
1031
self.assertRaises(errors.NoWorkingTree,
1032
branch.bzrdir.open_workingtree)
1033
branch.bzrdir.open_repository()
1035
def test_create_branch_convenience_force_tree_not_local_fails(self):
1036
# outside a repo the default convenience output is a repo+branch_tree
1037
format = bzrdir.format_registry.make_bzrdir('knit')
1038
self.assertRaises(errors.NotLocalUrl,
1039
bzrdir.BzrDir.create_branch_convenience,
1040
self.get_url('foo'),
1041
force_new_tree=True,
1043
t = self.get_transport()
1044
self.assertFalse(t.has('foo'))
1046
def test_clone(self):
1047
# clone into a nonlocal path works
1048
format = bzrdir.format_registry.make_bzrdir('knit')
1049
branch = bzrdir.BzrDir.create_branch_convenience('local',
1051
branch.bzrdir.open_workingtree()
1052
result = branch.bzrdir.clone(self.get_url('remote'))
1053
self.assertRaises(errors.NoWorkingTree,
1054
result.open_workingtree)
1055
result.open_branch()
1056
result.open_repository()
1058
def test_checkout_metadir(self):
1059
# checkout_metadir has reasonable working tree format even when no
1060
# working tree is present
1061
self.make_branch('branch-knit2', format='dirstate-with-subtree')
1062
my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
1063
checkout_format = my_bzrdir.checkout_metadir()
1064
self.assertIsInstance(checkout_format.workingtree_format,
1065
workingtree_4.WorkingTreeFormat4)
1068
class TestHTTPRedirections(object):
1069
"""Test redirection between two http servers.
1071
This MUST be used by daughter classes that also inherit from
1072
TestCaseWithTwoWebservers.
1074
We can't inherit directly from TestCaseWithTwoWebservers or the
1075
test framework will try to create an instance which cannot
1076
run, its implementation being incomplete.
1079
def create_transport_readonly_server(self):
1080
# We don't set the http protocol version, relying on the default
1081
return http_utils.HTTPServerRedirecting()
1083
def create_transport_secondary_server(self):
1084
# We don't set the http protocol version, relying on the default
1085
return http_utils.HTTPServerRedirecting()
1088
super(TestHTTPRedirections, self).setUp()
1089
# The redirections will point to the new server
1090
self.new_server = self.get_readonly_server()
1091
# The requests to the old server will be redirected
1092
self.old_server = self.get_secondary_server()
1093
# Configure the redirections
1094
self.old_server.redirect_to(self.new_server.host, self.new_server.port)
1096
def test_loop(self):
1097
# Both servers redirect to each other creating a loop
1098
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
1099
# Starting from either server should loop
1100
old_url = self._qualified_url(self.old_server.host,
1101
self.old_server.port)
1102
oldt = self._transport(old_url)
1103
self.assertRaises(errors.NotBranchError,
1104
bzrdir.BzrDir.open_from_transport, oldt)
1105
new_url = self._qualified_url(self.new_server.host,
1106
self.new_server.port)
1107
newt = self._transport(new_url)
1108
self.assertRaises(errors.NotBranchError,
1109
bzrdir.BzrDir.open_from_transport, newt)
1111
def test_qualifier_preserved(self):
1112
wt = self.make_branch_and_tree('branch')
1113
old_url = self._qualified_url(self.old_server.host,
1114
self.old_server.port)
1115
start = self._transport(old_url).clone('branch')
1116
bdir = bzrdir.BzrDir.open_from_transport(start)
1117
# Redirection should preserve the qualifier, hence the transport class
1119
self.assertIsInstance(bdir.root_transport, type(start))
1122
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1123
http_utils.TestCaseWithTwoWebservers):
1124
"""Tests redirections for urllib implementation"""
1126
_transport = HttpTransport_urllib
1128
def _qualified_url(self, host, port):
1129
result = 'http+urllib://%s:%s' % (host, port)
1130
self.permit_url(result)
1135
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1136
TestHTTPRedirections,
1137
http_utils.TestCaseWithTwoWebservers):
1138
"""Tests redirections for pycurl implementation"""
1140
def _qualified_url(self, host, port):
1141
result = 'http+pycurl://%s:%s' % (host, port)
1142
self.permit_url(result)
1146
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1147
http_utils.TestCaseWithTwoWebservers):
1148
"""Tests redirections for the nosmart decorator"""
1150
_transport = NoSmartTransportDecorator
1152
def _qualified_url(self, host, port):
1153
result = 'nosmart+http://%s:%s' % (host, port)
1154
self.permit_url(result)
1158
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1159
http_utils.TestCaseWithTwoWebservers):
1160
"""Tests redirections for readonly decoratror"""
1162
_transport = ReadonlyTransportDecorator
1164
def _qualified_url(self, host, port):
1165
result = 'readonly+http://%s:%s' % (host, port)
1166
self.permit_url(result)
1170
class TestDotBzrHidden(TestCaseWithTransport):
1173
if sys.platform == 'win32':
1174
ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
1177
f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
1178
stderr=subprocess.PIPE)
1179
out, err = f.communicate()
1180
self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
1182
return out.splitlines()
1184
def test_dot_bzr_hidden(self):
1185
if sys.platform == 'win32' and not win32utils.has_win32file:
1186
raise TestSkipped('unable to make file hidden without pywin32 library')
1187
b = bzrdir.BzrDir.create('.')
1188
self.build_tree(['a'])
1189
self.assertEquals(['a'], self.get_ls())
1191
def test_dot_bzr_hidden_with_url(self):
1192
if sys.platform == 'win32' and not win32utils.has_win32file:
1193
raise TestSkipped('unable to make file hidden without pywin32 library')
1194
b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
1195
self.build_tree(['a'])
1196
self.assertEquals(['a'], self.get_ls())
1199
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
1200
"""Test BzrDirFormat implementation for TestBzrDirSprout."""
1202
def _open(self, transport):
1203
return _TestBzrDir(transport, self)
1206
class _TestBzrDir(bzrdir.BzrDirMeta1):
1207
"""Test BzrDir implementation for TestBzrDirSprout.
1209
When created a _TestBzrDir already has repository and a branch. The branch
1210
is a test double as well.
1213
def __init__(self, *args, **kwargs):
1214
super(_TestBzrDir, self).__init__(*args, **kwargs)
1215
self.test_branch = _TestBranch(self.transport)
1216
self.test_branch.repository = self.create_repository()
1218
def open_branch(self, unsupported=False):
1219
return self.test_branch
1221
def cloning_metadir(self, require_stacking=False):
1222
return _TestBzrDirFormat()
1225
class _TestBranchFormat(bzrlib.branch.BranchFormat):
1226
"""Test Branch format for TestBzrDirSprout."""
1229
class _TestBranch(bzrlib.branch.Branch):
1230
"""Test Branch implementation for TestBzrDirSprout."""
1232
def __init__(self, transport, *args, **kwargs):
1233
self._format = _TestBranchFormat()
1234
self._transport = transport
1235
self.base = transport.base
1236
super(_TestBranch, self).__init__(*args, **kwargs)
1240
def sprout(self, *args, **kwargs):
1241
self.calls.append('sprout')
1242
return _TestBranch(self._transport)
1244
def copy_content_into(self, destination, revision_id=None):
1245
self.calls.append('copy_content_into')
1247
def last_revision(self):
1248
return _mod_revision.NULL_REVISION
1250
def get_parent(self):
1253
def _get_config(self):
1254
return config.TransportConfig(self._transport, 'branch.conf')
1256
def set_parent(self, parent):
1257
self._parent = parent
1259
def lock_read(self):
1260
return lock.LogicalLockResult(self.unlock)
1266
class TestBzrDirSprout(TestCaseWithMemoryTransport):
1268
def test_sprout_uses_branch_sprout(self):
1269
"""BzrDir.sprout calls Branch.sprout.
1271
Usually, BzrDir.sprout should delegate to the branch's sprout method
1272
for part of the work. This allows the source branch to control the
1273
choice of format for the new branch.
1275
There are exceptions, but this tests avoids them:
1276
- if there's no branch in the source bzrdir,
1277
- or if the stacking has been requested and the format needs to be
1278
overridden to satisfy that.
1280
# Make an instrumented bzrdir.
1281
t = self.get_transport('source')
1283
source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
1284
# The instrumented bzrdir has a test_branch attribute that logs calls
1285
# made to the branch contained in that bzrdir. Initially the test
1286
# branch exists but no calls have been made to it.
1287
self.assertEqual([], source_bzrdir.test_branch.calls)
1290
target_url = self.get_url('target')
1291
result = source_bzrdir.sprout(target_url, recurse='no')
1293
# The bzrdir called the branch's sprout method.
1294
self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
1296
def test_sprout_parent(self):
1297
grandparent_tree = self.make_branch('grandparent')
1298
parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
1299
branch_tree = parent.bzrdir.sprout('branch').open_branch()
1300
self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
1303
class TestBzrDirHooks(TestCaseWithMemoryTransport):
1305
def test_pre_open_called(self):
1307
bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
1308
transport = self.get_transport('foo')
1309
url = transport.base
1310
self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
1311
self.assertEqual([transport.base], [t.base for t in calls])
1313
def test_pre_open_actual_exceptions_raised(self):
1315
def fail_once(transport):
1318
raise errors.BzrError("fail")
1319
bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
1320
transport = self.get_transport('foo')
1321
url = transport.base
1322
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1323
self.assertEqual('fail', err._preformatted_string)
1325
def test_post_repo_init(self):
1326
from bzrlib.bzrdir import RepoInitHookParams
1328
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1330
self.make_repository('foo')
1331
self.assertLength(1, calls)
1333
self.assertIsInstance(params, RepoInitHookParams)
1334
self.assertTrue(hasattr(params, 'bzrdir'))
1335
self.assertTrue(hasattr(params, 'repository'))
1337
def test_post_repo_init_hook_repr(self):
1339
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1340
lambda params: param_reprs.append(repr(params)), None)
1341
self.make_repository('foo')
1342
self.assertLength(1, param_reprs)
1343
param_repr = param_reprs[0]
1344
self.assertStartsWith(param_repr, '<RepoInitHookParams for ')
1347
class TestGenerateBackupName(TestCaseWithMemoryTransport):
1348
# FIXME: This may need to be unified with test_osutils.TestBackupNames or
1349
# moved to per_bzrdir or per_transport for better coverage ?
1353
super(TestGenerateBackupName, self).setUp()
1354
self._transport = self.get_transport()
1355
bzrdir.BzrDir.create(self.get_url(),
1356
possible_transports=[self._transport])
1357
self._bzrdir = bzrdir.BzrDir.open_from_transport(self._transport)
1359
def test_deprecated_generate_backup_name(self):
1360
res = self.applyDeprecated(
1361
symbol_versioning.deprecated_in((2, 3, 0)),
1362
self._bzrdir.generate_backup_name, 'whatever')
1365
self.assertEqual("a.~1~", self._bzrdir._available_backup_name("a"))
1367
def test_exiting(self):
1368
self._transport.put_bytes("a.~1~", "some content")
1369
self.assertEqual("a.~2~", self._bzrdir._available_backup_name("a"))
added
removed
bzrlib/tests/test_bzrdir.py
