13
13
# You should have received a copy of the GNU General Public License
14
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
17
"""Tests for the BzrDir facility and any format specific tests.
19
For interface contract tests, see tests/per_bzr_dir.
19
For interface contract tests, see tests/bzr_dir_implementations.
23
from StringIO import StringIO
26
25
from bzrlib import (
35
revision as _mod_revision,
38
transport as _mod_transport,
44
34
import bzrlib.branch
45
from bzrlib.branchfmt.fullhistory import BzrBranchFormat5
46
from bzrlib.errors import (
48
NoColocatedBranchSupport,
50
UnsupportedFormatError,
35
from bzrlib.errors import (NotBranchError,
37
UnsupportedFormatError,
52
39
from bzrlib.tests import (
54
TestCaseWithMemoryTransport,
55
41
TestCaseWithTransport,
58
from bzrlib.tests import(
44
from bzrlib.tests.HttpServer import HttpServer
45
from bzrlib.tests.HTTPTestUtil import (
46
TestCaseWithTwoWebservers,
47
HTTPServerRedirecting,
62
49
from bzrlib.tests.test_http import TestWithTransport_pycurl
63
from bzrlib.transport import (
50
from bzrlib.transport import get_transport
67
51
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
52
from bzrlib.transport.memory import MemoryServer
53
from bzrlib.repofmt import knitrepo, weaverepo
73
56
class TestDefaultFormat(TestCase):
75
58
def test_get_set_default_format(self):
76
59
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())
60
# default is BzrDirFormat6
61
self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
62
self.applyDeprecated(symbol_versioning.zero_fourteen,
63
bzrdir.BzrDirFormat.set_default_format,
80
65
# creating a bzr dir should now create an instrumented dir.
82
67
result = bzrdir.BzrDir.create('memory:///')
83
self.assertIsInstance(result, SampleBzrDir)
68
self.failUnless(isinstance(result, SampleBzrDir))
85
controldir.ControlDirFormat._set_default_format(old_format)
70
self.applyDeprecated(symbol_versioning.zero_fourteen,
71
bzrdir.BzrDirFormat.set_default_format, old_format)
86
72
self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
89
class DeprecatedBzrDirFormat(bzrdir.BzrDirFormat):
90
"""A deprecated bzr dir format."""
93
75
class TestFormatRegistry(TestCase):
95
77
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',
78
my_format_registry = bzrdir.BzrDirFormatRegistry()
79
my_format_registry.register('weave', bzrdir.BzrDirFormat6,
80
'Pre-0.8 format. Slower and does not support checkouts or shared'
81
' repositories', deprecated=True)
82
my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir',
83
'BzrDirFormat6', 'Format registered lazily', deprecated=True)
84
my_format_registry.register_metadir('knit',
104
85
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
105
86
'Format using knits',
107
88
my_format_registry.set_default('knit')
108
bzrdir.register_metadir(my_format_registry,
89
my_format_registry.register_metadir(
110
91
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
111
92
'Experimental successor to knit. Use at your own risk.',
112
branch_format='bzrlib.branch.BzrBranchFormat6',
114
bzrdir.register_metadir(my_format_registry,
93
branch_format='bzrlib.branch.BzrBranchFormat6')
94
my_format_registry.register_metadir(
116
96
'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
117
97
'Experimental successor to knit. Use at your own risk.',
118
98
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)
99
my_format_registry.register('hiddenweave', bzrdir.BzrDirFormat6,
100
'Pre-0.8 format. Slower and does not support checkouts or shared'
101
' repositories', hidden=True)
102
my_format_registry.register_lazy('hiddenlazy', 'bzrlib.bzrdir',
103
'BzrDirFormat6', 'Format registered lazily', deprecated=True,
124
105
return my_format_registry
126
107
def test_format_registry(self):
127
108
my_format_registry = self.make_format_registry()
128
109
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)
110
self.assertIsInstance(my_bzrdir, bzrdir.BzrDirFormat6)
111
my_bzrdir = my_format_registry.make_bzrdir('weave')
112
self.assertIsInstance(my_bzrdir, bzrdir.BzrDirFormat6)
132
113
my_bzrdir = my_format_registry.make_bzrdir('default')
133
self.assertIsInstance(my_bzrdir.repository_format,
114
self.assertIsInstance(my_bzrdir.repository_format,
134
115
knitrepo.RepositoryFormatKnit1)
135
116
my_bzrdir = my_format_registry.make_bzrdir('knit')
136
self.assertIsInstance(my_bzrdir.repository_format,
117
self.assertIsInstance(my_bzrdir.repository_format,
137
118
knitrepo.RepositoryFormatKnit1)
138
119
my_bzrdir = my_format_registry.make_bzrdir('branch6')
139
120
self.assertIsInstance(my_bzrdir.get_branch_format(),
143
124
my_format_registry = self.make_format_registry()
144
125
self.assertEqual('Format registered lazily',
145
126
my_format_registry.get_help('lazy'))
146
self.assertEqual('Format using knits',
127
self.assertEqual('Format using knits',
147
128
my_format_registry.get_help('knit'))
148
self.assertEqual('Format using knits',
129
self.assertEqual('Format using knits',
149
130
my_format_registry.get_help('default'))
150
self.assertEqual('Some format. Slower and unawesome and deprecated.',
151
my_format_registry.get_help('deprecated'))
131
self.assertEqual('Pre-0.8 format. Slower and does not support'
132
' checkouts or shared repositories',
133
my_format_registry.get_help('weave'))
153
135
def test_help_topic(self):
154
136
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')
137
topics.register('formats', self.make_format_registry().help_topic,
139
topic = topics.get_detail('formats')
140
new, deprecated = topic.split('Deprecated formats')
141
self.assertContainsRe(new, 'Bazaar directory formats')
142
self.assertContainsRe(new,
143
' knit/default:\n \(native\) Format using knits\n')
144
self.assertContainsRe(deprecated,
145
' lazy:\n \(native\) Format registered lazily\n')
170
146
self.assertNotContainsRe(new, 'hidden')
172
148
def test_set_default_repository(self):
173
default_factory = controldir.format_registry.get('default')
174
old_default = [k for k, v in controldir.format_registry.iteritems()
149
default_factory = bzrdir.format_registry.get('default')
150
old_default = [k for k, v in bzrdir.format_registry.iteritems()
175
151
if v == default_factory and k != 'default'][0]
176
controldir.format_registry.set_default_repository('dirstate-with-subtree')
152
bzrdir.format_registry.set_default_repository('dirstate-with-subtree')
178
self.assertIs(controldir.format_registry.get('dirstate-with-subtree'),
179
controldir.format_registry.get('default'))
154
self.assertIs(bzrdir.format_registry.get('dirstate-with-subtree'),
155
bzrdir.format_registry.get('default'))
181
repository.format_registry.get_default().__class__,
157
repository.RepositoryFormat.get_default_format().__class__,
182
158
knitrepo.RepositoryFormatKnit3)
184
controldir.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())
160
bzrdir.format_registry.set_default_repository(old_default)
197
163
class SampleBranch(bzrlib.branch.Branch):
464
413
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 = controldir.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
416
class ChrootedTests(TestCaseWithTransport):
631
417
"""A support class that provides readonly urls outside the local namespace.
654
437
branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
655
438
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
440
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')))
441
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
442
get_transport(self.get_readonly_url('')))
443
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
444
get_transport(self.get_readonly_url('g/p/q')))
752
445
control = bzrdir.BzrDir.create(self.get_url())
753
446
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
754
_mod_transport.get_transport_from_url(
755
self.get_readonly_url('')))
447
get_transport(self.get_readonly_url('')))
756
448
self.assertEqual('', relpath)
757
449
branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
758
_mod_transport.get_transport_from_url(
759
self.get_readonly_url('g/p/q')))
450
get_transport(self.get_readonly_url('g/p/q')))
760
451
self.assertEqual('g/p/q', relpath)
762
453
def test_open_containing_tree_or_branch(self):
454
def local_branch_path(branch):
455
return os.path.realpath(
456
urlutils.local_path_from_url(branch.base))
763
458
self.make_branch_and_tree('topdir')
764
459
tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
766
461
self.assertEqual(os.path.realpath('topdir'),
767
462
os.path.realpath(tree.basedir))
768
463
self.assertEqual(os.path.realpath('topdir'),
769
self.local_branch_path(branch))
464
local_branch_path(branch))
770
465
self.assertIs(tree.bzrdir, branch.bzrdir)
771
466
self.assertEqual('foo', relpath)
772
467
# opening from non-local should not return the tree
781
476
self.assertIs(tree, None)
782
477
self.assertEqual(os.path.realpath('topdir/foo'),
783
self.local_branch_path(branch))
478
local_branch_path(branch))
784
479
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
481
def test_open_from_transport(self):
806
482
# transport pointing at bzrdir should give a bzrdir with root transport
807
483
# set to the given transport
808
484
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)
485
transport = get_transport(self.get_url())
486
opened_bzrdir = bzrdir.BzrDir.open_from_transport(transport)
487
self.assertEqual(transport.base, opened_bzrdir.root_transport.base)
812
488
self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
814
490
def test_open_from_transport_no_bzrdir(self):
815
t = self.get_transport()
816
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
491
transport = get_transport(self.get_url())
492
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
818
495
def test_open_from_transport_bzrdir_in_parent(self):
819
496
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)
497
transport = get_transport(self.get_url())
498
transport.mkdir('subdir')
499
transport = transport.clone('subdir')
500
self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
825
503
def test_sprout_recursive(self):
826
tree = self.make_branch_and_tree('tree1',
827
format='development-subtree')
504
tree = self.make_branch_and_tree('tree1', format='dirstate-with-subtree')
828
505
sub_tree = self.make_branch_and_tree('tree1/subtree',
829
format='development-subtree')
830
sub_tree.set_root_id('subtree-root')
506
format='dirstate-with-subtree')
831
507
tree.add_reference(sub_tree)
832
508
self.build_tree(['tree1/subtree/file'])
833
509
sub_tree.add('file')
834
510
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'))
511
tree.bzrdir.sprout('tree2')
512
self.failUnlessExists('tree2/subtree/file')
841
514
def test_cloning_metadir(self):
842
515
"""Ensure that cloning metadir is suitable"""
845
518
branch = self.make_branch('branch', format='knit')
846
519
format = branch.bzrdir.cloning_metadir()
847
520
self.assertIsInstance(format.workingtree_format,
848
workingtree_4.WorkingTreeFormat6)
521
workingtree.WorkingTreeFormat3)
850
523
def test_sprout_recursive_treeless(self):
851
524
tree = self.make_branch_and_tree('tree1',
852
format='development-subtree')
525
format='dirstate-with-subtree')
853
526
sub_tree = self.make_branch_and_tree('tree1/subtree',
854
format='development-subtree')
527
format='dirstate-with-subtree')
855
528
tree.add_reference(sub_tree)
856
529
self.build_tree(['tree1/subtree/file'])
857
530
sub_tree.add('file')
858
531
tree.commit('Initial commit')
859
# The following line force the orhaning to reveal bug #634470
860
tree.branch.get_config_stack().set(
861
'bzr.transform.orphan_policy', 'move')
862
532
tree.bzrdir.destroy_workingtree()
863
# FIXME: subtree/.bzr is left here which allows the test to pass (or
864
# fail :-( ) -- vila 20100909
865
533
repo = self.make_repository('repo', shared=True,
866
format='development-subtree')
534
format='dirstate-with-subtree')
867
535
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/')
536
tree.bzrdir.sprout('repo/tree2')
537
self.failUnlessExists('repo/tree2/subtree')
538
self.failIfExists('repo/tree2/subtree/file')
986
541
class TestMeta1DirFormat(TestCaseWithTransport):
1015
569
Metadirs should compare equal iff they have the same repo, branch and
1018
mydir = controldir.format_registry.make_bzrdir('knit')
572
mydir = bzrdir.format_registry.make_bzrdir('knit')
1019
573
self.assertEqual(mydir, mydir)
1020
574
self.assertFalse(mydir != mydir)
1021
otherdir = controldir.format_registry.make_bzrdir('knit')
575
otherdir = bzrdir.format_registry.make_bzrdir('knit')
1022
576
self.assertEqual(otherdir, mydir)
1023
577
self.assertFalse(otherdir != mydir)
1024
otherdir2 = controldir.format_registry.make_bzrdir('development-subtree')
578
otherdir2 = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
1025
579
self.assertNotEqual(otherdir2, mydir)
1026
580
self.assertFalse(otherdir2 == mydir)
1028
def test_with_features(self):
1029
tree = self.make_branch_and_tree('tree', format='2a')
1030
tree.bzrdir.update_feature_flags({"bar": "required"})
1031
self.assertRaises(errors.MissingFeature, bzrdir.BzrDir.open, 'tree')
1032
bzrdir.BzrDirMetaFormat1.register_feature('bar')
1033
self.addCleanup(bzrdir.BzrDirMetaFormat1.unregister_feature, 'bar')
1034
dir = bzrdir.BzrDir.open('tree')
1035
self.assertEqual("required", dir._format.features.get("bar"))
1036
tree.bzrdir.update_feature_flags({"bar": None, "nonexistant": None})
1037
dir = bzrdir.BzrDir.open('tree')
1038
self.assertEqual({}, dir._format.features)
1040
582
def test_needs_conversion_different_working_tree(self):
1041
583
# meta1dirs need an conversion if any element is not the default.
1042
new_format = controldir.format_registry.make_bzrdir('dirstate')
1043
tree = self.make_branch_and_tree('tree', format='knit')
1044
self.assertTrue(tree.bzrdir.needs_format_conversion(
1047
def test_initialize_on_format_uses_smart_transport(self):
1048
self.setup_smart_server_with_call_log()
1049
new_format = controldir.format_registry.make_bzrdir('dirstate')
1050
transport = self.get_transport('target')
1051
transport.ensure_base()
1052
self.reset_smart_call_log()
1053
instance = new_format.initialize_on_transport(transport)
1054
self.assertIsInstance(instance, remote.RemoteBzrDir)
1055
rpc_count = len(self.hpss_calls)
1056
# This figure represent the amount of work to perform this use case. It
1057
# is entirely ok to reduce this number if a test fails due to rpc_count
1058
# being too low. If rpc_count increases, more network roundtrips have
1059
# become necessary for this use case. Please do not adjust this number
1060
# upwards without agreement from bzr's network support maintainers.
1061
self.assertEqual(2, rpc_count)
584
old_format = bzrdir.BzrDirFormat.get_default_format()
586
new_default = bzrdir.format_registry.make_bzrdir('dirstate')
587
bzrdir.BzrDirFormat._set_default_format(new_default)
589
tree = self.make_branch_and_tree('tree', format='knit')
590
self.assertTrue(tree.bzrdir.needs_format_conversion())
592
bzrdir.BzrDirFormat._set_default_format(old_format)
595
class TestFormat5(TestCaseWithTransport):
596
"""Tests specific to the version 5 bzrdir format."""
598
def test_same_lockfiles_between_tree_repo_branch(self):
599
# this checks that only a single lockfiles instance is created
600
# for format 5 objects
601
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
602
def check_dir_components_use_same_lock(dir):
603
ctrl_1 = dir.open_repository().control_files
604
ctrl_2 = dir.open_branch().control_files
605
ctrl_3 = dir.open_workingtree()._control_files
606
self.assertTrue(ctrl_1 is ctrl_2)
607
self.assertTrue(ctrl_2 is ctrl_3)
608
check_dir_components_use_same_lock(dir)
609
# and if we open it normally.
610
dir = bzrdir.BzrDir.open(self.get_url())
611
check_dir_components_use_same_lock(dir)
613
def test_can_convert(self):
614
# format 5 dirs are convertable
615
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
616
self.assertTrue(dir.can_convert_format())
618
def test_needs_conversion(self):
619
# format 5 dirs need a conversion if they are not the default.
620
# and they start of not the default.
621
old_format = bzrdir.BzrDirFormat.get_default_format()
622
bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirFormat5())
624
dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
625
self.assertFalse(dir.needs_format_conversion())
627
bzrdir.BzrDirFormat._set_default_format(old_format)
628
self.assertTrue(dir.needs_format_conversion())
631
class TestFormat6(TestCaseWithTransport):
632
"""Tests specific to the version 6 bzrdir format."""
634
def test_same_lockfiles_between_tree_repo_branch(self):
635
# this checks that only a single lockfiles instance is created
636
# for format 6 objects
637
dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
638
def check_dir_components_use_same_lock(dir):
639
ctrl_1 = dir.open_repository().control_files
640
ctrl_2 = dir.open_branch().control_files
641
ctrl_3 = dir.open_workingtree()._control_files
642
self.assertTrue(ctrl_1 is ctrl_2)
643
self.assertTrue(ctrl_2 is ctrl_3)
644
check_dir_components_use_same_lock(dir)
645
# and if we open it normally.
646
dir = bzrdir.BzrDir.open(self.get_url())
647
check_dir_components_use_same_lock(dir)
649
def test_can_convert(self):
650
# format 6 dirs are convertable
651
dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
652
self.assertTrue(dir.can_convert_format())
654
def test_needs_conversion(self):
655
# format 6 dirs need an conversion if they are not the default.
656
old_format = bzrdir.BzrDirFormat.get_default_format()
657
bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirMetaFormat1())
659
dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
660
self.assertTrue(dir.needs_format_conversion())
662
bzrdir.BzrDirFormat._set_default_format(old_format)
665
class NotBzrDir(bzrlib.bzrdir.BzrDir):
666
"""A non .bzr based control directory."""
668
def __init__(self, transport, format):
669
self._format = format
670
self.root_transport = transport
671
self.transport = transport.clone('.not')
674
class NotBzrDirFormat(bzrlib.bzrdir.BzrDirFormat):
675
"""A test class representing any non-.bzr based disk format."""
677
def initialize_on_transport(self, transport):
678
"""Initialize a new .not dir in the base directory of a Transport."""
679
transport.mkdir('.not')
680
return self.open(transport)
682
def open(self, transport):
683
"""Open this directory."""
684
return NotBzrDir(transport, self)
687
def _known_formats(self):
688
return set([NotBzrDirFormat()])
691
def probe_transport(self, transport):
692
"""Our format is present if the transport ends in '.not/'."""
693
if transport.has('.not'):
694
return NotBzrDirFormat()
697
class TestNotBzrDir(TestCaseWithTransport):
698
"""Tests for using the bzrdir api with a non .bzr based disk format.
700
If/when one of these is in the core, we can let the implementation tests
704
def test_create_and_find_format(self):
705
# create a .notbzr dir
706
format = NotBzrDirFormat()
707
dir = format.initialize(self.get_url())
708
self.assertIsInstance(dir, NotBzrDir)
710
bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
712
found = bzrlib.bzrdir.BzrDirFormat.find_format(
713
get_transport(self.get_url()))
714
self.assertIsInstance(found, NotBzrDirFormat)
716
bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
718
def test_included_in_known_formats(self):
719
bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
721
formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
722
for format in formats:
723
if isinstance(format, NotBzrDirFormat):
725
self.fail("No NotBzrDirFormat in %s" % formats)
727
bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
1064
730
class NonLocalTests(TestCaseWithTransport):
1107
773
my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
1108
774
checkout_format = my_bzrdir.checkout_metadir()
1109
775
self.assertIsInstance(checkout_format.workingtree_format,
1110
workingtree_4.WorkingTreeFormat4)
1113
class TestHTTPRedirections(object):
1114
"""Test redirection between two http servers.
776
workingtree.WorkingTreeFormat3)
779
class TestHTTPRedirectionLoop(object):
780
"""Test redirection loop between two http servers.
1116
782
This MUST be used by daughter classes that also inherit from
1117
783
TestCaseWithTwoWebservers.
1119
785
We can't inherit directly from TestCaseWithTwoWebservers or the
1120
786
test framework will try to create an instance which cannot
1121
run, its implementation being incomplete.
787
run, its implementation being incomplete.
790
# Should be defined by daughter classes to ensure redirection
791
# still use the same transport implementation (not currently
792
# enforced as it's a bit tricky to get right (see the FIXME
793
# in BzrDir.open_from_transport for the unique use case so
1124
797
def create_transport_readonly_server(self):
1125
# We don't set the http protocol version, relying on the default
1126
return http_utils.HTTPServerRedirecting()
798
return HTTPServerRedirecting()
1128
800
def create_transport_secondary_server(self):
1129
# We don't set the http protocol version, relying on the default
1130
return http_utils.HTTPServerRedirecting()
801
return HTTPServerRedirecting()
1132
803
def setUp(self):
1133
super(TestHTTPRedirections, self).setUp()
804
# Both servers redirect to each server creating a loop
805
super(TestHTTPRedirectionLoop, self).setUp()
1134
806
# The redirections will point to the new server
1135
807
self.new_server = self.get_readonly_server()
1136
808
# The requests to the old server will be redirected
1137
809
self.old_server = self.get_secondary_server()
1138
810
# Configure the redirections
1139
811
self.old_server.redirect_to(self.new_server.host, self.new_server.port)
812
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
814
def _qualified_url(self, host, port):
815
return 'http+%s://%s:%s' % (self._qualifier, host, port)
1141
817
def test_loop(self):
1142
# Both servers redirect to each other creating a loop
1143
self.new_server.redirect_to(self.old_server.host, self.old_server.port)
1144
818
# Starting from either server should loop
1145
old_url = self._qualified_url(self.old_server.host,
819
old_url = self._qualified_url(self.old_server.host,
1146
820
self.old_server.port)
1147
821
oldt = self._transport(old_url)
1148
822
self.assertRaises(errors.NotBranchError,
1149
823
bzrdir.BzrDir.open_from_transport, oldt)
1150
new_url = self._qualified_url(self.new_server.host,
824
new_url = self._qualified_url(self.new_server.host,
1151
825
self.new_server.port)
1152
826
newt = self._transport(new_url)
1153
827
self.assertRaises(errors.NotBranchError,
1154
828
bzrdir.BzrDir.open_from_transport, newt)
1156
def test_qualifier_preserved(self):
1157
wt = self.make_branch_and_tree('branch')
1158
old_url = self._qualified_url(self.old_server.host,
1159
self.old_server.port)
1160
start = self._transport(old_url).clone('branch')
1161
bdir = bzrdir.BzrDir.open_from_transport(start)
1162
# Redirection should preserve the qualifier, hence the transport class
1164
self.assertIsInstance(bdir.root_transport, type(start))
1167
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1168
http_utils.TestCaseWithTwoWebservers):
831
class TestHTTPRedirections_urllib(TestHTTPRedirectionLoop,
832
TestCaseWithTwoWebservers):
1169
833
"""Tests redirections for urllib implementation"""
835
_qualifier = 'urllib'
1171
836
_transport = HttpTransport_urllib
1173
def _qualified_url(self, host, port):
1174
result = 'http+urllib://%s:%s' % (host, port)
1175
self.permit_url(result)
1180
840
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1181
TestHTTPRedirections,
1182
http_utils.TestCaseWithTwoWebservers):
841
TestHTTPRedirectionLoop,
842
TestCaseWithTwoWebservers):
1183
843
"""Tests redirections for pycurl implementation"""
1185
def _qualified_url(self, host, port):
1186
result = 'http+pycurl://%s:%s' % (host, port)
1187
self.permit_url(result)
1191
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1192
http_utils.TestCaseWithTwoWebservers):
1193
"""Tests redirections for the nosmart decorator"""
1195
_transport = NoSmartTransportDecorator
1197
def _qualified_url(self, host, port):
1198
result = 'nosmart+http://%s:%s' % (host, port)
1199
self.permit_url(result)
1203
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1204
http_utils.TestCaseWithTwoWebservers):
1205
"""Tests redirections for readonly decoratror"""
1207
_transport = ReadonlyTransportDecorator
1209
def _qualified_url(self, host, port):
1210
result = 'readonly+http://%s:%s' % (host, port)
1211
self.permit_url(result)
1215
class TestDotBzrHidden(TestCaseWithTransport):
1218
if sys.platform == 'win32':
1219
ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
1222
f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
1223
stderr=subprocess.PIPE)
1224
out, err = f.communicate()
1225
self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
1227
return out.splitlines()
1229
def test_dot_bzr_hidden(self):
1230
if sys.platform == 'win32' and not win32utils.has_win32file:
1231
raise TestSkipped('unable to make file hidden without pywin32 library')
1232
b = bzrdir.BzrDir.create('.')
1233
self.build_tree(['a'])
1234
self.assertEqual(['a'], self.get_ls())
1236
def test_dot_bzr_hidden_with_url(self):
1237
if sys.platform == 'win32' and not win32utils.has_win32file:
1238
raise TestSkipped('unable to make file hidden without pywin32 library')
1239
b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
1240
self.build_tree(['a'])
1241
self.assertEqual(['a'], self.get_ls())
1244
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
1245
"""Test BzrDirFormat implementation for TestBzrDirSprout."""
1247
def _open(self, transport):
1248
return _TestBzrDir(transport, self)
1251
class _TestBzrDir(bzrdir.BzrDirMeta1):
1252
"""Test BzrDir implementation for TestBzrDirSprout.
1254
When created a _TestBzrDir already has repository and a branch. The branch
1255
is a test double as well.
1258
def __init__(self, *args, **kwargs):
1259
super(_TestBzrDir, self).__init__(*args, **kwargs)
1260
self.test_branch = _TestBranch(self.transport)
1261
self.test_branch.repository = self.create_repository()
1263
def open_branch(self, unsupported=False, possible_transports=None):
1264
return self.test_branch
1266
def cloning_metadir(self, require_stacking=False):
1267
return _TestBzrDirFormat()
1270
class _TestBranchFormat(bzrlib.branch.BranchFormat):
1271
"""Test Branch format for TestBzrDirSprout."""
1274
class _TestBranch(bzrlib.branch.Branch):
1275
"""Test Branch implementation for TestBzrDirSprout."""
1277
def __init__(self, transport, *args, **kwargs):
1278
self._format = _TestBranchFormat()
1279
self._transport = transport
1280
self.base = transport.base
1281
super(_TestBranch, self).__init__(*args, **kwargs)
1285
def sprout(self, *args, **kwargs):
1286
self.calls.append('sprout')
1287
return _TestBranch(self._transport)
1289
def copy_content_into(self, destination, revision_id=None):
1290
self.calls.append('copy_content_into')
1292
def last_revision(self):
1293
return _mod_revision.NULL_REVISION
1295
def get_parent(self):
1298
def _get_config(self):
1299
return config.TransportConfig(self._transport, 'branch.conf')
1301
def _get_config_store(self):
1302
return config.BranchStore(self)
1304
def set_parent(self, parent):
1305
self._parent = parent
1307
def lock_read(self):
1308
return lock.LogicalLockResult(self.unlock)
1314
class TestBzrDirSprout(TestCaseWithMemoryTransport):
1316
def test_sprout_uses_branch_sprout(self):
1317
"""BzrDir.sprout calls Branch.sprout.
1319
Usually, BzrDir.sprout should delegate to the branch's sprout method
1320
for part of the work. This allows the source branch to control the
1321
choice of format for the new branch.
1323
There are exceptions, but this tests avoids them:
1324
- if there's no branch in the source bzrdir,
1325
- or if the stacking has been requested and the format needs to be
1326
overridden to satisfy that.
1328
# Make an instrumented bzrdir.
1329
t = self.get_transport('source')
1331
source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
1332
# The instrumented bzrdir has a test_branch attribute that logs calls
1333
# made to the branch contained in that bzrdir. Initially the test
1334
# branch exists but no calls have been made to it.
1335
self.assertEqual([], source_bzrdir.test_branch.calls)
1338
target_url = self.get_url('target')
1339
result = source_bzrdir.sprout(target_url, recurse='no')
1341
# The bzrdir called the branch's sprout method.
1342
self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
1344
def test_sprout_parent(self):
1345
grandparent_tree = self.make_branch('grandparent')
1346
parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
1347
branch_tree = parent.bzrdir.sprout('branch').open_branch()
1348
self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
1351
class TestBzrDirHooks(TestCaseWithMemoryTransport):
1353
def test_pre_open_called(self):
1355
bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
1356
transport = self.get_transport('foo')
1357
url = transport.base
1358
self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
1359
self.assertEqual([transport.base], [t.base for t in calls])
1361
def test_pre_open_actual_exceptions_raised(self):
1363
def fail_once(transport):
1366
raise errors.BzrError("fail")
1367
bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
1368
transport = self.get_transport('foo')
1369
url = transport.base
1370
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1371
self.assertEqual('fail', err._preformatted_string)
1373
def test_post_repo_init(self):
1374
from bzrlib.controldir import RepoInitHookParams
1376
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1378
self.make_repository('foo')
1379
self.assertLength(1, calls)
1381
self.assertIsInstance(params, RepoInitHookParams)
1382
self.assertTrue(hasattr(params, 'bzrdir'))
1383
self.assertTrue(hasattr(params, 'repository'))
1385
def test_post_repo_init_hook_repr(self):
1387
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1388
lambda params: param_reprs.append(repr(params)), None)
1389
self.make_repository('foo')
1390
self.assertLength(1, param_reprs)
1391
param_repr = param_reprs[0]
1392
self.assertStartsWith(param_repr, '<RepoInitHookParams for ')
1395
class TestGenerateBackupName(TestCaseWithMemoryTransport):
1396
# FIXME: This may need to be unified with test_osutils.TestBackupNames or
1397
# moved to per_bzrdir or per_transport for better coverage ?
1401
super(TestGenerateBackupName, self).setUp()
1402
self._transport = self.get_transport()
1403
bzrdir.BzrDir.create(self.get_url(),
1404
possible_transports=[self._transport])
1405
self._bzrdir = bzrdir.BzrDir.open_from_transport(self._transport)
1408
self.assertEqual("a.~1~", self._bzrdir._available_backup_name("a"))
1410
def test_exiting(self):
1411
self._transport.put_bytes("a.~1~", "some content")
1412
self.assertEqual("a.~2~", self._bzrdir._available_backup_name("a"))
1415
class TestMeta1DirColoFormat(TestCaseWithTransport):
1416
"""Tests specific to the meta1 dir with colocated branches format."""
1418
def test_supports_colo(self):
1419
format = bzrdir.BzrDirMetaFormat1Colo()
1420
self.assertTrue(format.colocated_branches)
1422
def test_upgrade_from_2a(self):
1423
tree = self.make_branch_and_tree('.', format='2a')
1424
format = bzrdir.BzrDirMetaFormat1Colo()
1425
self.assertTrue(tree.bzrdir.needs_format_conversion(format))
1426
converter = tree.bzrdir._format.get_converter(format)
1427
result = converter.convert(tree.bzrdir, None)
1428
self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1Colo)
1429
self.assertFalse(result.needs_format_conversion(format))
1431
def test_downgrade_to_2a(self):
1432
tree = self.make_branch_and_tree('.', format='development-colo')
1433
format = bzrdir.BzrDirMetaFormat1()
1434
self.assertTrue(tree.bzrdir.needs_format_conversion(format))
1435
converter = tree.bzrdir._format.get_converter(format)
1436
result = converter.convert(tree.bzrdir, None)
1437
self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1)
1438
self.assertFalse(result.needs_format_conversion(format))
1440
def test_downgrade_to_2a_too_many_branches(self):
1441
tree = self.make_branch_and_tree('.', format='development-colo')
1442
tree.bzrdir.create_branch(name="another-colocated-branch")
1443
converter = tree.bzrdir._format.get_converter(
1444
bzrdir.BzrDirMetaFormat1())
1445
result = converter.convert(tree.bzrdir, bzrdir.BzrDirMetaFormat1())
1446
self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1)
1448
def test_nested(self):
1449
tree = self.make_branch_and_tree('.', format='development-colo')
1450
tree.bzrdir.create_branch(name='foo')
1451
tree.bzrdir.create_branch(name='fool/bla')
1453
errors.ParentBranchExists, tree.bzrdir.create_branch,
1456
def test_parent(self):
1457
tree = self.make_branch_and_tree('.', format='development-colo')
1458
tree.bzrdir.create_branch(name='fool/bla')
1459
tree.bzrdir.create_branch(name='foo/bar')
1461
errors.AlreadyBranchError, tree.bzrdir.create_branch,
1465
class SampleBzrFormat(bzrdir.BzrFormat):
1468
def get_format_string(cls):
1469
return "First line\n"
1472
class TestBzrFormat(TestCase):
1473
"""Tests for BzrFormat."""
1475
def test_as_string(self):
1476
format = SampleBzrFormat()
1477
format.features = {"foo": "required"}
1478
self.assertEqual(format.as_string(),
1481
format.features["another"] = "optional"
1482
self.assertEqual(format.as_string(),
1485
"optional another\n")
1487
def test_network_name(self):
1488
# The network string should include the feature info
1489
format = SampleBzrFormat()
1490
format.features = {"foo": "required"}
1492
"First line\nrequired foo\n",
1493
format.network_name())
1495
def test_from_string_no_features(self):
1497
format = SampleBzrFormat.from_string(
1499
self.assertEqual({}, format.features)
1501
def test_from_string_with_feature(self):
1503
format = SampleBzrFormat.from_string(
1504
"First line\nrequired foo\n")
1505
self.assertEqual("required", format.features.get("foo"))
1507
def test_from_string_format_string_mismatch(self):
1508
# The first line has to match the format string
1509
self.assertRaises(AssertionError, SampleBzrFormat.from_string,
1510
"Second line\nrequired foo\n")
1512
def test_from_string_missing_space(self):
1513
# At least one space is required in the feature lines
1514
self.assertRaises(errors.ParseFormatError, SampleBzrFormat.from_string,
1515
"First line\nfoo\n")
1517
def test_from_string_with_spaces(self):
1518
# Feature with spaces (in case we add stuff like this in the future)
1519
format = SampleBzrFormat.from_string(
1520
"First line\nrequired foo with spaces\n")
1521
self.assertEqual("required", format.features.get("foo with spaces"))
1524
format1 = SampleBzrFormat()
1525
format1.features = {"nested-trees": "optional"}
1526
format2 = SampleBzrFormat()
1527
format2.features = {"nested-trees": "optional"}
1528
self.assertEqual(format1, format1)
1529
self.assertEqual(format1, format2)
1530
format3 = SampleBzrFormat()
1531
self.assertNotEqual(format1, format3)
1533
def test_check_support_status_optional(self):
1534
# Optional, so silently ignore
1535
format = SampleBzrFormat()
1536
format.features = {"nested-trees": "optional"}
1537
format.check_support_status(True)
1538
self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
1539
SampleBzrFormat.register_feature("nested-trees")
1540
format.check_support_status(True)
1542
def test_check_support_status_required(self):
1543
# Optional, so trigger an exception
1544
format = SampleBzrFormat()
1545
format.features = {"nested-trees": "required"}
1546
self.assertRaises(errors.MissingFeature, format.check_support_status,
1548
self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
1549
SampleBzrFormat.register_feature("nested-trees")
1550
format.check_support_status(True)
1552
def test_check_support_status_unknown(self):
1553
# treat unknown necessity as required
1554
format = SampleBzrFormat()
1555
format.features = {"nested-trees": "unknown"}
1556
self.assertRaises(errors.MissingFeature, format.check_support_status,
1558
self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
1559
SampleBzrFormat.register_feature("nested-trees")
1560
format.check_support_status(True)
1562
def test_feature_already_registered(self):
1563
# a feature can only be registered once
1564
self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
1565
SampleBzrFormat.register_feature("nested-trees")
1566
self.assertRaises(errors.FeatureAlreadyRegistered,
1567
SampleBzrFormat.register_feature, "nested-trees")
1569
def test_feature_with_space(self):
1570
# spaces are not allowed in feature names
1571
self.assertRaises(ValueError, SampleBzrFormat.register_feature,
845
_qualifier = 'pycurl'
added
removed
bzrlib/tests/test_bzrdir.py
