← Back to branch summary

~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-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bzrdir.py
1
 
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2006-2010 Canonical Ltd
2
2
#
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
16
16
 
17
17
"""Tests for the BzrDir facility and any format specific tests.
18
18
 
19
 
For interface contract tests, see tests/bzr_dir_implementations.
 
19
For interface contract tests, see tests/per_bzr_dir.
20
20
"""
21
21
 
22
22
import os
23
 
import os.path
24
 
from StringIO import StringIO
25
23
import subprocess
26
24
import sys
27
25
 
28
26
from bzrlib import (
 
27
    branch,
29
28
    bzrdir,
 
29
    controldir,
30
30
    errors,
31
31
    help_topics,
32
32
    repository,
33
33
    osutils,
34
 
    symbol_versioning,
35
34
    remote,
36
35
    urlutils,
37
36
    win32utils,
39
38
    )
40
39
import bzrlib.branch
41
40
from bzrlib.errors import (NotBranchError,
 
41
                           NoColocatedBranchSupport,
42
42
                           UnknownFormatError,
43
43
                           UnsupportedFormatError,
44
44
                           )
47
47
    TestCaseWithMemoryTransport,
48
48
    TestCaseWithTransport,
49
49
    TestSkipped,
50
 
    test_sftp_transport
51
50
    )
52
51
from bzrlib.tests import(
53
52
    http_server,
54
53
    http_utils,
55
54
    )
56
55
from bzrlib.tests.test_http import TestWithTransport_pycurl
57
 
from bzrlib.transport import get_transport
 
56
from bzrlib.transport import (
 
57
    get_transport,
 
58
    memory,
 
59
    pathfilter,
 
60
    )
58
61
from bzrlib.transport.http._urllib import HttpTransport_urllib
59
 
from bzrlib.transport.memory import MemoryServer
60
62
from bzrlib.transport.nosmart import NoSmartTransportDecorator
61
63
from bzrlib.transport.readonly import ReadonlyTransportDecorator
62
64
from bzrlib.repofmt import knitrepo, weaverepo, pack_repo
68
70
        old_format = bzrdir.BzrDirFormat.get_default_format()
69
71
        # default is BzrDirFormat6
70
72
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
71
 
        bzrdir.BzrDirFormat._set_default_format(SampleBzrDirFormat())
 
73
        controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
72
74
        # creating a bzr dir should now create an instrumented dir.
73
75
        try:
74
76
            result = bzrdir.BzrDir.create('memory:///')
75
77
            self.failUnless(isinstance(result, SampleBzrDir))
76
78
        finally:
77
 
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
79
            controldir.ControlDirFormat._set_default_format(old_format)
78
80
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
79
81
 
80
82
 
81
83
class TestFormatRegistry(TestCase):
82
84
 
83
85
    def make_format_registry(self):
84
 
        my_format_registry = bzrdir.BzrDirFormatRegistry()
 
86
        my_format_registry = controldir.ControlDirFormatRegistry()
85
87
        my_format_registry.register('weave', bzrdir.BzrDirFormat6,
86
88
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
87
89
            ' repositories', deprecated=True)
88
90
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir',
89
91
            'BzrDirFormat6', 'Format registered lazily', deprecated=True)
