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
def from_string(cls, format_string):
261
class BzrDirFormatTest1(bzrdir.BzrDirMetaFormat1):
264
def get_format_string():
265
return "Test format 1"
268
class BzrDirFormatTest2(bzrdir.BzrDirMetaFormat1):
271
def get_format_string():
272
return "Test format 2"
275
class TestBzrDirFormat(TestCaseWithTransport):
276
"""Tests for the BzrDirFormat facility."""
278
def test_find_format(self):
279
# is the right format object found for a branch?
280
# create a branch with a few known format objects.
281
bzrdir.BzrProber.formats.register(BzrDirFormatTest1.get_format_string(),
283
self.addCleanup(bzrdir.BzrProber.formats.remove,
284
BzrDirFormatTest1.get_format_string())
285
bzrdir.BzrProber.formats.register(BzrDirFormatTest2.get_format_string(),
287
self.addCleanup(bzrdir.BzrProber.formats.remove,
288
BzrDirFormatTest2.get_format_string())
289
t = self.get_transport()
290
self.build_tree(["foo/", "bar/"], transport=t)
291
def check_format(format, url):
292
format.initialize(url)
293
t = _mod_transport.get_transport_from_path(url)
294
found_format = bzrdir.BzrDirFormat.find_format(t)
295
self.assertIsInstance(found_format, format.__class__)
296
check_format(BzrDirFormatTest1(), "foo")
297
check_format(BzrDirFormatTest2(), "bar")
299
def test_find_format_nothing_there(self):
300
self.assertRaises(NotBranchError,
301
bzrdir.BzrDirFormat.find_format,
302
_mod_transport.get_transport_from_path('.'))
304
def test_find_format_unknown_format(self):
305
t = self.get_transport()
307
t.put_bytes('.bzr/branch-format', '')
308
self.assertRaises(UnknownFormatError,
309
bzrdir.BzrDirFormat.find_format,
310
_mod_transport.get_transport_from_path('.'))
312
def test_register_unregister_format(self):
313
format = SampleBzrDirFormat()
316
format.initialize(url)
317
# register a format for it.
318
bzrdir.BzrProber.formats.register(format.get_format_string(), format)
319
# which bzrdir.Open will refuse (not supported)
320
self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open, url)
321
# which bzrdir.open_containing will refuse (not supported)
322
self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open_containing, url)
323
# but open_downlevel will work
324
t = _mod_transport.get_transport_from_url(url)
325
self.assertEqual(format.open(t), bzrdir.BzrDir.open_unsupported(url))
326
# unregister the format
327
bzrdir.BzrProber.formats.remove(format.get_format_string())
328
# now open_downlevel should fail too.
329
self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
331
def test_create_branch_and_repo_uses_default(self):
332
format = SampleBzrDirFormat()
333
branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(),
335
self.assertTrue(isinstance(branch, SampleBranch))
337
def test_create_branch_and_repo_under_shared(self):
338
# creating a branch and repo in a shared repo uses the
340
format = bzrdir.format_registry.make_bzrdir('knit')
341
self.make_repository('.', shared=True, format=format)
342
branch = bzrdir.BzrDir.create_branch_and_repo(
343
self.get_url('child'), format=format)
344
self.assertRaises(errors.NoRepositoryPresent,
345
branch.bzrdir.open_repository)
347
def test_create_branch_and_repo_under_shared_force_new(self):
348
# creating a branch and repo in a shared repo can be forced to
350
format = bzrdir.format_registry.make_bzrdir('knit')
351
self.make_repository('.', shared=True, format=format)
352
branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
355
branch.bzrdir.open_repository()
357
def test_create_standalone_working_tree(self):
358
format = SampleBzrDirFormat()
359
# note this is deliberately readonly, as this failure should
360
# occur before any writes.
361
self.assertRaises(errors.NotLocalUrl,
362
bzrdir.BzrDir.create_standalone_workingtree,
363
self.get_readonly_url(), format=format)
364
tree = bzrdir.BzrDir.create_standalone_workingtree('.',
366
self.assertEqual('A tree', tree)
368
def test_create_standalone_working_tree_under_shared_repo(self):
369
# create standalone working tree always makes a repo.
370
format = bzrdir.format_registry.make_bzrdir('knit')
371
self.make_repository('.', shared=True, format=format)
372
# note this is deliberately readonly, as this failure should
373
# occur before any writes.
374
self.assertRaises(errors.NotLocalUrl,
375
bzrdir.BzrDir.create_standalone_workingtree,
376
self.get_readonly_url('child'), format=format)
377
tree = bzrdir.BzrDir.create_standalone_workingtree('child',
379
tree.bzrdir.open_repository()
381
def test_create_branch_convenience(self):
382
# outside a repo the default convenience output is a repo+branch_tree
383
format = bzrdir.format_registry.make_bzrdir('knit')
384
branch = bzrdir.BzrDir.create_branch_convenience('.', format=format)
385
branch.bzrdir.open_workingtree()
386
branch.bzrdir.open_repository()
388
def test_create_branch_convenience_possible_transports(self):
389
"""Check that the optional 'possible_transports' is recognized"""
390
format = bzrdir.format_registry.make_bzrdir('knit')
391
t = self.get_transport()
392
branch = bzrdir.BzrDir.create_branch_convenience(
393
'.', format=format, possible_transports=[t])
394
branch.bzrdir.open_workingtree()
395
branch.bzrdir.open_repository()
397
def test_create_branch_convenience_root(self):
398
"""Creating a branch at the root of a fs should work."""
399
self.vfs_transport_factory = memory.MemoryServer
400
# outside a repo the default convenience output is a repo+branch_tree
401
format = bzrdir.format_registry.make_bzrdir('knit')
402
branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
404
self.assertRaises(errors.NoWorkingTree,
405
branch.bzrdir.open_workingtree)
406
branch.bzrdir.open_repository()
408
def test_create_branch_convenience_under_shared_repo(self):
409
# inside a repo the default convenience output is a branch+ follow the
411
format = bzrdir.format_registry.make_bzrdir('knit')
412
self.make_repository('.', shared=True, format=format)
413
branch = bzrdir.BzrDir.create_branch_convenience('child',
415
branch.bzrdir.open_workingtree()
416
self.assertRaises(errors.NoRepositoryPresent,
417
branch.bzrdir.open_repository)
419
def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
420
# inside a repo the default convenience output is a branch+ follow the
421
# repo tree policy but we can override that
422
format = bzrdir.format_registry.make_bzrdir('knit')
423
self.make_repository('.', shared=True, format=format)
424
branch = bzrdir.BzrDir.create_branch_convenience('child',
425
force_new_tree=False, format=format)
426
self.assertRaises(errors.NoWorkingTree,
427
branch.bzrdir.open_workingtree)
428
self.assertRaises(errors.NoRepositoryPresent,
429
branch.bzrdir.open_repository)
431
def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
432
# inside a repo the default convenience output is a branch+ follow the
434
format = bzrdir.format_registry.make_bzrdir('knit')
435
repo = self.make_repository('.', shared=True, format=format)
436
repo.set_make_working_trees(False)
437
branch = bzrdir.BzrDir.create_branch_convenience('child',
439
self.assertRaises(errors.NoWorkingTree,
440
branch.bzrdir.open_workingtree)
441
self.assertRaises(errors.NoRepositoryPresent,
442
branch.bzrdir.open_repository)
444
def test_create_branch_convenience_under_shared_repo_no_tree_policy_force_tree(self):
445
# inside a repo the default convenience output is a branch+ follow the
446
# repo tree policy but we can override that
447
format = bzrdir.format_registry.make_bzrdir('knit')
448
repo = self.make_repository('.', shared=True, format=format)
449
repo.set_make_working_trees(False)
450
branch = bzrdir.BzrDir.create_branch_convenience('child',
451
force_new_tree=True, format=format)
452
branch.bzrdir.open_workingtree()
453
self.assertRaises(errors.NoRepositoryPresent,
454
branch.bzrdir.open_repository)
456
def test_create_branch_convenience_under_shared_repo_force_new_repo(self):
457
# inside a repo the default convenience output is overridable to give
459
format = bzrdir.format_registry.make_bzrdir('knit')
460
self.make_repository('.', shared=True, format=format)
461
branch = bzrdir.BzrDir.create_branch_convenience('child',
462
force_new_repo=True, format=format)
463
branch.bzrdir.open_repository()
464
branch.bzrdir.open_workingtree()
467
class TestRepositoryAcquisitionPolicy(TestCaseWithTransport):
469
def test_acquire_repository_standalone(self):
470
"""The default acquisition policy should create a standalone branch."""
471
my_bzrdir = self.make_bzrdir('.')
472
repo_policy = my_bzrdir.determine_repository_policy()
473
repo, is_new = repo_policy.acquire_repository()
474
self.assertEqual(repo.bzrdir.root_transport.base,
475
my_bzrdir.root_transport.base)
476
self.assertFalse(repo.is_shared())
478
def test_determine_stacking_policy(self):
479
parent_bzrdir = self.make_bzrdir('.')
480
child_bzrdir = self.make_bzrdir('child')
481
parent_bzrdir.get_config().set_default_stack_on('http://example.org')
482
repo_policy = child_bzrdir.determine_repository_policy()
483
self.assertEqual('http://example.org', repo_policy._stack_on)
485
def test_determine_stacking_policy_relative(self):
486
parent_bzrdir = self.make_bzrdir('.')
487
child_bzrdir = self.make_bzrdir('child')
488
parent_bzrdir.get_config().set_default_stack_on('child2')
489
repo_policy = child_bzrdir.determine_repository_policy()
490
self.assertEqual('child2', repo_policy._stack_on)
491
self.assertEqual(parent_bzrdir.root_transport.base,
492
repo_policy._stack_on_pwd)
494
def prepare_default_stacking(self, child_format='1.6'):
495
parent_bzrdir = self.make_bzrdir('.')
496
child_branch = self.make_branch('child', format=child_format)
497
parent_bzrdir.get_config().set_default_stack_on(child_branch.base)
498
new_child_transport = parent_bzrdir.transport.clone('child2')
499
return child_branch, new_child_transport
501
def test_clone_on_transport_obeys_stacking_policy(self):
502
child_branch, new_child_transport = self.prepare_default_stacking()
503
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
504
self.assertEqual(child_branch.base,
505
new_child.open_branch().get_stacked_on_url())
507
def test_default_stacking_with_stackable_branch_unstackable_repo(self):
508
# Make stackable source branch with an unstackable repo format.
509
source_bzrdir = self.make_bzrdir('source')
510
knitpack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
511
source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
513
# Make a directory with a default stacking policy
514
parent_bzrdir = self.make_bzrdir('parent')
515
stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
516
parent_bzrdir.get_config().set_default_stack_on(stacked_on.base)
517
# Clone source into directory
518
target = source_bzrdir.clone(self.get_url('parent/target'))
520
def test_format_initialize_on_transport_ex_stacked_on(self):
521
# trunk is a stackable format. Note that its in the same server area
522
# which is what launchpad does, but not sufficient to exercise the
524
trunk = self.make_branch('trunk', format='1.9')
525
t = self.get_transport('stacked')
526
old_fmt = bzrdir.format_registry.make_bzrdir('pack-0.92')
527
repo_name = old_fmt.repository_format.network_name()
528
# Should end up with a 1.9 format (stackable)
529
repo, control, require_stacking, repo_policy = \
530
old_fmt.initialize_on_transport_ex(t,
531
repo_format_name=repo_name, stacked_on='../trunk',
534
# Repositories are open write-locked
535
self.assertTrue(repo.is_write_locked())
536
self.addCleanup(repo.unlock)
538
repo = control.open_repository()
539
self.assertIsInstance(control, bzrdir.BzrDir)
540
opened = bzrdir.BzrDir.open(t.base)
541
if not isinstance(old_fmt, remote.RemoteBzrDirFormat):
542
self.assertEqual(control._format.network_name(),
543
old_fmt.network_name())
544
self.assertEqual(control._format.network_name(),
545
opened._format.network_name())
546
self.assertEqual(control.__class__, opened.__class__)
547
self.assertLength(1, repo._fallback_repositories)
549
def test_sprout_obeys_stacking_policy(self):
550
child_branch, new_child_transport = self.prepare_default_stacking()
551
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
552
self.assertEqual(child_branch.base,
553
new_child.open_branch().get_stacked_on_url())
555
def test_clone_ignores_policy_for_unsupported_formats(self):
556
child_branch, new_child_transport = self.prepare_default_stacking(
557
child_format='pack-0.92')
558
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
559
self.assertRaises(errors.UnstackableBranchFormat,
560
new_child.open_branch().get_stacked_on_url)
562
def test_sprout_ignores_policy_for_unsupported_formats(self):
563
child_branch, new_child_transport = self.prepare_default_stacking(
564
child_format='pack-0.92')
565
new_child = child_branch.bzrdir.sprout(new_child_transport.base)
566
self.assertRaises(errors.UnstackableBranchFormat,
567
new_child.open_branch().get_stacked_on_url)
569
def test_sprout_upgrades_format_if_stacked_specified(self):
570
child_branch, new_child_transport = self.prepare_default_stacking(
571
child_format='pack-0.92')
572
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
574
self.assertEqual(child_branch.bzrdir.root_transport.base,
575
new_child.open_branch().get_stacked_on_url())
576
repo = new_child.open_repository()
577
self.assertTrue(repo._format.supports_external_lookups)
578
self.assertFalse(repo.supports_rich_root())
580
def test_clone_on_transport_upgrades_format_if_stacked_on_specified(self):
581
child_branch, new_child_transport = self.prepare_default_stacking(
582
child_format='pack-0.92')
583
new_child = child_branch.bzrdir.clone_on_transport(new_child_transport,
584
stacked_on=child_branch.bzrdir.root_transport.base)
585
self.assertEqual(child_branch.bzrdir.root_transport.base,
586
new_child.open_branch().get_stacked_on_url())
587
repo = new_child.open_repository()
588
self.assertTrue(repo._format.supports_external_lookups)
589
self.assertFalse(repo.supports_rich_root())
591
def test_sprout_upgrades_to_rich_root_format_if_needed(self):
592
child_branch, new_child_transport = self.prepare_default_stacking(
593
child_format='rich-root-pack')
594
new_child = child_branch.bzrdir.sprout(new_child_transport.base,
596
repo = new_child.open_repository()
597
self.assertTrue(repo._format.supports_external_lookups)
598
self.assertTrue(repo.supports_rich_root())
600
def test_add_fallback_repo_handles_absolute_urls(self):
601
stack_on = self.make_branch('stack_on', format='1.6')
602
repo = self.make_repository('repo', format='1.6')
603
policy = bzrdir.UseExistingRepository(repo, stack_on.base)
604
policy._add_fallback(repo)
606
def test_add_fallback_repo_handles_relative_urls(self):
607
stack_on = self.make_branch('stack_on', format='1.6')
608
repo = self.make_repository('repo', format='1.6')
609
policy = bzrdir.UseExistingRepository(repo, '.', stack_on.base)
610
policy._add_fallback(repo)
612
def test_configure_relative_branch_stacking_url(self):
613
stack_on = self.make_branch('stack_on', format='1.6')
614
stacked = self.make_branch('stack_on/stacked', format='1.6')
615
policy = bzrdir.UseExistingRepository(stacked.repository,
617
policy.configure_branch(stacked)
618
self.assertEqual('..', stacked.get_stacked_on_url())
620
def test_relative_branch_stacking_to_absolute(self):
621
stack_on = self.make_branch('stack_on', format='1.6')
622
stacked = self.make_branch('stack_on/stacked', format='1.6')
623
policy = bzrdir.UseExistingRepository(stacked.repository,
624
'.', self.get_readonly_url('stack_on'))
625
policy.configure_branch(stacked)
626
self.assertEqual(self.get_readonly_url('stack_on'),
627
stacked.get_stacked_on_url())
630
class ChrootedTests(TestCaseWithTransport):
631
"""A support class that provides readonly urls outside the local namespace.
633
This is done by checking if self.transport_server is a MemoryServer. if it
634
is then we are chrooted already, if it is not then an HttpServer is used
639
super(ChrootedTests, self).setUp()
640
if not self.vfs_transport_factory == memory.MemoryServer:
641
self.transport_readonly_server = http_server.HttpServer
643
def local_branch_path(self, branch):
644
return os.path.realpath(urlutils.local_path_from_url(branch.base))
646
def test_open_containing(self):
647
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
648
self.get_readonly_url(''))
649
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
650
self.get_readonly_url('g/p/q'))
651
control = bzrdir.BzrDir.create(self.get_url())
652
branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url(''))
653
self.assertEqual('', relpath)
654
branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
655
self.assertEqual('g/p/q', relpath)
657
def test_open_containing_tree_branch_or_repository_empty(self):
658
self.assertRaises(errors.NotBranchError,
659
bzrdir.BzrDir.open_containing_tree_branch_or_repository,
660
self.get_readonly_url(''))
662
def test_open_containing_tree_branch_or_repository_all(self):
663
self.make_branch_and_tree('topdir')
664
tree, branch, repo, relpath = \
665
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
667
self.assertEqual(os.path.realpath('topdir'),
668
os.path.realpath(tree.basedir))
669
self.assertEqual(os.path.realpath('topdir'),
670
self.local_branch_path(branch))
672
osutils.realpath(os.path.join('topdir', '.bzr', 'repository')),
673
repo.bzrdir.transport.local_abspath('repository'))
674
self.assertEqual(relpath, 'foo')
676
def test_open_containing_tree_branch_or_repository_no_tree(self):
677
self.make_branch('branch')
678
tree, branch, repo, relpath = \
679
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
681
self.assertEqual(tree, None)
682
self.assertEqual(os.path.realpath('branch'),
683
self.local_branch_path(branch))
685
osutils.realpath(os.path.join('branch', '.bzr', 'repository')),
686
repo.bzrdir.transport.local_abspath('repository'))
687
self.assertEqual(relpath, 'foo')
689
def test_open_containing_tree_branch_or_repository_repo(self):
690
self.make_repository('repo')
691
tree, branch, repo, relpath = \
692
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
694
self.assertEqual(tree, None)
695
self.assertEqual(branch, None)
697
osutils.realpath(os.path.join('repo', '.bzr', 'repository')),
698
repo.bzrdir.transport.local_abspath('repository'))
699
self.assertEqual(relpath, '')
701
def test_open_containing_tree_branch_or_repository_shared_repo(self):
702
self.make_repository('shared', shared=True)
703
bzrdir.BzrDir.create_branch_convenience('shared/branch',
704
force_new_tree=False)
705
tree, branch, repo, relpath = \
706
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
708
self.assertEqual(tree, None)
709
self.assertEqual(os.path.realpath('shared/branch'),
710
self.local_branch_path(branch))
712
osutils.realpath(os.path.join('shared', '.bzr', 'repository')),
713
repo.bzrdir.transport.local_abspath('repository'))
714
self.assertEqual(relpath, '')
716
def test_open_containing_tree_branch_or_repository_branch_subdir(self):
717
self.make_branch_and_tree('foo')
718
self.build_tree(['foo/bar/'])
719
tree, branch, repo, relpath = \
720
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
722
self.assertEqual(os.path.realpath('foo'),
723
os.path.realpath(tree.basedir))
724
self.assertEqual(os.path.realpath('foo'),
725
self.local_branch_path(branch))
727
osutils.realpath(os.path.join('foo', '.bzr', 'repository')),
728
repo.bzrdir.transport.local_abspath('repository'))
729
self.assertEqual(relpath, 'bar')
731
def test_open_containing_tree_branch_or_repository_repo_subdir(self):
732
self.make_repository('bar')
733
self.build_tree(['bar/baz/'])
734
tree, branch, repo, relpath = \
735
bzrdir.BzrDir.open_containing_tree_branch_or_repository(
737
self.assertEqual(tree, None)
738
self.assertEqual(branch, None)
740
osutils.realpath(os.path.join('bar', '.bzr', 'repository')),
741
repo.bzrdir.transport.local_abspath('repository'))
742
self.assertEqual(relpath, 'baz')
744
def test_open_containing_from_transport(self):
745
self.assertRaises(NotBranchError,
746
bzrdir.BzrDir.open_containing_from_transport,
747
_mod_transport.get_transport_from_url(self.get_readonly_url('')))
748
self.assertRaises(NotBranchError,
749
bzrdir.BzrDir.open_containing_from_transport,
750
_mod_transport.get_transport_from_url(
751
self.get_readonly_url('g/p/q')))
752
control = bzrdir.BzrDir.create(self.get_url())
753
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
754
_mod_transport.get_transport_from_url(
755
self.get_readonly_url('')))
756
self.assertEqual('', relpath)
757
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
758
_mod_transport.get_transport_from_url(
759
self.get_readonly_url('g/p/q')))
760
self.assertEqual('g/p/q', relpath)
762
def test_open_containing_tree_or_branch(self):
763
self.make_branch_and_tree('topdir')
764
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
766
self.assertEqual(os.path.realpath('topdir'),
767
os.path.realpath(tree.basedir))
768
self.assertEqual(os.path.realpath('topdir'),
769
self.local_branch_path(branch))
770
self.assertIs(tree.bzrdir, branch.bzrdir)
771
self.assertEqual('foo', relpath)
772
# opening from non-local should not return the tree
773
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
774
self.get_readonly_url('topdir/foo'))
775
self.assertEqual(None, tree)
776
self.assertEqual('foo', relpath)
778
self.make_branch('topdir/foo')
779
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
781
self.assertIs(tree, None)
782
self.assertEqual(os.path.realpath('topdir/foo'),
783
self.local_branch_path(branch))
784
self.assertEqual('', relpath)
786
def test_open_tree_or_branch(self):
787
self.make_branch_and_tree('topdir')
788
tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir')
789
self.assertEqual(os.path.realpath('topdir'),
790
os.path.realpath(tree.basedir))
791
self.assertEqual(os.path.realpath('topdir'),
792
self.local_branch_path(branch))
793
self.assertIs(tree.bzrdir, branch.bzrdir)
794
# opening from non-local should not return the tree
795
tree, branch = bzrdir.BzrDir.open_tree_or_branch(
796
self.get_readonly_url('topdir'))
797
self.assertEqual(None, tree)
799
self.make_branch('topdir/foo')
800
tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir/foo')
801
self.assertIs(tree, None)
802
self.assertEqual(os.path.realpath('topdir/foo'),
803
self.local_branch_path(branch))
805
def test_open_from_transport(self):
806
# transport pointing at bzrdir should give a bzrdir with root transport
807
# set to the given transport
808
control = bzrdir.BzrDir.create(self.get_url())
809
t = self.get_transport()
810
opened_bzrdir = bzrdir.BzrDir.open_from_transport(t)
811
self.assertEqual(t.base, opened_bzrdir.root_transport.base)
812
self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
814
def test_open_from_transport_no_bzrdir(self):
815
t = self.get_transport()
816
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
818
def test_open_from_transport_bzrdir_in_parent(self):
819
control = bzrdir.BzrDir.create(self.get_url())
820
t = self.get_transport()
822
t = t.clone('subdir')
823
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
825
def test_sprout_recursive(self):
826
tree = self.make_branch_and_tree('tree1',
827
format='dirstate-with-subtree')
828
sub_tree = self.make_branch_and_tree('tree1/subtree',
829
format='dirstate-with-subtree')
830
sub_tree.set_root_id('subtree-root')
831
tree.add_reference(sub_tree)
832
self.build_tree(['tree1/subtree/file'])
834
tree.commit('Initial commit')
835
tree2 = tree.bzrdir.sprout('tree2').open_workingtree()
837
self.addCleanup(tree2.unlock)
838
self.assertPathExists('tree2/subtree/file')
839
self.assertEqual('tree-reference', tree2.kind('subtree-root'))
841
def test_cloning_metadir(self):
842
"""Ensure that cloning metadir is suitable"""
843
bzrdir = self.make_bzrdir('bzrdir')
844
bzrdir.cloning_metadir()
845
branch = self.make_branch('branch', format='knit')
846
format = branch.bzrdir.cloning_metadir()
847
self.assertIsInstance(format.workingtree_format,
848
workingtree_4.WorkingTreeFormat6)
850
def test_sprout_recursive_treeless(self):
851
tree = self.make_branch_and_tree('tree1',
852
format='dirstate-with-subtree')
853
sub_tree = self.make_branch_and_tree('tree1/subtree',
854
format='dirstate-with-subtree')
855
tree.add_reference(sub_tree)
856
self.build_tree(['tree1/subtree/file'])
858
tree.commit('Initial commit')
859
# The following line force the orhaning to reveal bug #634470
860
tree.branch.get_config().set_user_option(
861
'bzr.transform.orphan_policy', 'move')
862
tree.bzrdir.destroy_workingtree()
863
# FIXME: subtree/.bzr is left here which allows the test to pass (or
864
# fail :-( ) -- vila 20100909
865
repo = self.make_repository('repo', shared=True,
866
format='dirstate-with-subtree')
867
repo.set_make_working_trees(False)
868
# FIXME: we just deleted the workingtree and now we want to use it ????
869
# At a minimum, we should use tree.branch below (but this fails too
870
# currently) or stop calling this test 'treeless'. Specifically, I've
871
# turn the line below into an assertRaises when 'subtree/.bzr' is
872
# orphaned and sprout tries to access the branch there (which is left
873
# by bzrdir.BzrDirMeta1.destroy_workingtree when it ignores the
874
# [DeletingParent('Not deleting', u'subtree', None)] conflict). See bug
875
# #634470. -- vila 20100909
876
self.assertRaises(errors.NotBranchError,
877
tree.bzrdir.sprout, 'repo/tree2')
878
# self.assertPathExists('repo/tree2/subtree')
879
# self.assertPathDoesNotExist('repo/tree2/subtree/file')
881
def make_foo_bar_baz(self):
882
foo = bzrdir.BzrDir.create_branch_convenience('foo').bzrdir
883
bar = self.make_branch('foo/bar').bzrdir
884
baz = self.make_branch('baz').bzrdir
887
def test_find_bzrdirs(self):
888
foo, bar, baz = self.make_foo_bar_baz()
889
t = self.get_transport()
890
self.assertEqualBzrdirs([baz, foo, bar], bzrdir.BzrDir.find_bzrdirs(t))
892
def make_fake_permission_denied_transport(self, transport, paths):
893
"""Create a transport that raises PermissionDenied for some paths."""
896
raise errors.PermissionDenied(path)
898
path_filter_server = pathfilter.PathFilteringServer(transport, filter)
899
path_filter_server.start_server()
900
self.addCleanup(path_filter_server.stop_server)
901
path_filter_transport = pathfilter.PathFilteringTransport(
902
path_filter_server, '.')
903
return (path_filter_server, path_filter_transport)
905
def assertBranchUrlsEndWith(self, expect_url_suffix, actual_bzrdirs):
906
"""Check that each branch url ends with the given suffix."""
907
for actual_bzrdir in actual_bzrdirs:
908
self.assertEndsWith(actual_bzrdir.user_url, expect_url_suffix)
910
def test_find_bzrdirs_permission_denied(self):
911
foo, bar, baz = self.make_foo_bar_baz()
912
t = self.get_transport()
913
path_filter_server, path_filter_transport = \
914
self.make_fake_permission_denied_transport(t, ['foo'])
916
self.assertBranchUrlsEndWith('/baz/',
917
bzrdir.BzrDir.find_bzrdirs(path_filter_transport))
919
smart_transport = self.make_smart_server('.',
920
backing_server=path_filter_server)
921
self.assertBranchUrlsEndWith('/baz/',
922
bzrdir.BzrDir.find_bzrdirs(smart_transport))
924
def test_find_bzrdirs_list_current(self):
925
def list_current(transport):
926
return [s for s in transport.list_dir('') if s != 'baz']
928
foo, bar, baz = self.make_foo_bar_baz()
929
t = self.get_transport()
930
self.assertEqualBzrdirs(
932
bzrdir.BzrDir.find_bzrdirs(t, list_current=list_current))
934
def test_find_bzrdirs_evaluate(self):
935
def evaluate(bzrdir):
937
repo = bzrdir.open_repository()
938
except errors.NoRepositoryPresent:
939
return True, bzrdir.root_transport.base
941
return False, bzrdir.root_transport.base
943
foo, bar, baz = self.make_foo_bar_baz()
944
t = self.get_transport()
945
self.assertEqual([baz.root_transport.base, foo.root_transport.base],
946
list(bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate)))
948
def assertEqualBzrdirs(self, first, second):
950
second = list(second)
951
self.assertEqual(len(first), len(second))
952
for x, y in zip(first, second):
953
self.assertEqual(x.root_transport.base, y.root_transport.base)
955
def test_find_branches(self):
956
root = self.make_repository('', shared=True)
957
foo, bar, baz = self.make_foo_bar_baz()
958
qux = self.make_bzrdir('foo/qux')
959
t = self.get_transport()
960
branches = bzrdir.BzrDir.find_branches(t)
961
self.assertEqual(baz.root_transport.base, branches[0].base)
962
self.assertEqual(foo.root_transport.base, branches[1].base)
963
self.assertEqual(bar.root_transport.base, branches[2].base)
965
# ensure this works without a top-level repo
966
branches = bzrdir.BzrDir.find_branches(t.clone('foo'))
967
self.assertEqual(foo.root_transport.base, branches[0].base)
968
self.assertEqual(bar.root_transport.base, branches[1].base)
971
class TestMissingRepoBranchesSkipped(TestCaseWithMemoryTransport):
973
def test_find_bzrdirs_missing_repo(self):
974
t = self.get_transport()
975
arepo = self.make_repository('arepo', shared=True)
976
abranch_url = arepo.user_url + '/abranch'
977
abranch = bzrdir.BzrDir.create(abranch_url).create_branch()
978
t.delete_tree('arepo/.bzr')
979
self.assertRaises(errors.NoRepositoryPresent,
980
branch.Branch.open, abranch_url)
981
self.make_branch('baz')
982
for actual_bzrdir in bzrdir.BzrDir.find_branches(t):
983
self.assertEndsWith(actual_bzrdir.user_url, '/baz/')
986
class TestMeta1DirFormat(TestCaseWithTransport):
987
"""Tests specific to the meta1 dir format."""
989
def test_right_base_dirs(self):
990
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
992
branch_base = t.clone('branch').base
993
self.assertEqual(branch_base, dir.get_branch_transport(None).base)
994
self.assertEqual(branch_base,
995
dir.get_branch_transport(bzrlib.branch.BzrBranchFormat5()).base)
996
repository_base = t.clone('repository').base
997
self.assertEqual(repository_base, dir.get_repository_transport(None).base)
998
repository_format = repository.format_registry.get_default()
999
self.assertEqual(repository_base,
1000
dir.get_repository_transport(repository_format).base)
1001
checkout_base = t.clone('checkout').base
1002
self.assertEqual(checkout_base, dir.get_workingtree_transport(None).base)
1003
self.assertEqual(checkout_base,
1004
dir.get_workingtree_transport(workingtree_3.WorkingTreeFormat3()).base)
1006
def test_meta1dir_uses_lockdir(self):
1007
"""Meta1 format uses a LockDir to guard the whole directory, not a file."""
1008
dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
1010
self.assertIsDirectory('branch-lock', t)
1012
def test_comparison(self):
1013
"""Equality and inequality behave properly.
1015
Metadirs should compare equal iff they have the same repo, branch and
1018
mydir = bzrdir.format_registry.make_bzrdir('knit')
1019
self.assertEqual(mydir, mydir)
1020
self.assertFalse(mydir != mydir)
1021
otherdir = bzrdir.format_registry.make_bzrdir('knit')
1022
self.assertEqual(otherdir, mydir)
1023
self.assertFalse(otherdir != mydir)
1024
otherdir2 = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
1025
self.assertNotEqual(otherdir2, mydir)
1026
self.assertFalse(otherdir2 == mydir)
1028
def test_needs_conversion_different_working_tree(self):
1029
# meta1dirs need an conversion if any element is not the default.
1030
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
1031
tree = self.make_branch_and_tree('tree', format='knit')
1032
self.assertTrue(tree.bzrdir.needs_format_conversion(
1035
def test_initialize_on_format_uses_smart_transport(self):
1036
self.setup_smart_server_with_call_log()
1037
new_format = bzrdir.format_registry.make_bzrdir('dirstate')
1038
transport = self.get_transport('target')
1039
transport.ensure_base()
1040
self.reset_smart_call_log()
1041
instance = new_format.initialize_on_transport(transport)
1042
self.assertIsInstance(instance, remote.RemoteBzrDir)
1043
rpc_count = len(self.hpss_calls)
1044
# This figure represent the amount of work to perform this use case. It
1045
# is entirely ok to reduce this number if a test fails due to rpc_count
1046
# being too low. If rpc_count increases, more network roundtrips have
1047
# become necessary for this use case. Please do not adjust this number
1048
# upwards without agreement from bzr's network support maintainers.
1049
self.assertEqual(2, rpc_count)
1052
class NonLocalTests(TestCaseWithTransport):
1053
"""Tests for bzrdir static behaviour on non local paths."""
1056
super(NonLocalTests, self).setUp()
1057
self.vfs_transport_factory = memory.MemoryServer
1059
def test_create_branch_convenience(self):
1060
# outside a repo the default convenience output is a repo+branch_tree
1061
format = bzrdir.format_registry.make_bzrdir('knit')
1062
branch = bzrdir.BzrDir.create_branch_convenience(
1063
self.get_url('foo'), format=format)
1064
self.assertRaises(errors.NoWorkingTree,
1065
branch.bzrdir.open_workingtree)
1066
branch.bzrdir.open_repository()
1068
def test_create_branch_convenience_force_tree_not_local_fails(self):
1069
# outside a repo the default convenience output is a repo+branch_tree
1070
format = bzrdir.format_registry.make_bzrdir('knit')
1071
self.assertRaises(errors.NotLocalUrl,
1072
bzrdir.BzrDir.create_branch_convenience,
1073
self.get_url('foo'),
1074
force_new_tree=True,
1076
t = self.get_transport()
1077
self.assertFalse(t.has('foo'))
1079
def test_clone(self):
1080
# clone into a nonlocal path works
1081
format = bzrdir.format_registry.make_bzrdir('knit')
1082
branch = bzrdir.BzrDir.create_branch_convenience('local',
1084
branch.bzrdir.open_workingtree()
1085
result = branch.bzrdir.clone(self.get_url('remote'))
1086
self.assertRaises(errors.NoWorkingTree,
1087
result.open_workingtree)
1088
result.open_branch()
1089
result.open_repository()
1091
def test_checkout_metadir(self):
1092
# checkout_metadir has reasonable working tree format even when no
1093
# working tree is present
1094
self.make_branch('branch-knit2', format='dirstate-with-subtree')
1095
my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
1096
checkout_format = my_bzrdir.checkout_metadir()
1097
self.assertIsInstance(checkout_format.workingtree_format,
1098
workingtree_4.WorkingTreeFormat4)
1101
class TestHTTPRedirections(object):
1102
"""Test redirection between two http servers.
1104
This MUST be used by daughter classes that also inherit from
1105
TestCaseWithTwoWebservers.
1107
We can't inherit directly from TestCaseWithTwoWebservers or the
1108
test framework will try to create an instance which cannot
1109
run, its implementation being incomplete.
1112
def create_transport_readonly_server(self):
1113
# We don't set the http protocol version, relying on the default
1114
return http_utils.HTTPServerRedirecting()
1116
def create_transport_secondary_server(self):
1117
# We don't set the http protocol version, relying on the default
1118
return http_utils.HTTPServerRedirecting()
1121
super(TestHTTPRedirections, self).setUp()
1122
# The redirections will point to the new server
1123
self.new_server = self.get_readonly_server()
1124
# The requests to the old server will be redirected
1125
self.old_server = self.get_secondary_server()
1126
# Configure the redirections
1127
self.old_server.redirect_to(self.new_server.host, self.new_server.port)
1129
def test_loop(self):
1130
# Both servers redirect to each other creating a loop
1131
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
1132
# Starting from either server should loop
1133
old_url = self._qualified_url(self.old_server.host,
1134
self.old_server.port)
1135
oldt = self._transport(old_url)
1136
self.assertRaises(errors.NotBranchError,
1137
bzrdir.BzrDir.open_from_transport, oldt)
1138
new_url = self._qualified_url(self.new_server.host,
1139
self.new_server.port)
1140
newt = self._transport(new_url)
1141
self.assertRaises(errors.NotBranchError,
1142
bzrdir.BzrDir.open_from_transport, newt)
1144
def test_qualifier_preserved(self):
1145
wt = self.make_branch_and_tree('branch')
1146
old_url = self._qualified_url(self.old_server.host,
1147
self.old_server.port)
1148
start = self._transport(old_url).clone('branch')
1149
bdir = bzrdir.BzrDir.open_from_transport(start)
1150
# Redirection should preserve the qualifier, hence the transport class
1152
self.assertIsInstance(bdir.root_transport, type(start))
1155
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1156
http_utils.TestCaseWithTwoWebservers):
1157
"""Tests redirections for urllib implementation"""
1159
_transport = HttpTransport_urllib
1161
def _qualified_url(self, host, port):
1162
result = 'http+urllib://%s:%s' % (host, port)
1163
self.permit_url(result)
1168
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1169
TestHTTPRedirections,
1170
http_utils.TestCaseWithTwoWebservers):
1171
"""Tests redirections for pycurl implementation"""
1173
def _qualified_url(self, host, port):
1174
result = 'http+pycurl://%s:%s' % (host, port)
1175
self.permit_url(result)
1179
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1180
http_utils.TestCaseWithTwoWebservers):
1181
"""Tests redirections for the nosmart decorator"""
1183
_transport = NoSmartTransportDecorator
1185
def _qualified_url(self, host, port):
1186
result = 'nosmart+http://%s:%s' % (host, port)
1187
self.permit_url(result)
1191
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1192
http_utils.TestCaseWithTwoWebservers):
1193
"""Tests redirections for readonly decoratror"""
1195
_transport = ReadonlyTransportDecorator
1197
def _qualified_url(self, host, port):
1198
result = 'readonly+http://%s:%s' % (host, port)
1199
self.permit_url(result)
1203
class TestDotBzrHidden(TestCaseWithTransport):
1206
if sys.platform == 'win32':
1207
ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
1210
f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
1211
stderr=subprocess.PIPE)
1212
out, err = f.communicate()
1213
self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
1215
return out.splitlines()
1217
def test_dot_bzr_hidden(self):
1218
if sys.platform == 'win32' and not win32utils.has_win32file:
1219
raise TestSkipped('unable to make file hidden without pywin32 library')
1220
b = bzrdir.BzrDir.create('.')
1221
self.build_tree(['a'])
1222
self.assertEquals(['a'], self.get_ls())
1224
def test_dot_bzr_hidden_with_url(self):
1225
if sys.platform == 'win32' and not win32utils.has_win32file:
1226
raise TestSkipped('unable to make file hidden without pywin32 library')
1227
b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
1228
self.build_tree(['a'])
1229
self.assertEquals(['a'], self.get_ls())
1232
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
1233
"""Test BzrDirFormat implementation for TestBzrDirSprout."""
1235
def _open(self, transport):
1236
return _TestBzrDir(transport, self)
1239
class _TestBzrDir(bzrdir.BzrDirMeta1):
1240
"""Test BzrDir implementation for TestBzrDirSprout.
1242
When created a _TestBzrDir already has repository and a branch. The branch
1243
is a test double as well.
1246
def __init__(self, *args, **kwargs):
1247
super(_TestBzrDir, self).__init__(*args, **kwargs)
1248
self.test_branch = _TestBranch(self.transport)
1249
self.test_branch.repository = self.create_repository()
1251
def open_branch(self, unsupported=False, possible_transports=None):
1252
return self.test_branch
1254
def cloning_metadir(self, require_stacking=False):
1255
return _TestBzrDirFormat()
1258
class _TestBranchFormat(bzrlib.branch.BranchFormat):
1259
"""Test Branch format for TestBzrDirSprout."""
1262
class _TestBranch(bzrlib.branch.Branch):
1263
"""Test Branch implementation for TestBzrDirSprout."""
1265
def __init__(self, transport, *args, **kwargs):
1266
self._format = _TestBranchFormat()
1267
self._transport = transport
1268
self.base = transport.base
1269
super(_TestBranch, self).__init__(*args, **kwargs)
1273
def sprout(self, *args, **kwargs):
1274
self.calls.append('sprout')
1275
return _TestBranch(self._transport)
1277
def copy_content_into(self, destination, revision_id=None):
1278
self.calls.append('copy_content_into')
1280
def last_revision(self):
1281
return _mod_revision.NULL_REVISION
1283
def get_parent(self):
1286
def _get_config(self):
1287
return config.TransportConfig(self._transport, 'branch.conf')
1289
def set_parent(self, parent):
1290
self._parent = parent
1292
def lock_read(self):
1293
return lock.LogicalLockResult(self.unlock)
1299
class TestBzrDirSprout(TestCaseWithMemoryTransport):
1301
def test_sprout_uses_branch_sprout(self):
1302
"""BzrDir.sprout calls Branch.sprout.
1304
Usually, BzrDir.sprout should delegate to the branch's sprout method
1305
for part of the work. This allows the source branch to control the
1306
choice of format for the new branch.
1308
There are exceptions, but this tests avoids them:
1309
- if there's no branch in the source bzrdir,
1310
- or if the stacking has been requested and the format needs to be
1311
overridden to satisfy that.
1313
# Make an instrumented bzrdir.
1314
t = self.get_transport('source')
1316
source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
1317
# The instrumented bzrdir has a test_branch attribute that logs calls
1318
# made to the branch contained in that bzrdir. Initially the test
1319
# branch exists but no calls have been made to it.
1320
self.assertEqual([], source_bzrdir.test_branch.calls)
1323
target_url = self.get_url('target')
1324
result = source_bzrdir.sprout(target_url, recurse='no')
1326
# The bzrdir called the branch's sprout method.
1327
self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
1329
def test_sprout_parent(self):
1330
grandparent_tree = self.make_branch('grandparent')
1331
parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
1332
branch_tree = parent.bzrdir.sprout('branch').open_branch()
1333
self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
1336
class TestBzrDirHooks(TestCaseWithMemoryTransport):
1338
def test_pre_open_called(self):
1340
bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
1341
transport = self.get_transport('foo')
1342
url = transport.base
1343
self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
1344
self.assertEqual([transport.base], [t.base for t in calls])
1346
def test_pre_open_actual_exceptions_raised(self):
1348
def fail_once(transport):
1351
raise errors.BzrError("fail")
1352
bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
1353
transport = self.get_transport('foo')
1354
url = transport.base
1355
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1356
self.assertEqual('fail', err._preformatted_string)
1358
def test_post_repo_init(self):
1359
from bzrlib.controldir import RepoInitHookParams
1361
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1363
self.make_repository('foo')
1364
self.assertLength(1, calls)
1366
self.assertIsInstance(params, RepoInitHookParams)
1367
self.assertTrue(hasattr(params, 'bzrdir'))
1368
self.assertTrue(hasattr(params, 'repository'))
1370
def test_post_repo_init_hook_repr(self):
1372
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1373
lambda params: param_reprs.append(repr(params)), None)
1374
self.make_repository('foo')
1375
self.assertLength(1, param_reprs)
1376
param_repr = param_reprs[0]
1377
self.assertStartsWith(param_repr, '<RepoInitHookParams for ')
1380
class TestGenerateBackupName(TestCaseWithMemoryTransport):
1381
# FIXME: This may need to be unified with test_osutils.TestBackupNames or
1382
# moved to per_bzrdir or per_transport for better coverage ?
1386
super(TestGenerateBackupName, self).setUp()
1387
self._transport = self.get_transport()
1388
bzrdir.BzrDir.create(self.get_url(),
1389
possible_transports=[self._transport])
1390
self._bzrdir = bzrdir.BzrDir.open_from_transport(self._transport)
1392
def test_deprecated_generate_backup_name(self):
1393
res = self.applyDeprecated(
1394
symbol_versioning.deprecated_in((2, 3, 0)),
1395
self._bzrdir.generate_backup_name, 'whatever')
1398
self.assertEqual("a.~1~", self._bzrdir._available_backup_name("a"))
1400
def test_exiting(self):
1401
self._transport.put_bytes("a.~1~", "some content")
1402
self.assertEqual("a.~2~", self._bzrdir._available_backup_name("a"))
1405
class TestMeta1DirColoFormat(TestCaseWithTransport):
1406
"""Tests specific to the meta1 dir with colocated branches format."""
1408
def test_supports_colo(self):
1409
format = bzrdir.BzrDirMetaFormat1Colo()
1410
self.assertTrue(format.colocated_branches)
1412
def test_upgrade_from_2a(self):
1413
tree = self.make_branch_and_tree('.', format='2a')
1414
format = bzrdir.BzrDirMetaFormat1Colo()
1415
self.assertTrue(tree.bzrdir.needs_format_conversion(format))
1416
converter = tree.bzrdir._format.get_converter(format)
1417
result = converter.convert(tree.bzrdir, None)
1418
self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1Colo)
1419
self.assertFalse(result.needs_format_conversion(format))
1421
def test_downgrade_to_2a(self):
1422
tree = self.make_branch_and_tree('.', format='development-colo')
1423
format = bzrdir.BzrDirMetaFormat1()
1424
self.assertTrue(tree.bzrdir.needs_format_conversion(format))
1425
converter = tree.bzrdir._format.get_converter(format)
1426
result = converter.convert(tree.bzrdir, None)
1427
self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1)
1428
self.assertFalse(result.needs_format_conversion(format))
1430
def test_downgrade_to_2a_too_many_branches(self):
1431
tree = self.make_branch_and_tree('.', format='development-colo')
1432
tree.bzrdir.create_branch(name="another-colocated-branch")
1433
converter = tree.bzrdir._format.get_converter(
1434
bzrdir.BzrDirMetaFormat1())
1435
self.assertRaises(errors.BzrError, converter.convert, tree.bzrdir,