← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

Add bzrlib.pyutils, which has get_named_object, a wrapper around __import__.

This is used to replace various ad hoc implementations of the same logic,
notably the version used in registry's _LazyObjectGetter which had a bug when
getting a module without also getting a member.  And of course, this new
function has unit tests, unlike the replaced code.

This also adds a KnownHooksRegistry subclass to provide a more natural home for
some other logic.

I'm not thrilled about the name of the new module or the new functions, but it's
hard to think of good names for such generic functionality.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bzrdir.py
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,
54
56
from bzrlib.transport import (
55
57
    get_transport,
56
58
    memory,
 
59
    pathfilter,
57
60
    )
58
61
from bzrlib.transport.http._urllib import HttpTransport_urllib
59
62
from bzrlib.transport.nosmart import NoSmartTransportDecorator
67
70
        old_format = bzrdir.BzrDirFormat.get_default_format()
68
71
        # default is BzrDirFormat6
69
72
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
70
 
        bzrdir.BzrDirFormat._set_default_format(SampleBzrDirFormat())
 
73
        controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
71
74
        # creating a bzr dir should now create an instrumented dir.
72
75
        try:
73
76
            result = bzrdir.BzrDir.create('memory:///')
74
77
            self.failUnless(isinstance(result, SampleBzrDir))
75
78
        finally:
76
 
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
79
            controldir.ControlDirFormat._set_default_format(old_format)
77
80
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
78
81
 
79
82
 
80
83
class TestFormatRegistry(TestCase):
81
84
 
82
85
    def make_format_registry(self):
83
 
        my_format_registry = bzrdir.BzrDirFormatRegistry()
 
86
        my_format_registry = controldir.ControlDirFormatRegistry()
84
87
        my_format_registry.register('weave', bzrdir.BzrDirFormat6,
85
88
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
86
89
            ' repositories', deprecated=True)
87
90
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir',
88
91
            'BzrDirFormat6', 'Format registered lazily', deprecated=True)
89
 
        my_format_registry.register_metadir('knit',
 
92
        bzrdir.register_metadir(my_format_registry, 'knit',
90
93
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
91
94
            'Format using knits',
92
95
            )
93
96
        my_format_registry.set_default('knit')
94
 
        my_format_registry.register_metadir(
 
97
        bzrdir.register_metadir(my_format_registry,
95
98
            'branch6',
96
99
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
97
100
            'Experimental successor to knit.  Use at your own risk.',
98
101
            branch_format='bzrlib.branch.BzrBranchFormat6',
99
102
            experimental=True)
100
 
        my_format_registry.register_metadir(
 
103
        bzrdir.register_metadir(my_format_registry,
101
104
            'hidden format',
102
105
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
103
106
            'Experimental successor to knit.  Use at your own risk.',
172
175
            bzrdir.format_registry.set_default_repository(old_default)
173
176
 
174
177
    def test_aliases(self):
175
 
        a_registry = bzrdir.BzrDirFormatRegistry()
 
178
        a_registry = controldir.ControlDirFormatRegistry()
176
179
        a_registry.register('weave', bzrdir.BzrDirFormat6,
177
180
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
178
181
            ' repositories', deprecated=True)
807
810
        self.assertEqualBzrdirs([baz, foo, bar],
808
811
                                bzrdir.BzrDir.find_bzrdirs(transport))
809
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
 
810
845
    def test_find_bzrdirs_list_current(self):
811
846
        def list_current(transport):
812
847
            return [s for s in transport.list_dir('') if s != 'baz']
817
852
                                bzrdir.BzrDir.find_bzrdirs(transport,
818
853
                                    list_current=list_current))
819
854
 
820
 
 
821
855
    def test_find_bzrdirs_evaluate(self):
822
856
        def evaluate(bzrdir):
823
857
            try:
856
890
        self.assertEqual(bar.root_transport.base, branches[1].base)
857
891
 
858
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
 
859
908
class TestMeta1DirFormat(TestCaseWithTransport):
860
909
    """Tests specific to the meta1 dir format."""
861
910
 
1010
1059
    def _known_formats(self):
1011
1060
        return set([NotBzrDirFormat()])
1012
1061
 
1013
 
    @classmethod
 
1062
 
 
1063
class NotBzrDirProber(controldir.Prober):
 
1064
 
1014
1065
    def probe_transport(self, transport):
1015
1066
        """Our format is present if the transport ends in '.not/'."""
1016
1067
        if transport.has('.not'):
1030
1081
        dir = format.initialize(self.get_url())
1031
1082
        self.assertIsInstance(dir, NotBzrDir)
1032
1083
        # now probe for it.
1033
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
 
1084
        controldir.ControlDirFormat.register_prober(NotBzrDirProber)
1034
1085
        try:
1035
1086
            found = bzrlib.bzrdir.BzrDirFormat.find_format(
1036
1087
                get_transport(self.get_url()))
1037
1088
            self.assertIsInstance(found, NotBzrDirFormat)
1038
1089
        finally:
1039
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
 
1090
            controldir.ControlDirFormat.unregister_prober(NotBzrDirProber)
1040
1091
 
1041
1092
    def test_included_in_known_formats(self):
1042
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
 
1093
        not_format = NotBzrDirFormat()
 
1094
        bzrlib.controldir.ControlDirFormat.register_format(not_format)
1043
1095
        try:
1044
1096
            formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
1045
1097
            for format in formats:
1047
1099
                    return
1048
1100
            self.fail("No NotBzrDirFormat in %s" % formats)
1049
1101
        finally:
1050
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
 
1102
            bzrlib.controldir.ControlDirFormat.unregister_format(not_format)
1051
1103
 
1052
1104
 
1053
1105
class NonLocalTests(TestCaseWithTransport):
1111
1163
    """
1112
1164
 
1113
1165
    def create_transport_readonly_server(self):
 
1166
        # We don't set the http protocol version, relying on the default
1114
1167
        return http_utils.HTTPServerRedirecting()
1115
1168
 
1116
1169
    def create_transport_secondary_server(self):
 
1170
        # We don't set the http protocol version, relying on the default
1117
1171
        return http_utils.HTTPServerRedirecting()
1118
1172
 
1119
1173
    def setUp(self):
1351
1405
        self.assertIsInstance(params, RepoInitHookParams)
1352
1406
        self.assertTrue(hasattr(params, 'bzrdir'))
1353
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"))