1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
1
# Copyright (C) 2006-2010 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
17
17
"""Tests for the BzrDir facility and any format specific tests.
19
For interface contract tests, see tests/bzr_dir_implementations.
19
For interface contract tests, see tests/per_bzr_dir.
24
from StringIO import StringIO
28
26
from bzrlib import (
47
46
TestCaseWithMemoryTransport,
48
47
TestCaseWithTransport,
52
50
from bzrlib.tests import(
56
54
from bzrlib.tests.test_http import TestWithTransport_pycurl
57
from bzrlib.transport import get_transport
55
from bzrlib.transport import (
58
60
from bzrlib.transport.http._urllib import HttpTransport_urllib
59
from bzrlib.transport.memory import MemoryServer
60
61
from bzrlib.transport.nosmart import NoSmartTransportDecorator
61
62
from bzrlib.transport.readonly import ReadonlyTransportDecorator
62
63
from bzrlib.repofmt import knitrepo, weaverepo, pack_repo
149
150
new = topics.get_detail('current-formats')
150
151
rest = topics.get_detail('other-formats')
151
152
experimental, deprecated = rest.split('Deprecated formats')
152
self.assertContainsRe(new, 'bzr help formats')
153
self.assertContainsRe(new, 'formats-help')
153
154
self.assertContainsRe(new,
154
155
':knit:\n \(native\) \(default\) Format using knits\n')
155
156
self.assertContainsRe(experimental,
208
209
"""See BzrDir.open_repository."""
209
210
return SampleRepository(self)
211
def create_branch(self):
212
def create_branch(self, name=None):
212
213
"""See BzrDir.create_branch."""
215
raise NoColocatedBranchSupport(self)
213
216
return SampleBranch(self)
215
218
def create_workingtree(self):
359
362
def test_create_branch_convenience_root(self):
360
363
"""Creating a branch at the root of a fs should work."""
361
self.vfs_transport_factory = MemoryServer
364
self.vfs_transport_factory = memory.MemoryServer
362
365
# outside a repo the default convenience output is a repo+branch_tree
363
366
format = bzrdir.format_registry.make_bzrdir('knit')
364
367
branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
470
473
# Make stackable source branch with an unstackable repo format.
471
474
source_bzrdir = self.make_bzrdir('source')
472
475
pack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
473
source_branch = bzrlib.branch.BzrBranchFormat7().initialize(source_bzrdir)
476
source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
474
478
# Make a directory with a default stacking policy
475
479
parent_bzrdir = self.make_bzrdir('parent')
476
480
stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
571
575
super(ChrootedTests, self).setUp()
572
if not self.vfs_transport_factory == MemoryServer:
576
if not self.vfs_transport_factory == memory.MemoryServer:
573
577
self.transport_readonly_server = http_server.HttpServer
575
579
def local_branch_path(self, branch):
805
809
self.assertEqualBzrdirs([baz, foo, bar],
806
810
bzrdir.BzrDir.find_bzrdirs(transport))
812
def make_fake_permission_denied_transport(self, transport, paths):
813
"""Create a transport that raises PermissionDenied for some paths."""
816
raise errors.PermissionDenied(path)
818
path_filter_server = pathfilter.PathFilteringServer(transport, filter)
819
path_filter_server.start_server()
820
self.addCleanup(path_filter_server.stop_server)
821
path_filter_transport = pathfilter.PathFilteringTransport(
822
path_filter_server, '.')
823
return (path_filter_server, path_filter_transport)
825
def assertBranchUrlsEndWith(self, expect_url_suffix, actual_bzrdirs):
826
"""Check that each branch url ends with the given suffix."""
827
for actual_bzrdir in actual_bzrdirs:
828
self.assertEndsWith(actual_bzrdir.user_url, expect_url_suffix)
830
def test_find_bzrdirs_permission_denied(self):
831
foo, bar, baz = self.make_foo_bar_baz()
832
transport = get_transport(self.get_url())
833
path_filter_server, path_filter_transport = \
834
self.make_fake_permission_denied_transport(transport, ['foo'])
836
self.assertBranchUrlsEndWith('/baz/',
837
bzrdir.BzrDir.find_bzrdirs(path_filter_transport))
839
smart_transport = self.make_smart_server('.',
840
backing_server=path_filter_server)
841
self.assertBranchUrlsEndWith('/baz/',
842
bzrdir.BzrDir.find_bzrdirs(smart_transport))
808
844
def test_find_bzrdirs_list_current(self):
809
845
def list_current(transport):
810
846
return [s for s in transport.list_dir('') if s != 'baz']
854
889
self.assertEqual(bar.root_transport.base, branches[1].base)
892
class TestMissingRepoBranchesSkipped(TestCaseWithMemoryTransport):
894
def test_find_bzrdirs_missing_repo(self):
895
transport = get_transport(self.get_url())
896
arepo = self.make_repository('arepo', shared=True)
897
abranch_url = arepo.user_url + '/abranch'
898
abranch = bzrdir.BzrDir.create(abranch_url).create_branch()
899
transport.delete_tree('arepo/.bzr')
900
self.assertRaises(errors.NoRepositoryPresent,
901
branch.Branch.open, abranch_url)
902
self.make_branch('baz')
903
for actual_bzrdir in bzrdir.BzrDir.find_branches(transport):
904
self.assertEndsWith(actual_bzrdir.user_url, '/baz/')
857
907
class TestMeta1DirFormat(TestCaseWithTransport):
858
908
"""Tests specific to the meta1 dir format."""
1054
1104
def setUp(self):
1055
1105
super(NonLocalTests, self).setUp()
1056
self.vfs_transport_factory = MemoryServer
1106
self.vfs_transport_factory = memory.MemoryServer
1058
1108
def test_create_branch_convenience(self):
1059
1109
# outside a repo the default convenience output is a repo+branch_tree
1111
1161
def create_transport_readonly_server(self):
1162
# We don't set the http protocol version, relying on the default
1112
1163
return http_utils.HTTPServerRedirecting()
1114
1165
def create_transport_secondary_server(self):
1166
# We don't set the http protocol version, relying on the default
1115
1167
return http_utils.HTTPServerRedirecting()
1117
1169
def setUp(self):
1156
1208
_transport = HttpTransport_urllib
1158
1210
def _qualified_url(self, host, port):
1159
return 'http+urllib://%s:%s' % (host, port)
1211
result = 'http+urllib://%s:%s' % (host, port)
1212
self.permit_url(result)
1166
1220
"""Tests redirections for pycurl implementation"""
1168
1222
def _qualified_url(self, host, port):
1169
return 'http+pycurl://%s:%s' % (host, port)
1223
result = 'http+pycurl://%s:%s' % (host, port)
1224
self.permit_url(result)
1172
1228
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1176
1232
_transport = NoSmartTransportDecorator
1178
1234
def _qualified_url(self, host, port):
1179
return 'nosmart+http://%s:%s' % (host, port)
1235
result = 'nosmart+http://%s:%s' % (host, port)
1236
self.permit_url(result)
1182
1240
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1186
1244
_transport = ReadonlyTransportDecorator
1188
1246
def _qualified_url(self, host, port):
1189
return 'readonly+http://%s:%s' % (host, port)
1247
result = 'readonly+http://%s:%s' % (host, port)
1248
self.permit_url(result)
1192
1252
class TestDotBzrHidden(TestCaseWithTransport):
1329
1389
url = transport.base
1330
1390
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1331
1391
self.assertEqual('fail', err._preformatted_string)
1393
def test_post_repo_init(self):
1394
from bzrlib.bzrdir import RepoInitHookParams
1396
bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
1398
self.make_repository('foo')
1399
self.assertLength(1, calls)
1401
self.assertIsInstance(params, RepoInitHookParams)
1402
self.assertTrue(hasattr(params, 'bzrdir'))
1403
self.assertTrue(hasattr(params, 'repository'))