~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-29 22:03:03 UTC
  • mfrom: (5416.2.6 jam-integration)
  • Revision ID: pqm@pqm.ubuntu.com-20100929220303-cr95h8iwtggco721
(mbp) Add 'break-lock --force'

Show diffs side-by-side

added added

removed removed

Lines of Context:
24
24
import sys
25
25
 
26
26
from bzrlib import (
 
27
    branch,
27
28
    bzrdir,
 
29
    controldir,
28
30
    errors,
29
31
    help_topics,
30
32
    repository,
36
38
    )
37
39
import bzrlib.branch
38
40
from bzrlib.errors import (NotBranchError,
 
41
                           NoColocatedBranchSupport,
39
42
                           UnknownFormatError,
40
43
                           UnsupportedFormatError,
41
44
                           )
53
56
from bzrlib.transport import (
54
57
    get_transport,
55
58
    memory,
 
59
    pathfilter,
56
60
    )
57
61
from bzrlib.transport.http._urllib import HttpTransport_urllib
58
62
from bzrlib.transport.nosmart import NoSmartTransportDecorator
66
70
        old_format = bzrdir.BzrDirFormat.get_default_format()
67
71
        # default is BzrDirFormat6
68
72
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
69
 
        bzrdir.BzrDirFormat._set_default_format(SampleBzrDirFormat())
 
73
        controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
70
74
        # creating a bzr dir should now create an instrumented dir.
71
75
        try:
72
76
            result = bzrdir.BzrDir.create('memory:///')
73
77
            self.failUnless(isinstance(result, SampleBzrDir))
74
78
        finally:
75
 
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
79
            controldir.ControlDirFormat._set_default_format(old_format)
76
80
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
77
81
 
78
82
 
79
83
class TestFormatRegistry(TestCase):
80
84
 
81
85
    def make_format_registry(self):
82
 
        my_format_registry = bzrdir.BzrDirFormatRegistry()
 
86
        my_format_registry = controldir.ControlDirFormatRegistry()
83
87
        my_format_registry.register('weave', bzrdir.BzrDirFormat6,
84
88
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
85
89
            ' repositories', deprecated=True)
86
90
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir',
87
91
            'BzrDirFormat6', 'Format registered lazily', deprecated=True)