90
 
        my_format_registry.register_metadir('knit',
 
92
        bzrdir.register_metadir(my_format_registry, 'knit',
91
93
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
92
94
            'Format using knits',
93
95
            )
94
96
        my_format_registry.set_default('knit')
95
 
        my_format_registry.register_metadir(
 
97
        bzrdir.register_metadir(my_format_registry,
96
98
            'branch6',
97
99
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
98
100
            'Experimental successor to knit.  Use at your own risk.',
99
101
            branch_format='bzrlib.branch.BzrBranchFormat6',
100
102
            experimental=True)
101
 
        my_format_registry.register_metadir(
 
103
        bzrdir.register_metadir(my_format_registry,
102
104
            'hidden format',
103
105
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
104
106
            'Experimental successor to knit.  Use at your own risk.',
149
151
        new = topics.get_detail('current-formats')
150
152
        rest = topics.get_detail('other-formats')
151
153
        experimental, deprecated = rest.split('Deprecated formats')
152
 
        self.assertContainsRe(new, 'bzr help formats')
 
154
        self.assertContainsRe(new, 'formats-help')
153
155
        self.assertContainsRe(new,
154
156
                ':knit:\n    \(native\) \(default\) Format using knits\n')
155
157
        self.assertContainsRe(experimental,
173
175
            bzrdir.format_registry.set_default_repository(old_default)
174
176
 
175
177
    def test_aliases(self):
176
 
        a_registry = bzrdir.BzrDirFormatRegistry()
 
178
        a_registry = controldir.ControlDirFormatRegistry()
177
179
        a_registry.register('weave', bzrdir.BzrDirFormat6,
178
180
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
179
181
            ' repositories', deprecated=True)
208
210
        """See BzrDir.open_repository."""
209
211
        return SampleRepository(self)
210
212
 
211
 
    def create_branch(self):
 
213
    def create_branch(self, name=None):
212
214
        """See BzrDir.create_branch."""
 
215
        if name is not None:
 
216
            raise NoColocatedBranchSupport(self)
213
217
        return SampleBranch(self)
214
218
 
215
219
    def create_workingtree(self):
358
362
 
359
363
    def test_create_branch_convenience_root(self):
360
364
        """Creating a branch at the root of a fs should work."""
361
 
        self.vfs_transport_factory = MemoryServer
 
365
        self.vfs_transport_factory = memory.MemoryServer
362
366
        # outside a repo the default convenience output is a repo+branch_tree
363
367
        format = bzrdir.format_registry.make_bzrdir('knit')
364
368
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
470
474
        # Make stackable source branch with an unstackable repo format.
471
475
        source_bzrdir = self.make_bzrdir('source')
472
476
        pack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
473
 
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(source_bzrdir)
 
477
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
 
478
            source_bzrdir)
474
479
        # Make a directory with a default stacking policy
475
480
        parent_bzrdir = self.make_bzrdir('parent')
476
481
        stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
569
574
 
570
575
    def setUp(self):
571
576
        super(ChrootedTests, self).setUp()
572
 
        if not self.vfs_transport_factory == MemoryServer:
 
577
        if not self.vfs_transport_factory == memory.MemoryServer:
573
578
            self.transport_readonly_server = http_server.HttpServer
574
579
 
575
580
    def local_branch_path(self, branch):
805
810
        self.assertEqualBzrdirs([baz, foo, bar],
806
811
                                bzrdir.BzrDir.find_bzrdirs(transport))
807
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
 
808
845
    def test_find_bzrdirs_list_current(self):
809
846
        def list_current(transport):
810
847
            return [s for s in transport.list_dir('') if s != 'baz']
815
852
                                bzrdir.BzrDir.find_bzrdirs(transport,
816
853
                                    list_current=list_current))
817
854
 
818
 
 
819
855
    def test_find_bzrdirs_evaluate(self):
820
856
        def evaluate(bzrdir):
821
857
            try:
854
890
        self.assertEqual(bar.root_transport.base, branches[1].base)
855
891
 
856
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
 
857
908
class TestMeta1DirFormat(TestCaseWithTransport):
858
909
    """Tests specific to the meta1 dir format."""
859
910
 
1008
1059
    def _known_formats(self):
1009
1060
        return set([NotBzrDirFormat()])
1010
1061
 
1011
 
    @classmethod
 
1062
 
 
1063
class NotBzrDirProber(controldir.Prober):
 
1064
 
1012
1065
    def probe_transport(self, transport):
1013
1066
        """Our format is present if the transport ends in '.not/'."""
1014
1067
        if transport.has('.not'):
1028
1081
        dir = format.initialize(self.get_url())
1029
1082
        self.assertIsInstance(dir, NotBzrDir)
1030
1083
        # now probe for it.
1031
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
 
1084
        controldir.ControlDirFormat.register_prober(NotBzrDirProber)
1032
1085
        try:
1033
1086
            found = bzrlib.bzrdir.BzrDirFormat.find_format(
1034
1087
                get_transport(self.get_url()))
1035
1088
            self.assertIsInstance(found, NotBzrDirFormat)
1036
1089
        finally:
1037
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
 
1090
            controldir.ControlDirFormat.unregister_prober(NotBzrDirProber)
1038
1091
 
1039
1092
    def test_included_in_known_formats(self):
1040
 
        bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
 
1093
        not_format = NotBzrDirFormat()
 
1094
        bzrlib.controldir.ControlDirFormat.register_format(not_format)
1041
1095
        try:
1042
1096
            formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
1043
1097
            for format in formats:
1045
1099
                    return
1046
1100
            self.fail("No NotBzrDirFormat in %s" % formats)
1047
1101
        finally:
1048
 
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
 
1102
            bzrlib.controldir.ControlDirFormat.unregister_format(not_format)
1049
1103
 
1050
1104
 
1051
1105
class NonLocalTests(TestCaseWithTransport):
1053
1107
 
1054
1108
    def setUp(self):
1055
1109
        super(NonLocalTests, self).setUp()
1056
 
        self.vfs_transport_factory = MemoryServer
 
1110
        self.vfs_transport_factory = memory.MemoryServer
1057
1111
 
1058
1112
    def test_create_branch_convenience(self):
1059
1113
        # outside a repo the default convenience output is a repo+branch_tree
1109
1163
    """
1110
1164
 
1111
1165
    def create_transport_readonly_server(self):
 
1166
        # We don't set the http protocol version, relying on the default
1112
1167
        return http_utils.HTTPServerRedirecting()
1113
1168
 
1114
1169
    def create_transport_secondary_server(self):
 
1170
        # We don't set the http protocol version, relying on the default
1115
1171
        return http_utils.HTTPServerRedirecting()
1116
1172
 
1117
1173
    def setUp(self):
1156
1212
    _transport = HttpTransport_urllib
1157
1213
 
1158
1214
    def _qualified_url(self, host, port):
1159
 
        return 'http+urllib://%s:%s' % (host, port)
 
1215
        result = 'http+urllib://%s:%s' % (host, port)
 
1216
        self.permit_url(result)
 
1217
        return result
1160
1218
 
1161
1219
 
1162
1220
 
1166
1224
    """Tests redirections for pycurl implementation"""
1167
1225
 
1168
1226
    def _qualified_url(self, host, port):
1169
 
        return 'http+pycurl://%s:%s' % (host, port)
 
1227
        result = 'http+pycurl://%s:%s' % (host, port)
 
1228
        self.permit_url(result)
 
1229
        return result
1170
1230
 
1171
1231
 
1172
1232
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
1176
1236
    _transport = NoSmartTransportDecorator
1177
1237
 
1178
1238
    def _qualified_url(self, host, port):
1179
 
        return 'nosmart+http://%s:%s' % (host, port)
 
1239
        result = 'nosmart+http://%s:%s' % (host, port)
 
1240
        self.permit_url(result)
 
1241
        return result
1180
1242
 
1181
1243
 
1182
1244
class TestHTTPRedirections_readonly(TestHTTPRedirections,
1186
1248
    _transport = ReadonlyTransportDecorator
1187
1249
 
1188
1250
    def _qualified_url(self, host, port):
1189
 
        return 'readonly+http://%s:%s' % (host, port)
 
1251
        result = 'readonly+http://%s:%s' % (host, port)
 
1252
        self.permit_url(result)
 
1253
        return result
1190
1254
 
1191
1255
 
1192
1256
class TestDotBzrHidden(TestCaseWithTransport):
1329
1393
        url = transport.base
1330
1394
        err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
1331
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"))