809
807
self.assertEqualBzrdirs([baz, foo, bar],
810
808
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))
844
810
def test_find_bzrdirs_list_current(self):
845
811
def list_current(transport):
846
812
return [s for s in transport.list_dir('') if s != 'baz']
889
856
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/')
907
859
class TestMeta1DirFormat(TestCaseWithTransport):
908
860
"""Tests specific to the meta1 dir format."""
1161
1113
def create_transport_readonly_server(self):
1162
# We don't set the http protocol version, relying on the default
1163
1114
return http_utils.HTTPServerRedirecting()
1165
1116
def create_transport_secondary_server(self):
1166
# We don't set the http protocol version, relying on the default
1167
1117
return http_utils.HTTPServerRedirecting()
1169
1119
def setUp(self):
1389
1339
url = transport.base
1390
1340
err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1391
1341
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'))