88
 
        my_format_registry.register_metadir('knit',
 
92
        bzrdir.register_metadir(my_format_registry, 'knit',
89
93
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
90
94
            'Format using knits',
91
95
            )
92
96
        my_format_registry.set_default('knit')
93
 
        my_format_registry.register_metadir(
 
97
        bzrdir.register_metadir(my_format_registry,
94
98
            'branch6',
95
99
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
96
100
            'Experimental successor to knit.  Use at your own risk.',
97
101
            branch_format='bzrlib.branch.BzrBranchFormat6',
98
102
            experimental=True)
99
 
        my_format_registry.register_metadir(
 
103
        bzrdir.register_metadir(my_format_registry,
100
104
            'hidden format',
101
105
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
102
106
            'Experimental successor to knit.  Use at your own risk.',
171
175
            bzrdir.format_registry.set_default_repository(old_default)
172
176
 
173
177
    def test_aliases(self):
174
 
        a_registry = bzrdir.BzrDirFormatRegistry()
 
178
        a_registry = controldir.ControlDirFormatRegistry()
175
179
        a_registry.register('weave', bzrdir.BzrDirFormat6,
176
180
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
177
181
            ' repositories', deprecated=True)
206
210
        """See BzrDir.open_repository."""
207
211
        return SampleRepository(self)
208
212
 
209
 
    def create_branch(self):
 
213
    def create_branch(self, name=None):
210
214
        """See BzrDir.create_branch."""
 
215
        if name is not None:
 
216
            raise NoColocatedBranchSupport(self)
211
217
        return SampleBranch(self)
212
218
 
213
219
    def create_workingtree(self):
468
474
        # Make stackable source branch with an unstackable repo format.
469
475
        source_bzrdir = self.make_bzrdir('source')
470
476
        pack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
471
 
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(source_bzrdir)
 
477
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
 
478
            source_bzrdir)
472
479
        # Make a directory with a default stacking policy
473
480
        parent_bzrdir = self.make_bzrdir('parent')
474
481
        stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
803
810
        self.assertEqualBzrdirs([baz, foo, bar],
804
811
                                bzrdir.BzrDir.find_bzrdirs(transport))
805
812
 
 
813
    def make_fake_permission_denied_transport(self, transport, paths):
 
814
        """Create a transport that raises PermissionDenied for some paths."""
 
815
        def filter(path):
 
816
            if path in paths:
 
817
                raise errors.PermissionDenied(path)
 
818
            return path
 
819
        path_filter_server = pathfilter.PathFilteringServer(transport, filter)
 
820
        path_filter_server.start_server()
 
821
        self.addCleanup(path_filter_server.stop_server)
 
822
        path_filter_transport = pathfilter.PathFilteringTransport(
 
823
            path_filter_server, '.')
 
824
        return (path_filter_server, path_filter_transport)
 
825
 
 
826
    def assertBranchUrlsEndWith(self, expect_url_suffix, actual_bzrdirs):
 
827
        """Check that each branch url ends with the given suffix."""
 
828
        for actual_bzrdir in actual_bzrdirs:
 
829
            self.assertEndsWith(actual_bzrdir.user_url, expect_url_suffix)
 
830
 
 
831
    def test_find_bzrdirs_permission_denied(self):
 
832
        foo, bar, baz = self.make_foo_bar_baz()
 
833
        transport = get_transport(self.get_url())
 
834
        path_filter_server, path_filter_transport = \
 
835
            self.make_fake_permission_denied_transport(transport, ['foo'])
 
836
        # local transport
 
837
        self.assertBranchUrlsEndWith('/baz/',
 
838
            bzrdir.BzrDir.find_bzrdirs(path_filter_transport))
 
839
        # smart server
 
840
        smart_transport = self.make_smart_server('.',
 
841
            backing_server=path_filter_server)
 
842
        self.assertBranchUrlsEndWith('/baz/',
 
843
            bzrdir.BzrDir.find_bzrdirs(smart_transport))
 
844
 
806
845
    def test_find_bzrdirs_list_current(self):
807
846
        def list_current(transport):
808
847
            return [s for s in transport.list_dir('') if s != 'baz']
813
852
                                bzrdir.BzrDir.find_bzrdirs(transport,
814
853
                                    list_current=list_current))
815
854
 
816
 
 
817
855
    def test_find_bzrdirs_evaluate(self):
818
856
        def evaluate(bzrdir):
819
857
            try:
852
890
        self.assertEqual(bar.root_transport.base, branches[1].base)
853
891
 
854
892
 
 
893
class TestMissingRepoBranchesSkipped(TestCaseWithMemoryTransport):
 
894
 
 
895
    def test_find_bzrdirs_missing_repo(self):
 
896
        transport = get_transport(self.get_url())
 
897
        arepo = self.make_repository('arepo', shared=True)
 
898
        abranch_url = arepo.user_url + '/abranch'
 
899
        abranch = bzrdir.BzrDir.create(abranch_url).create_branch()
 
900
        transport.delete_tree('arepo/.bzr')
 
901
        self.assertRaises(errors.NoRepositoryPresent,
 
902
            branch.Branch.open, abranch_url)
 
903
        self.make_branch('baz')
 
904
        for actual_bzrdir in bzrdir.BzrDir.find_branches(transport):
 
905
            self.assertEndsWith(actual_bzrdir.user_url, '/baz/')
 
906
 
 
907
 
855
908
class TestMeta1DirFormat(TestCaseWithTransport):
856
909
    """Tests specific to the meta1 dir format."""
857
910
 
1006
1059
    def _known_formats(self):
1007
1060
        return set([NotBzrDirFormat()])
1008
1061
 
1009
 
    @classmethod
 
1062
 
 
1063
class NotBzrDirProber(controldir.Prober):
 
1064
 
1010
1065
    def probe_transport(self, transport):
1011
1066
        """Our format is present if the transport ends in '.not/'."""
1012
1067
        if transport.has('.not'):
1026
1081
        dir = format.initialize(self.get_url())
1027
1082
        self.assertIsInstance(dir, NotBzrDir)
1028
1083
        # now probe for it.
1029
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
 
1084
        controldir.ControlDirFormat.register_prober(NotBzrDirProber)
1030
1085
        try:
1031
1086
            found = bzrlib.bzrdir.BzrDirFormat.find_format(
1032
1087
                get_transport(self.get_url()))
1033
1088
            self.assertIsInstance(found, NotBzrDirFormat)
1034
1089
        finally:
1035
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
 
1090
            controldir.ControlDirFormat.unregister_prober(NotBzrDirProber)
1036
1091
 
1037
1092
    def test_included_in_known_formats(self):
1038
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
 
1093
        not_format = NotBzrDirFormat()
 
1094
        bzrlib.controldir.ControlDirFormat.register_format(not_format)
1039
1095
        try:
1040
1096
            formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
1041
1097
            for format in formats:
1043
1099
                    return
1044
1100
            self.fail("No NotBzrDirFormat in %s" % formats)
1045
1101
        finally:
1046
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
 
1102
            bzrlib.controldir.ControlDirFormat.unregister_format(not_format)
1047
1103
 
1048
1104
 
1049
1105
class NonLocalTests(TestCaseWithTransport):
1107
1163
    """
1108
1164
 
1109
1165
    def create_transport_readonly_server(self):
 
1166
        # We don't set the http protocol version, relying on the default
1110
1167
        return http_utils.HTTPServerRedirecting()
1111
1168
 
1112
1169
    def create_transport_secondary_server(self):
 
1170
        # We don't set the http protocol version, relying on the default
1113
1171
        return http_utils.HTTPServerRedirecting()
1114
1172
 
1115
1173
    def setUp(self):
1335
1393
        url = transport.base
1336
1394
        err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1337
1395
        self.assertEqual('fail', err._preformatted_string)
 
1396
 
 
1397
    def test_post_repo_init(self):
 
1398
        from bzrlib.bzrdir import RepoInitHookParams
 
1399
        calls = []
 
1400
        bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
 
1401
            calls.append, None)
 
1402
        self.make_repository('foo')
 
1403
        self.assertLength(1, calls)
 
1404
        params = calls[0]
 
1405
        self.assertIsInstance(params, RepoInitHookParams)
 
1406
        self.assertTrue(hasattr(params, 'bzrdir'))
 
1407
        self.assertTrue(hasattr(params, 'repository'))
 
1408
 
 
1409
    def test_post_repo_init_hook_repr(self):
 
1410
        param_reprs = []
 
1411
        bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
 
1412
            lambda params: param_reprs.append(repr(params)), None)
 
1413
        self.make_repository('foo')
 
1414
        self.assertLength(1, param_reprs)
 
1415
        param_repr = param_reprs[0]
 
1416
        self.assertStartsWith(param_repr, '<RepoInitHookParams for ')
 
1417
 
 
1418
 
 
1419
class TestGenerateBackupName(TestCaseWithMemoryTransport):
 
1420
 
 
1421
    def setUp(self):
 
1422
        super(TestGenerateBackupName, self).setUp()
 
1423
        self._transport = get_transport(self.get_url())
 
1424
        bzrdir.BzrDir.create(self.get_url(),
 
1425
            possible_transports=[self._transport])
 
1426
        self._bzrdir = bzrdir.BzrDir.open_from_transport(self._transport)
 
1427
 
 
1428
    def test_new(self):
 
1429
        self.assertEqual("a.~1~", self._bzrdir.generate_backup_name("a"))
 
1430
 
 
1431
    def test_exiting(self):
 
1432
        self._transport.put_bytes("a.~1~", "some content")
 
1433
        self.assertEqual("a.~2~", self._bzrdir.generate_backup_name("a"))