← 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: John Arbash Meinel
  • Date: 2009-03-30 18:42:38 UTC
  • mto: (3735.2.171 brisbane-core)
  • mto: This revision was merged to the branch mainline in revision 4280.
  • Revision ID: john@arbash-meinel.com-20090330184238-yc8t2i9wr2mjkxiu
For now, we handle the case of a root node being referenced in the unwanted set.
However, it requires buffering, which is something we want to get rid of.

Show diffs side-by-side

added added

removed removed

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