← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

(vila) Make all transport put_bytes() raises TypeError when given unicode
 strings rather than bytes (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bzrdir.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
2
 
# 
 
1
# Copyright (C) 2006-2011 Canonical Ltd
 
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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
# 
 
7
#
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
# 
 
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
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
 
from StringIO import StringIO
 
22
import os
 
23
import subprocess
 
24
import sys
23
25
 
 
26
from bzrlib import (
 
27
    branch,
 
28
    bzrdir,
 
29
    config,
 
30
    controldir,
 
31
    errors,
 
32
    help_topics,
 
33
    lock,
 
34
    repository,
 
35
    revision as _mod_revision,
 
36
    osutils,
 
37
    remote,
 
38
    transport as _mod_transport,
 
39
    urlutils,
 
40
    win32utils,
 
41
    workingtree_3,
 
42
    workingtree_4,
 
43
    )
24
44
import bzrlib.branch
25
 
import bzrlib.bzrdir as bzrdir
26
 
import bzrlib.errors as errors
27
 
from bzrlib.errors import (NotBranchError,
28
 
                           UnknownFormatError,
29
 
                           UnsupportedFormatError,
30
 
                           )
31
 
import bzrlib.repository as repository
32
 
from bzrlib.tests import TestCase, TestCaseWithTransport
33
 
from bzrlib.transport import get_transport
34
 
from bzrlib.transport.http import HttpServer
35
 
from bzrlib.transport.memory import MemoryServer
36
 
import bzrlib.workingtree as workingtree
 
45
from bzrlib.branchfmt.fullhistory import BzrBranchFormat5
 
46
from bzrlib.errors import (
 
47
    NotBranchError,
 
48
    NoColocatedBranchSupport,
 
49
    UnknownFormatError,
 
50
    UnsupportedFormatError,
 
51
    )
 
52
from bzrlib.tests import (
 
53
    TestCase,
 
54
    TestCaseWithMemoryTransport,
 
55
    TestCaseWithTransport,
 
56
    TestSkipped,
 
57
    )
 
58
from bzrlib.tests import(
 
59
    http_server,
 
60
    http_utils,
 
61
    )
 
62
from bzrlib.tests.test_http import TestWithTransport_pycurl
 
63
from bzrlib.transport import (
 
64
    memory,
 
65
    pathfilter,
 
66
    )
 
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
68
from bzrlib.transport.nosmart import NoSmartTransportDecorator
 
69
from bzrlib.transport.readonly import ReadonlyTransportDecorator
 
70
from bzrlib.repofmt import knitrepo, knitpack_repo
37
71
 
38
72
 
39
73
class TestDefaultFormat(TestCase):
40
74
 
41
75
    def test_get_set_default_format(self):
42
76
        old_format = bzrdir.BzrDirFormat.get_default_format()
43
 
        # default is BzrDirFormat6
44
 
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
45
 
        bzrdir.BzrDirFormat.set_default_format(SampleBzrDirFormat())
 
77
        # default is BzrDirMetaFormat1
 
78
        self.assertIsInstance(old_format, bzrdir.BzrDirMetaFormat1)
 
79
        controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
46
80
        # creating a bzr dir should now create an instrumented dir.
47
81
        try:
48
 
            result = bzrdir.BzrDir.create('memory:/')
49
 
            self.failUnless(isinstance(result, SampleBzrDir))
 
82
            result = bzrdir.BzrDir.create('memory:///')
 
83
            self.assertIsInstance(result, SampleBzrDir)
50
84
        finally:
51
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
85
            controldir.ControlDirFormat._set_default_format(old_format)
52
86
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
53
87
 
54
88
 
 
89
class DeprecatedBzrDirFormat(bzrdir.BzrDirFormat):
 
90
    """A deprecated bzr dir format."""
 
91
 
 
92
 
 
93
class TestFormatRegistry(TestCase):
 
94
 
 
95
    def make_format_registry(self):
 
96
        my_format_registry = controldir.ControlDirFormatRegistry()
 
97
        my_format_registry.register('deprecated', DeprecatedBzrDirFormat,
 
98
            'Some format.  Slower and unawesome and deprecated.',
 
99
            deprecated=True)
 
100
        my_format_registry.register_lazy('lazy', 'bzrlib.tests.test_bzrdir',
 
101
            'DeprecatedBzrDirFormat', 'Format registered lazily',
 
102
            deprecated=True)
 
103
        bzrdir.register_metadir(my_format_registry, 'knit',
 
104
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
 
105
            'Format using knits',
 
106
            )
 
107
        my_format_registry.set_default('knit')
 
108
        bzrdir.register_metadir(my_format_registry,
 
109
            'branch6',
 
110
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
 
111
            'Experimental successor to knit.  Use at your own risk.',
 
112
            branch_format='bzrlib.branch.BzrBranchFormat6',
 
113
            experimental=True)
 
114
        bzrdir.register_metadir(my_format_registry,
 
115
            'hidden format',
 
116
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
 
117
            'Experimental successor to knit.  Use at your own risk.',
 
118
            branch_format='bzrlib.branch.BzrBranchFormat6', hidden=True)
 
119
        my_format_registry.register('hiddendeprecated', DeprecatedBzrDirFormat,
 
120
            'Old format.  Slower and does not support things. ', hidden=True)
 
121
        my_format_registry.register_lazy('hiddenlazy', 'bzrlib.tests.test_bzrdir',
 
122
            'DeprecatedBzrDirFormat', 'Format registered lazily',
 
123
            deprecated=True, hidden=True)
 
124
        return my_format_registry
 
125
 
 
126
    def test_format_registry(self):
 
127
        my_format_registry = self.make_format_registry()
 
128
        my_bzrdir = my_format_registry.make_bzrdir('lazy')
 
129
        self.assertIsInstance(my_bzrdir, DeprecatedBzrDirFormat)
 
130
        my_bzrdir = my_format_registry.make_bzrdir('deprecated')
 
131
        self.assertIsInstance(my_bzrdir, DeprecatedBzrDirFormat)
 
132
        my_bzrdir = my_format_registry.make_bzrdir('default')
 
133
        self.assertIsInstance(my_bzrdir.repository_format,
 
134
            knitrepo.RepositoryFormatKnit1)
 
135
        my_bzrdir = my_format_registry.make_bzrdir('knit')
 
136
        self.assertIsInstance(my_bzrdir.repository_format,
 
137
            knitrepo.RepositoryFormatKnit1)
 
138
        my_bzrdir = my_format_registry.make_bzrdir('branch6')
 
139
        self.assertIsInstance(my_bzrdir.get_branch_format(),
 
140
                              bzrlib.branch.BzrBranchFormat6)
 
141
 
 
142
    def test_get_help(self):
 
143
        my_format_registry = self.make_format_registry()
 
144
        self.assertEqual('Format registered lazily',
 
145
                         my_format_registry.get_help('lazy'))
 
146
        self.assertEqual('Format using knits',
 
147
                         my_format_registry.get_help('knit'))
 
148
        self.assertEqual('Format using knits',
 
149
                         my_format_registry.get_help('default'))
 
150
        self.assertEqual('Some format.  Slower and unawesome and deprecated.',
 
151
                         my_format_registry.get_help('deprecated'))
 
152
 
 
153
    def test_help_topic(self):
 
154
        topics = help_topics.HelpTopicRegistry()
 
155
        registry = self.make_format_registry()
 
156
        topics.register('current-formats', registry.help_topic,
 
157
                        'Current formats')
 
158
        topics.register('other-formats', registry.help_topic,
 
159
                        'Other formats')
 
160
        new = topics.get_detail('current-formats')
 
161
        rest = topics.get_detail('other-formats')
 
162
        experimental, deprecated = rest.split('Deprecated formats')
 
163
        self.assertContainsRe(new, 'formats-help')
 
164
        self.assertContainsRe(new,
 
165
                ':knit:\n    \(native\) \(default\) Format using knits\n')
 
166
        self.assertContainsRe(experimental,
 
167
                ':branch6:\n    \(native\) Experimental successor to knit')
 
168
        self.assertContainsRe(deprecated,
 
169
                ':lazy:\n    \(native\) Format registered lazily\n')
 
170
        self.assertNotContainsRe(new, 'hidden')
 
171
 
 
172
    def test_set_default_repository(self):
 
173
        default_factory = controldir.format_registry.get('default')
 
174
        old_default = [k for k, v in controldir.format_registry.iteritems()
 
175
                       if v == default_factory and k != 'default'][0]
 
176
        controldir.format_registry.set_default_repository('dirstate-with-subtree')
 
177
        try:
 
178
            self.assertIs(controldir.format_registry.get('dirstate-with-subtree'),
 
179
                          controldir.format_registry.get('default'))
 
180
            self.assertIs(
 
181
                repository.format_registry.get_default().__class__,
 
182
                knitrepo.RepositoryFormatKnit3)
 
183
        finally:
 
184
            controldir.format_registry.set_default_repository(old_default)
 
185
 
 
186
    def test_aliases(self):
 
187
        a_registry = controldir.ControlDirFormatRegistry()
 
188
        a_registry.register('deprecated', DeprecatedBzrDirFormat,
 
189
            'Old format.  Slower and does not support stuff',
 
190
            deprecated=True)
 
191
        a_registry.register('deprecatedalias', DeprecatedBzrDirFormat,
 
192
            'Old format.  Slower and does not support stuff',
 
193
            deprecated=True, alias=True)
 
194
        self.assertEqual(frozenset(['deprecatedalias']), a_registry.aliases())
 
195
 
 
196
 
55
197
class SampleBranch(bzrlib.branch.Branch):
56
198
    """A dummy branch for guess what, dummy use."""
57
199
 
59
201
        self.bzrdir = dir
60
202
 
61
203
 
 
204
class SampleRepository(bzrlib.repository.Repository):
 
205
    """A dummy repo."""
 
206
 
 
207
    def __init__(self, dir):
 
208
        self.bzrdir = dir
 
209
 
 
210
 
62
211
class SampleBzrDir(bzrdir.BzrDir):
63
212
    """A sample BzrDir implementation to allow testing static methods."""
64
213
 
65
 
    def create_repository(self):
66
 
        """See BzrDir.create_repository."""
 
214
    def create_repository(self, shared=False):
 
215
        """See ControlDir.create_repository."""
67
216
        return "A repository"
68
217
 
69
218
    def open_repository(self):
70
 
        """See BzrDir.open_repository."""
71
 
        return "A repository"
 
219
        """See ControlDir.open_repository."""
 
220
        return SampleRepository(self)
72
221
 
73
 
    def create_branch(self):
74
 
        """See BzrDir.create_branch."""
 
222
    def create_branch(self, name=None):
 
223
        """See ControlDir.create_branch."""
 
224
        if name is not None:
 
225
            raise NoColocatedBranchSupport(self)
75
226
        return SampleBranch(self)
76
227
 
77
228
    def create_workingtree(self):
78
 
        """See BzrDir.create_workingtree."""
 
229
        """See ControlDir.create_workingtree."""
79
230
        return "A tree"
80
231
 
81
232
 
82
233
class SampleBzrDirFormat(bzrdir.BzrDirFormat):
83
234
    """A sample format
84
235
 
85
 
    this format is initializable, unsupported to aid in testing the 
 
236
    this format is initializable, unsupported to aid in testing the
86
237
    open and open_downlevel routines.
87
238
    """
88
239
 
90
241
        """See BzrDirFormat.get_format_string()."""
91
242
        return "Sample .bzr dir format."
92
243
 
93
 
    def initialize(self, url):
 
244
    def initialize_on_transport(self, t):
94
245
        """Create a bzr dir."""
95
 
        t = get_transport(url)
96
246
        t.mkdir('.bzr')
97
 
        t.put('.bzr/branch-format', StringIO(self.get_format_string()))
 
247
        t.put_bytes('.bzr/branch-format', self.get_format_string())
98
248
        return SampleBzrDir(t, self)
99
249
 
100
250
    def is_supported(self):
103
253
    def open(self, transport, _found=None):
104
254
        return "opened branch."
105
255
 
 
256
    @classmethod
 
257
    def from_string(cls, format_string):
 
258
        return cls()
 
259
 
 
260
 
 
261
class BzrDirFormatTest1(bzrdir.BzrDirMetaFormat1):
 
262
 
 
263
    @staticmethod
 
264
    def get_format_string():
 
265
        return "Test format 1"
 
266
 
 
267
 
 
268
class BzrDirFormatTest2(bzrdir.BzrDirMetaFormat1):
 
269
 
 
270
    @staticmethod
 
271
    def get_format_string():
 
272
        return "Test format 2"
 
273
 
106
274
 
107
275
class TestBzrDirFormat(TestCaseWithTransport):
108
276
    """Tests for the BzrDirFormat facility."""
110
278
    def test_find_format(self):
111
279
        # is the right format object found for a branch?
112
280
        # create a branch with a few known format objects.
113
 
        # this is not quite the same as 
114
 
        t = get_transport(self.get_url())
 
281
        bzrdir.BzrProber.formats.register(BzrDirFormatTest1.get_format_string(),
 
282
            BzrDirFormatTest1())
 
283
        self.addCleanup(bzrdir.BzrProber.formats.remove,
 
284
            BzrDirFormatTest1.get_format_string())
 
285
        bzrdir.BzrProber.formats.register(BzrDirFormatTest2.get_format_string(),
 
286
            BzrDirFormatTest2())
 
287
        self.addCleanup(bzrdir.BzrProber.formats.remove,
 
288
            BzrDirFormatTest2.get_format_string())
 
289
        t = self.get_transport()
115
290
        self.build_tree(["foo/", "bar/"], transport=t)
116
291
        def check_format(format, url):
117
292
            format.initialize(url)
118
 
            t = get_transport(url)
 
293
            t = _mod_transport.get_transport_from_path(url)
119
294
            found_format = bzrdir.BzrDirFormat.find_format(t)
120
 
            self.failUnless(isinstance(found_format, format.__class__))
121
 
        check_format(bzrdir.BzrDirFormat5(), "foo")
122
 
        check_format(bzrdir.BzrDirFormat6(), "bar")
123
 
        
 
295
            self.assertIsInstance(found_format, format.__class__)
 
296
        check_format(BzrDirFormatTest1(), "foo")
 
297
        check_format(BzrDirFormatTest2(), "bar")
 
298
 
124
299
    def test_find_format_nothing_there(self):
125
300
        self.assertRaises(NotBranchError,
126
301
                          bzrdir.BzrDirFormat.find_format,
127
 
                          get_transport('.'))
 
302
                          _mod_transport.get_transport_from_path('.'))
128
303
 
129
304
    def test_find_format_unknown_format(self):
130
 
        t = get_transport(self.get_url())
 
305
        t = self.get_transport()
131
306
        t.mkdir('.bzr')
132
 
        t.put('.bzr/branch-format', StringIO())
 
307
        t.put_bytes('.bzr/branch-format', '')
133
308
        self.assertRaises(UnknownFormatError,
134
309
                          bzrdir.BzrDirFormat.find_format,
135
 
                          get_transport('.'))
 
310
                          _mod_transport.get_transport_from_path('.'))
136
311
 
137
312
    def test_register_unregister_format(self):
138
313
        format = SampleBzrDirFormat()
140
315
        # make a bzrdir
141
316
        format.initialize(url)
142
317
        # register a format for it.
143
 
        bzrdir.BzrDirFormat.register_format(format)
 
318
        bzrdir.BzrProber.formats.register(format.get_format_string(), format)
144
319
        # which bzrdir.Open will refuse (not supported)
145
320
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open, url)
146
321
        # which bzrdir.open_containing will refuse (not supported)
147
322
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open_containing, url)
148
323
        # but open_downlevel will work
149
 
        t = get_transport(url)
 
324
        t = _mod_transport.get_transport_from_url(url)
150
325
        self.assertEqual(format.open(t), bzrdir.BzrDir.open_unsupported(url))
151
326
        # unregister the format
152
 
        bzrdir.BzrDirFormat.unregister_format(format)
 
327
        bzrdir.BzrProber.formats.remove(format.get_format_string())
153
328
        # now open_downlevel should fail too.
154
329
        self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
155
330
 
156
 
    def test_create_repository(self):
157
 
        format = SampleBzrDirFormat()
158
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
159
 
        bzrdir.BzrDirFormat.set_default_format(format)
160
 
        try:
161
 
            repo = bzrdir.BzrDir.create_repository(self.get_url())
162
 
            self.assertEqual('A repository', repo)
163
 
        finally:
164
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
165
 
 
166
 
    def test_create_repository_under_shared(self):
167
 
        # an explicit create_repository always does so.
168
 
        # we trust the format is right from the 'create_repository test'
169
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
170
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
171
 
        try:
172
 
            self.make_repository('.', shared=True)
173
 
            repo = bzrdir.BzrDir.create_repository(self.get_url('child'))
174
 
            self.assertTrue(isinstance(repo, repository.Repository))
175
 
            self.assertTrue(repo.bzrdir.root_transport.base.endswith('child/'))
176
 
        finally:
177
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
178
 
 
179
331
    def test_create_branch_and_repo_uses_default(self):
180
332
        format = SampleBzrDirFormat()
181
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
182
 
        bzrdir.BzrDirFormat.set_default_format(format)
183
 
        try:
184
 
            branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url())
185
 
            self.assertTrue(isinstance(branch, SampleBranch))
186
 
        finally:
187
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
333
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(),
 
334
                                                      format=format)
 
335
        self.assertTrue(isinstance(branch, SampleBranch))
188
336
 
189
337
    def test_create_branch_and_repo_under_shared(self):
190
338
        # creating a branch and repo in a shared repo uses the
191
339
        # shared repository
192
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
193
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
194
 
        try:
195
 
            self.make_repository('.', shared=True)
196
 
            branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'))
197
 
            self.assertRaises(errors.NoRepositoryPresent,
198
 
                              branch.bzrdir.open_repository)
199
 
        finally:
200
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
340
        format = controldir.format_registry.make_bzrdir('knit')
 
341
        self.make_repository('.', shared=True, format=format)
 
342
        branch = bzrdir.BzrDir.create_branch_and_repo(
 
343
            self.get_url('child'), format=format)
 
344
        self.assertRaises(errors.NoRepositoryPresent,
 
345
                          branch.bzrdir.open_repository)
201
346
 
202
347
    def test_create_branch_and_repo_under_shared_force_new(self):
203
 
        # creating a branch and repo in a shared repo can be forced to 
 
348
        # creating a branch and repo in a shared repo can be forced to
204
349
        # make a new repo
205
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
206
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
207
 
        try:
208
 
            self.make_repository('.', shared=True)
209
 
            branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
210
 
                                                          force_new_repo=True)
211
 
            branch.bzrdir.open_repository()
212
 
        finally:
213
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
350
        format = controldir.format_registry.make_bzrdir('knit')
 
351
        self.make_repository('.', shared=True, format=format)
 
352
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
 
353
                                                      force_new_repo=True,
 
354
                                                      format=format)
 
355
        branch.bzrdir.open_repository()
214
356
 
215
357
    def test_create_standalone_working_tree(self):
216
358
        format = SampleBzrDirFormat()
217
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
218
 
        bzrdir.BzrDirFormat.set_default_format(format)
219
 
        try:
220
 
            # note this is deliberately readonly, as this failure should 
221
 
            # occur before any writes.
222
 
            self.assertRaises(errors.NotLocalUrl,
223
 
                              bzrdir.BzrDir.create_standalone_workingtree,
224
 
                              self.get_readonly_url())
225
 
            tree = bzrdir.BzrDir.create_standalone_workingtree('.')
226
 
            self.assertEqual('A tree', tree)
227
 
        finally:
228
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
359
        # note this is deliberately readonly, as this failure should
 
360
        # occur before any writes.
 
361
        self.assertRaises(errors.NotLocalUrl,
 
362
                          bzrdir.BzrDir.create_standalone_workingtree,
 
363
                          self.get_readonly_url(), format=format)
 
364
        tree = bzrdir.BzrDir.create_standalone_workingtree('.',
 
365
                                                           format=format)
 
366
        self.assertEqual('A tree', tree)
229
367
 
230
368
    def test_create_standalone_working_tree_under_shared_repo(self):
231
369
        # create standalone working tree always makes a repo.
232
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
233
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
234
 
        try:
235
 
            self.make_repository('.', shared=True)
236
 
            # note this is deliberately readonly, as this failure should 
237
 
            # occur before any writes.
238
 
            self.assertRaises(errors.NotLocalUrl,
239
 
                              bzrdir.BzrDir.create_standalone_workingtree,
240
 
                              self.get_readonly_url('child'))
241
 
            tree = bzrdir.BzrDir.create_standalone_workingtree('child')
242
 
            tree.bzrdir.open_repository()
243
 
        finally:
244
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
370
        format = controldir.format_registry.make_bzrdir('knit')
 
371
        self.make_repository('.', shared=True, format=format)
 
372
        # note this is deliberately readonly, as this failure should
 
373
        # occur before any writes.
 
374
        self.assertRaises(errors.NotLocalUrl,
 
375
                          bzrdir.BzrDir.create_standalone_workingtree,
 
376
                          self.get_readonly_url('child'), format=format)
 
377
        tree = bzrdir.BzrDir.create_standalone_workingtree('child',
 
378
            format=format)
 
379
        tree.bzrdir.open_repository()
245
380
 
246
381
    def test_create_branch_convenience(self):
247
382
        # outside a repo the default convenience output is a repo+branch_tree
248
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
249
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
250
 
        try:
251
 
            branch = bzrdir.BzrDir.create_branch_convenience('.')
252
 
            branch.bzrdir.open_workingtree()
253
 
            branch.bzrdir.open_repository()
254
 
        finally:
255
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
383
        format = controldir.format_registry.make_bzrdir('knit')
 
384
        branch = bzrdir.BzrDir.create_branch_convenience('.', format=format)
 
385
        branch.bzrdir.open_workingtree()
 
386
        branch.bzrdir.open_repository()
 
387
 
 
388
    def test_create_branch_convenience_possible_transports(self):
 
389
        """Check that the optional 'possible_transports' is recognized"""
 
390
        format = controldir.format_registry.make_bzrdir('knit')
 
391
        t = self.get_transport()
 
392
        branch = bzrdir.BzrDir.create_branch_convenience(
 
393
            '.', format=format, possible_transports=[t])
 
394
        branch.bzrdir.open_workingtree()
 
395
        branch.bzrdir.open_repository()
 
396
 
 
397
    def test_create_branch_convenience_root(self):
 
398
        """Creating a branch at the root of a fs should work."""
 
399
        self.vfs_transport_factory = memory.MemoryServer
 
400
        # outside a repo the default convenience output is a repo+branch_tree
 
401
        format = controldir.format_registry.make_bzrdir('knit')
 
402
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
 
403
                                                         format=format)
 
404
        self.assertRaises(errors.NoWorkingTree,
 
405
                          branch.bzrdir.open_workingtree)
 
406
        branch.bzrdir.open_repository()
256
407
 
257
408
    def test_create_branch_convenience_under_shared_repo(self):
258
409
        # inside a repo the default convenience output is a branch+ follow the
259
410
        # repo tree policy
260
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
261
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
262
 
        try:
263
 
            self.make_repository('.', shared=True)
264
 
            branch = bzrdir.BzrDir.create_branch_convenience('child')
265
 
            branch.bzrdir.open_workingtree()
266
 
            self.assertRaises(errors.NoRepositoryPresent,
267
 
                              branch.bzrdir.open_repository)
268
 
        finally:
269
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
270
 
            
 
411
        format = controldir.format_registry.make_bzrdir('knit')
 
412
        self.make_repository('.', shared=True, format=format)
 
413
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
414
            format=format)
 
415
        branch.bzrdir.open_workingtree()
 
416
        self.assertRaises(errors.NoRepositoryPresent,
 
417
                          branch.bzrdir.open_repository)
 
418
 
271
419
    def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
272
420
        # inside a repo the default convenience output is a branch+ follow the
273
421
        # repo tree policy but we can override that
274
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
275
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
276
 
        try:
277
 
            self.make_repository('.', shared=True)
278
 
            branch = bzrdir.BzrDir.create_branch_convenience('child',
279
 
                force_new_tree=False)
280
 
            self.assertRaises(errors.NoWorkingTree,
281
 
                              branch.bzrdir.open_workingtree)
282
 
            self.assertRaises(errors.NoRepositoryPresent,
283
 
                              branch.bzrdir.open_repository)
284
 
        finally:
285
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
286
 
            
 
422
        format = controldir.format_registry.make_bzrdir('knit')
 
423
        self.make_repository('.', shared=True, format=format)
 
424
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
425
            force_new_tree=False, format=format)
 
426
        self.assertRaises(errors.NoWorkingTree,
 
427
                          branch.bzrdir.open_workingtree)
 
428
        self.assertRaises(errors.NoRepositoryPresent,
 
429
                          branch.bzrdir.open_repository)
 
430
 
287
431
    def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
288
432
        # inside a repo the default convenience output is a branch+ follow the
289
433
        # repo tree policy
290
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
291
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
292
 
        try:
293
 
            repo = self.make_repository('.', shared=True)
294
 
            repo.set_make_working_trees(False)
295
 
            branch = bzrdir.BzrDir.create_branch_convenience('child')
296
 
            self.assertRaises(errors.NoWorkingTree,
297
 
                              branch.bzrdir.open_workingtree)
298
 
            self.assertRaises(errors.NoRepositoryPresent,
299
 
                              branch.bzrdir.open_repository)
300
 
        finally:
301
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
434
        format = controldir.format_registry.make_bzrdir('knit')
 
435
        repo = self.make_repository('.', shared=True, format=format)
 
436
        repo.set_make_working_trees(False)
 
437
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
438
                                                         format=format)
 
439
        self.assertRaises(errors.NoWorkingTree,
 
440
                          branch.bzrdir.open_workingtree)
 
441
        self.assertRaises(errors.NoRepositoryPresent,
 
442
                          branch.bzrdir.open_repository)
302
443
 
303
444
    def test_create_branch_convenience_under_shared_repo_no_tree_policy_force_tree(self):
304
445
        # inside a repo the default convenience output is a branch+ follow the
305
446
        # repo tree policy but we can override that
306
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
307
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
308
 
        try:
309
 
            repo = self.make_repository('.', shared=True)
310
 
            repo.set_make_working_trees(False)
311
 
            branch = bzrdir.BzrDir.create_branch_convenience('child',
312
 
                force_new_tree=True)
313
 
            branch.bzrdir.open_workingtree()
314
 
            self.assertRaises(errors.NoRepositoryPresent,
315
 
                              branch.bzrdir.open_repository)
316
 
        finally:
317
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
447
        format = controldir.format_registry.make_bzrdir('knit')
 
448
        repo = self.make_repository('.', shared=True, format=format)
 
449
        repo.set_make_working_trees(False)
 
450
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
451
            force_new_tree=True, format=format)
 
452
        branch.bzrdir.open_workingtree()
 
453
        self.assertRaises(errors.NoRepositoryPresent,
 
454
                          branch.bzrdir.open_repository)
318
455
 
319
456
    def test_create_branch_convenience_under_shared_repo_force_new_repo(self):
320
457
        # inside a repo the default convenience output is overridable to give
321
458
        # repo+branch+tree
322
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
323
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
324
 
        try:
325
 
            self.make_repository('.', shared=True)
326
 
            branch = bzrdir.BzrDir.create_branch_convenience('child',
327
 
                force_new_repo=True)
328
 
            branch.bzrdir.open_repository()
329
 
            branch.bzrdir.open_workingtree()
330
 
        finally:
331
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
459
        format = controldir.format_registry.make_bzrdir('knit')
 
460
        self.make_repository('.', shared=True, format=format)
 
461
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
462
            force_new_repo=True, format=format)
 
463
        branch.bzrdir.open_repository()
 
464
        branch.bzrdir.open_workingtree()
 
465
 
 
466
 
 
467
class TestRepositoryAcquisitionPolicy(TestCaseWithTransport):
 
468
 
 
469
    def test_acquire_repository_standalone(self):
 
470
        """The default acquisition policy should create a standalone branch."""
 
471
        my_bzrdir = self.make_bzrdir('.')
 
472
        repo_policy = my_bzrdir.determine_repository_policy()
 
473
        repo, is_new = repo_policy.acquire_repository()
 
474
        self.assertEqual(repo.bzrdir.root_transport.base,
 
475
                         my_bzrdir.root_transport.base)
 
476
        self.assertFalse(repo.is_shared())
 
477
 
 
478
    def test_determine_stacking_policy(self):
 
479
        parent_bzrdir = self.make_bzrdir('.')
 
480
        child_bzrdir = self.make_bzrdir('child')
 
481
        parent_bzrdir.get_config().set_default_stack_on('http://example.org')
 
482
        repo_policy = child_bzrdir.determine_repository_policy()
 
483
        self.assertEqual('http://example.org', repo_policy._stack_on)
 
484
 
 
485
    def test_determine_stacking_policy_relative(self):
 
486
        parent_bzrdir = self.make_bzrdir('.')
 
487
        child_bzrdir = self.make_bzrdir('child')
 
488
        parent_bzrdir.get_config().set_default_stack_on('child2')
 
489
        repo_policy = child_bzrdir.determine_repository_policy()
 
490
        self.assertEqual('child2', repo_policy._stack_on)
 
491
        self.assertEqual(parent_bzrdir.root_transport.base,
 
492
                         repo_policy._stack_on_pwd)
 
493
 
 
494
    def prepare_default_stacking(self, child_format='1.6'):
 
495
        parent_bzrdir = self.make_bzrdir('.')
 
496
        child_branch = self.make_branch('child', format=child_format)
 
497
        parent_bzrdir.get_config().set_default_stack_on(child_branch.base)
 
498
        new_child_transport = parent_bzrdir.transport.clone('child2')
 
499
        return child_branch, new_child_transport
 
500
 
 
501
    def test_clone_on_transport_obeys_stacking_policy(self):
 
502
        child_branch, new_child_transport = self.prepare_default_stacking()
 
503
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
 
504
        self.assertEqual(child_branch.base,
 
505
                         new_child.open_branch().get_stacked_on_url())
 
506
 
 
507
    def test_default_stacking_with_stackable_branch_unstackable_repo(self):
 
508
        # Make stackable source branch with an unstackable repo format.
 
509
        source_bzrdir = self.make_bzrdir('source')
 
510
        knitpack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
 
511
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(
 
512
            source_bzrdir)
 
513
        # Make a directory with a default stacking policy
 
514
        parent_bzrdir = self.make_bzrdir('parent')
 
515
        stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
 
516
        parent_bzrdir.get_config().set_default_stack_on(stacked_on.base)
 
517
        # Clone source into directory
 
518
        target = source_bzrdir.clone(self.get_url('parent/target'))
 
519
 
 
520
    def test_format_initialize_on_transport_ex_stacked_on(self):
 
521
        # trunk is a stackable format.  Note that its in the same server area
 
522
        # which is what launchpad does, but not sufficient to exercise the
 
523
        # general case.
 
524
        trunk = self.make_branch('trunk', format='1.9')
 
525
        t = self.get_transport('stacked')
 
526
        old_fmt = controldir.format_registry.make_bzrdir('pack-0.92')
 
527
        repo_name = old_fmt.repository_format.network_name()
 
528
        # Should end up with a 1.9 format (stackable)
 
529
        repo, control, require_stacking, repo_policy = \
 
530
            old_fmt.initialize_on_transport_ex(t,
 
531
                    repo_format_name=repo_name, stacked_on='../trunk',
 
532
                    stack_on_pwd=t.base)
 
533
        if repo is not None:
 
534
            # Repositories are open write-locked
 
535
            self.assertTrue(repo.is_write_locked())
 
536
            self.addCleanup(repo.unlock)
 
537
        else:
 
538
            repo = control.open_repository()
 
539
        self.assertIsInstance(control, bzrdir.BzrDir)
 
540
        opened = bzrdir.BzrDir.open(t.base)
 
541
        if not isinstance(old_fmt, remote.RemoteBzrDirFormat):
 
542
            self.assertEqual(control._format.network_name(),
 
543
                old_fmt.network_name())
 
544
            self.assertEqual(control._format.network_name(),
 
545
                opened._format.network_name())
 
546
        self.assertEqual(control.__class__, opened.__class__)
 
547
        self.assertLength(1, repo._fallback_repositories)
 
548
 
 
549
    def test_sprout_obeys_stacking_policy(self):
 
550
        child_branch, new_child_transport = self.prepare_default_stacking()
 
551
        new_child = child_branch.bzrdir.sprout(new_child_transport.base)
 
552
        self.assertEqual(child_branch.base,
 
553
                         new_child.open_branch().get_stacked_on_url())
 
554
 
 
555
    def test_clone_ignores_policy_for_unsupported_formats(self):
 
556
        child_branch, new_child_transport = self.prepare_default_stacking(
 
557
            child_format='pack-0.92')
 
558
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
 
559
        self.assertRaises(errors.UnstackableBranchFormat,
 
560
                          new_child.open_branch().get_stacked_on_url)
 
561
 
 
562
    def test_sprout_ignores_policy_for_unsupported_formats(self):
 
563
        child_branch, new_child_transport = self.prepare_default_stacking(
 
564
            child_format='pack-0.92')
 
565
        new_child = child_branch.bzrdir.sprout(new_child_transport.base)
 
566
        self.assertRaises(errors.UnstackableBranchFormat,
 
567
                          new_child.open_branch().get_stacked_on_url)
 
568
 
 
569
    def test_sprout_upgrades_format_if_stacked_specified(self):
 
570
        child_branch, new_child_transport = self.prepare_default_stacking(
 
571
            child_format='pack-0.92')
 
572
        new_child = child_branch.bzrdir.sprout(new_child_transport.base,
 
573
                                               stacked=True)
 
574
        self.assertEqual(child_branch.bzrdir.root_transport.base,
 
575
                         new_child.open_branch().get_stacked_on_url())
 
576
        repo = new_child.open_repository()
 
577
        self.assertTrue(repo._format.supports_external_lookups)
 
578
        self.assertFalse(repo.supports_rich_root())
 
579
 
 
580
    def test_clone_on_transport_upgrades_format_if_stacked_on_specified(self):
 
581
        child_branch, new_child_transport = self.prepare_default_stacking(
 
582
            child_format='pack-0.92')
 
583
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport,
 
584
            stacked_on=child_branch.bzrdir.root_transport.base)
 
585
        self.assertEqual(child_branch.bzrdir.root_transport.base,
 
586
                         new_child.open_branch().get_stacked_on_url())
 
587
        repo = new_child.open_repository()
 
588
        self.assertTrue(repo._format.supports_external_lookups)
 
589
        self.assertFalse(repo.supports_rich_root())
 
590
 
 
591
    def test_sprout_upgrades_to_rich_root_format_if_needed(self):
 
592
        child_branch, new_child_transport = self.prepare_default_stacking(
 
593
            child_format='rich-root-pack')
 
594
        new_child = child_branch.bzrdir.sprout(new_child_transport.base,
 
595
                                               stacked=True)
 
596
        repo = new_child.open_repository()
 
597
        self.assertTrue(repo._format.supports_external_lookups)
 
598
        self.assertTrue(repo.supports_rich_root())
 
599
 
 
600
    def test_add_fallback_repo_handles_absolute_urls(self):
 
601
        stack_on = self.make_branch('stack_on', format='1.6')
 
602
        repo = self.make_repository('repo', format='1.6')
 
603
        policy = bzrdir.UseExistingRepository(repo, stack_on.base)
 
604
        policy._add_fallback(repo)
 
605
 
 
606
    def test_add_fallback_repo_handles_relative_urls(self):
 
607
        stack_on = self.make_branch('stack_on', format='1.6')
 
608
        repo = self.make_repository('repo', format='1.6')
 
609
        policy = bzrdir.UseExistingRepository(repo, '.', stack_on.base)
 
610
        policy._add_fallback(repo)
 
611
 
 
612
    def test_configure_relative_branch_stacking_url(self):
 
613
        stack_on = self.make_branch('stack_on', format='1.6')
 
614
        stacked = self.make_branch('stack_on/stacked', format='1.6')
 
615
        policy = bzrdir.UseExistingRepository(stacked.repository,
 
616
            '.', stack_on.base)
 
617
        policy.configure_branch(stacked)
 
618
        self.assertEqual('..', stacked.get_stacked_on_url())
 
619
 
 
620
    def test_relative_branch_stacking_to_absolute(self):
 
621
        stack_on = self.make_branch('stack_on', format='1.6')
 
622
        stacked = self.make_branch('stack_on/stacked', format='1.6')
 
623
        policy = bzrdir.UseExistingRepository(stacked.repository,
 
624
            '.', self.get_readonly_url('stack_on'))
 
625
        policy.configure_branch(stacked)
 
626
        self.assertEqual(self.get_readonly_url('stack_on'),
 
627
                         stacked.get_stacked_on_url())
332
628
 
333
629
 
334
630
class ChrootedTests(TestCaseWithTransport):
341
637
 
342
638
    def setUp(self):
343
639
        super(ChrootedTests, self).setUp()
344
 
        if not self.transport_server == MemoryServer:
345
 
            self.transport_readonly_server = HttpServer
 
640
        if not self.vfs_transport_factory == memory.MemoryServer:
 
641
            self.transport_readonly_server = http_server.HttpServer
 
642
 
 
643
    def local_branch_path(self, branch):
 
644
         return os.path.realpath(urlutils.local_path_from_url(branch.base))
346
645
 
347
646
    def test_open_containing(self):
348
647
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
355
654
        branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
356
655
        self.assertEqual('g/p/q', relpath)
357
656
 
 
657
    def test_open_containing_tree_branch_or_repository_empty(self):
 
658
        self.assertRaises(errors.NotBranchError,
 
659
            bzrdir.BzrDir.open_containing_tree_branch_or_repository,
 
660
            self.get_readonly_url(''))
 
661
 
 
662
    def test_open_containing_tree_branch_or_repository_all(self):
 
663
        self.make_branch_and_tree('topdir')
 
664
        tree, branch, repo, relpath = \
 
665
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
666
                'topdir/foo')
 
667
        self.assertEqual(os.path.realpath('topdir'),
 
668
                         os.path.realpath(tree.basedir))
 
669
        self.assertEqual(os.path.realpath('topdir'),
 
670
                         self.local_branch_path(branch))
 
671
        self.assertEqual(
 
672
            osutils.realpath(os.path.join('topdir', '.bzr', 'repository')),
 
673
            repo.bzrdir.transport.local_abspath('repository'))
 
674
        self.assertEqual(relpath, 'foo')
 
675
 
 
676
    def test_open_containing_tree_branch_or_repository_no_tree(self):
 
677
        self.make_branch('branch')
 
678
        tree, branch, repo, relpath = \
 
679
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
680
                'branch/foo')
 
681
        self.assertEqual(tree, None)
 
682
        self.assertEqual(os.path.realpath('branch'),
 
683
                         self.local_branch_path(branch))
 
684
        self.assertEqual(
 
685
            osutils.realpath(os.path.join('branch', '.bzr', 'repository')),
 
686
            repo.bzrdir.transport.local_abspath('repository'))
 
687
        self.assertEqual(relpath, 'foo')
 
688
 
 
689
    def test_open_containing_tree_branch_or_repository_repo(self):
 
690
        self.make_repository('repo')
 
691
        tree, branch, repo, relpath = \
 
692
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
693
                'repo')
 
694
        self.assertEqual(tree, None)
 
695
        self.assertEqual(branch, None)
 
696
        self.assertEqual(
 
697
            osutils.realpath(os.path.join('repo', '.bzr', 'repository')),
 
698
            repo.bzrdir.transport.local_abspath('repository'))
 
699
        self.assertEqual(relpath, '')
 
700
 
 
701
    def test_open_containing_tree_branch_or_repository_shared_repo(self):
 
702
        self.make_repository('shared', shared=True)
 
703
        bzrdir.BzrDir.create_branch_convenience('shared/branch',
 
704
                                                force_new_tree=False)
 
705
        tree, branch, repo, relpath = \
 
706
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
707
                'shared/branch')
 
708
        self.assertEqual(tree, None)
 
709
        self.assertEqual(os.path.realpath('shared/branch'),
 
710
                         self.local_branch_path(branch))
 
711
        self.assertEqual(
 
712
            osutils.realpath(os.path.join('shared', '.bzr', 'repository')),
 
713
            repo.bzrdir.transport.local_abspath('repository'))
 
714
        self.assertEqual(relpath, '')
 
715
 
 
716
    def test_open_containing_tree_branch_or_repository_branch_subdir(self):
 
717
        self.make_branch_and_tree('foo')
 
718
        self.build_tree(['foo/bar/'])
 
719
        tree, branch, repo, relpath = \
 
720
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
721
                'foo/bar')
 
722
        self.assertEqual(os.path.realpath('foo'),
 
723
                         os.path.realpath(tree.basedir))
 
724
        self.assertEqual(os.path.realpath('foo'),
 
725
                         self.local_branch_path(branch))
 
726
        self.assertEqual(
 
727
            osutils.realpath(os.path.join('foo', '.bzr', 'repository')),
 
728
            repo.bzrdir.transport.local_abspath('repository'))
 
729
        self.assertEqual(relpath, 'bar')
 
730
 
 
731
    def test_open_containing_tree_branch_or_repository_repo_subdir(self):
 
732
        self.make_repository('bar')
 
733
        self.build_tree(['bar/baz/'])
 
734
        tree, branch, repo, relpath = \
 
735
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
736
                'bar/baz')
 
737
        self.assertEqual(tree, None)
 
738
        self.assertEqual(branch, None)
 
739
        self.assertEqual(
 
740
            osutils.realpath(os.path.join('bar', '.bzr', 'repository')),
 
741
            repo.bzrdir.transport.local_abspath('repository'))
 
742
        self.assertEqual(relpath, 'baz')
 
743
 
358
744
    def test_open_containing_from_transport(self):
359
 
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
360
 
                          get_transport(self.get_readonly_url('')))
361
 
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
362
 
                          get_transport(self.get_readonly_url('g/p/q')))
 
745
        self.assertRaises(NotBranchError,
 
746
            bzrdir.BzrDir.open_containing_from_transport,
 
747
            _mod_transport.get_transport_from_url(self.get_readonly_url('')))
 
748
        self.assertRaises(NotBranchError,
 
749
            bzrdir.BzrDir.open_containing_from_transport,
 
750
            _mod_transport.get_transport_from_url(
 
751
                self.get_readonly_url('g/p/q')))
363
752
        control = bzrdir.BzrDir.create(self.get_url())
364
753
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
365
 
            get_transport(self.get_readonly_url('')))
 
754
            _mod_transport.get_transport_from_url(
 
755
                self.get_readonly_url('')))
366
756
        self.assertEqual('', relpath)
367
757
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
368
 
            get_transport(self.get_readonly_url('g/p/q')))
 
758
            _mod_transport.get_transport_from_url(
 
759
                self.get_readonly_url('g/p/q')))
369
760
        self.assertEqual('g/p/q', relpath)
370
761
 
 
762
    def test_open_containing_tree_or_branch(self):
 
763
        self.make_branch_and_tree('topdir')
 
764
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
765
            'topdir/foo')
 
766
        self.assertEqual(os.path.realpath('topdir'),
 
767
                         os.path.realpath(tree.basedir))
 
768
        self.assertEqual(os.path.realpath('topdir'),
 
769
                         self.local_branch_path(branch))
 
770
        self.assertIs(tree.bzrdir, branch.bzrdir)
 
771
        self.assertEqual('foo', relpath)
 
772
        # opening from non-local should not return the tree
 
773
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
774
            self.get_readonly_url('topdir/foo'))
 
775
        self.assertEqual(None, tree)
 
776
        self.assertEqual('foo', relpath)
 
777
        # without a tree:
 
778
        self.make_branch('topdir/foo')
 
779
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
780
            'topdir/foo')
 
781
        self.assertIs(tree, None)
 
782
        self.assertEqual(os.path.realpath('topdir/foo'),
 
783
                         self.local_branch_path(branch))
 
784
        self.assertEqual('', relpath)
 
785
 
 
786
    def test_open_tree_or_branch(self):
 
787
        self.make_branch_and_tree('topdir')
 
788
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir')
 
789
        self.assertEqual(os.path.realpath('topdir'),
 
790
                         os.path.realpath(tree.basedir))
 
791
        self.assertEqual(os.path.realpath('topdir'),
 
792
                         self.local_branch_path(branch))
 
793
        self.assertIs(tree.bzrdir, branch.bzrdir)
 
794
        # opening from non-local should not return the tree
 
795
        tree, branch = bzrdir.BzrDir.open_tree_or_branch(
 
796
            self.get_readonly_url('topdir'))
 
797
        self.assertEqual(None, tree)
 
798
        # without a tree:
 
799
        self.make_branch('topdir/foo')
 
800
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir/foo')
 
801
        self.assertIs(tree, None)
 
802
        self.assertEqual(os.path.realpath('topdir/foo'),
 
803
                         self.local_branch_path(branch))
 
804
 
 
805
    def test_open_from_transport(self):
 
806
        # transport pointing at bzrdir should give a bzrdir with root transport
 
807
        # set to the given transport
 
808
        control = bzrdir.BzrDir.create(self.get_url())
 
809
        t = self.get_transport()
 
810
        opened_bzrdir = bzrdir.BzrDir.open_from_transport(t)
 
811
        self.assertEqual(t.base, opened_bzrdir.root_transport.base)
 
812
        self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
 
813
 
 
814
    def test_open_from_transport_no_bzrdir(self):
 
815
        t = self.get_transport()
 
816
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
 
817
 
 
818
    def test_open_from_transport_bzrdir_in_parent(self):
 
819
        control = bzrdir.BzrDir.create(self.get_url())
 
820
        t = self.get_transport()
 
821
        t.mkdir('subdir')
 
822
        t = t.clone('subdir')
 
823
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
 
824
 
 
825
    def test_sprout_recursive(self):
 
826
        tree = self.make_branch_and_tree('tree1',
 
827
                                         format='development-subtree')
 
828
        sub_tree = self.make_branch_and_tree('tree1/subtree',
 
829
            format='development-subtree')
 
830
        sub_tree.set_root_id('subtree-root')
 
831
        tree.add_reference(sub_tree)
 
832
        self.build_tree(['tree1/subtree/file'])
 
833
        sub_tree.add('file')
 
834
        tree.commit('Initial commit')
 
835
        tree2 = tree.bzrdir.sprout('tree2').open_workingtree()
 
836
        tree2.lock_read()
 
837
        self.addCleanup(tree2.unlock)
 
838
        self.assertPathExists('tree2/subtree/file')
 
839
        self.assertEqual('tree-reference', tree2.kind('subtree-root'))
 
840
 
 
841
    def test_cloning_metadir(self):
 
842
        """Ensure that cloning metadir is suitable"""
 
843
        bzrdir = self.make_bzrdir('bzrdir')
 
844
        bzrdir.cloning_metadir()
 
845
        branch = self.make_branch('branch', format='knit')
 
846
        format = branch.bzrdir.cloning_metadir()
 
847
        self.assertIsInstance(format.workingtree_format,
 
848
            workingtree_4.WorkingTreeFormat6)
 
849
 
 
850
    def test_sprout_recursive_treeless(self):
 
851
        tree = self.make_branch_and_tree('tree1',
 
852
            format='development-subtree')
 
853
        sub_tree = self.make_branch_and_tree('tree1/subtree',
 
854
            format='development-subtree')
 
855
        tree.add_reference(sub_tree)
 
856
        self.build_tree(['tree1/subtree/file'])
 
857
        sub_tree.add('file')
 
858
        tree.commit('Initial commit')
 
859
        # The following line force the orhaning to reveal bug #634470
 
860
        tree.branch.get_config_stack().set(
 
861
            'bzr.transform.orphan_policy', 'move')
 
862
        tree.bzrdir.destroy_workingtree()
 
863
        # FIXME: subtree/.bzr is left here which allows the test to pass (or
 
864
        # fail :-( ) -- vila 20100909
 
865
        repo = self.make_repository('repo', shared=True,
 
866
            format='development-subtree')
 
867
        repo.set_make_working_trees(False)
 
868
        # FIXME: we just deleted the workingtree and now we want to use it ????
 
869
        # At a minimum, we should use tree.branch below (but this fails too
 
870
        # currently) or stop calling this test 'treeless'. Specifically, I've
 
871
        # turn the line below into an assertRaises when 'subtree/.bzr' is
 
872
        # orphaned and sprout tries to access the branch there (which is left
 
873
        # by bzrdir.BzrDirMeta1.destroy_workingtree when it ignores the
 
874
        # [DeletingParent('Not deleting', u'subtree', None)] conflict). See bug
 
875
        # #634470.  -- vila 20100909
 
876
        self.assertRaises(errors.NotBranchError,
 
877
                          tree.bzrdir.sprout, 'repo/tree2')
 
878
#        self.assertPathExists('repo/tree2/subtree')
 
879
#        self.assertPathDoesNotExist('repo/tree2/subtree/file')
 
880
 
 
881
    def make_foo_bar_baz(self):
 
882
        foo = bzrdir.BzrDir.create_branch_convenience('foo').bzrdir
 
883
        bar = self.make_branch('foo/bar').bzrdir
 
884
        baz = self.make_branch('baz').bzrdir
 
885
        return foo, bar, baz
 
886
 
 
887
    def test_find_bzrdirs(self):
 
888
        foo, bar, baz = self.make_foo_bar_baz()
 
889
        t = self.get_transport()
 
890
        self.assertEqualBzrdirs([baz, foo, bar], bzrdir.BzrDir.find_bzrdirs(t))
 
891
 
 
892
    def make_fake_permission_denied_transport(self, transport, paths):
 
893
        """Create a transport that raises PermissionDenied for some paths."""
 
894
        def filter(path):
 
895
            if path in paths:
 
896
                raise errors.PermissionDenied(path)
 
897
            return path
 
898
        path_filter_server = pathfilter.PathFilteringServer(transport, filter)
 
899
        path_filter_server.start_server()
 
900
        self.addCleanup(path_filter_server.stop_server)
 
901
        path_filter_transport = pathfilter.PathFilteringTransport(
 
902
            path_filter_server, '.')
 
903
        return (path_filter_server, path_filter_transport)
 
904
 
 
905
    def assertBranchUrlsEndWith(self, expect_url_suffix, actual_bzrdirs):
 
906
        """Check that each branch url ends with the given suffix."""
 
907
        for actual_bzrdir in actual_bzrdirs:
 
908
            self.assertEndsWith(actual_bzrdir.user_url, expect_url_suffix)
 
909
 
 
910
    def test_find_bzrdirs_permission_denied(self):
 
911
        foo, bar, baz = self.make_foo_bar_baz()
 
912
        t = self.get_transport()
 
913
        path_filter_server, path_filter_transport = \
 
914
            self.make_fake_permission_denied_transport(t, ['foo'])
 
915
        # local transport
 
916
        self.assertBranchUrlsEndWith('/baz/',
 
917
            bzrdir.BzrDir.find_bzrdirs(path_filter_transport))
 
918
        # smart server
 
919
        smart_transport = self.make_smart_server('.',
 
920
            backing_server=path_filter_server)
 
921
        self.assertBranchUrlsEndWith('/baz/',
 
922
            bzrdir.BzrDir.find_bzrdirs(smart_transport))
 
923
 
 
924
    def test_find_bzrdirs_list_current(self):
 
925
        def list_current(transport):
 
926
            return [s for s in transport.list_dir('') if s != 'baz']
 
927
 
 
928
        foo, bar, baz = self.make_foo_bar_baz()
 
929
        t = self.get_transport()
 
930
        self.assertEqualBzrdirs(
 
931
            [foo, bar],
 
932
            bzrdir.BzrDir.find_bzrdirs(t, list_current=list_current))
 
933
 
 
934
    def test_find_bzrdirs_evaluate(self):
 
935
        def evaluate(bzrdir):
 
936
            try:
 
937
                repo = bzrdir.open_repository()
 
938
            except errors.NoRepositoryPresent:
 
939
                return True, bzrdir.root_transport.base
 
940
            else:
 
941
                return False, bzrdir.root_transport.base
 
942
 
 
943
        foo, bar, baz = self.make_foo_bar_baz()
 
944
        t = self.get_transport()
 
945
        self.assertEqual([baz.root_transport.base, foo.root_transport.base],
 
946
                         list(bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate)))
 
947
 
 
948
    def assertEqualBzrdirs(self, first, second):
 
949
        first = list(first)
 
950
        second = list(second)
 
951
        self.assertEqual(len(first), len(second))
 
952
        for x, y in zip(first, second):
 
953
            self.assertEqual(x.root_transport.base, y.root_transport.base)
 
954
 
 
955
    def test_find_branches(self):
 
956
        root = self.make_repository('', shared=True)
 
957
        foo, bar, baz = self.make_foo_bar_baz()
 
958
        qux = self.make_bzrdir('foo/qux')
 
959
        t = self.get_transport()
 
960
        branches = bzrdir.BzrDir.find_branches(t)
 
961
        self.assertEqual(baz.root_transport.base, branches[0].base)
 
962
        self.assertEqual(foo.root_transport.base, branches[1].base)
 
963
        self.assertEqual(bar.root_transport.base, branches[2].base)
 
964
 
 
965
        # ensure this works without a top-level repo
 
966
        branches = bzrdir.BzrDir.find_branches(t.clone('foo'))
 
967
        self.assertEqual(foo.root_transport.base, branches[0].base)
 
968
        self.assertEqual(bar.root_transport.base, branches[1].base)
 
969
 
 
970
 
 
971
class TestMissingRepoBranchesSkipped(TestCaseWithMemoryTransport):
 
972
 
 
973
    def test_find_bzrdirs_missing_repo(self):
 
974
        t = self.get_transport()
 
975
        arepo = self.make_repository('arepo', shared=True)
 
976
        abranch_url = arepo.user_url + '/abranch'
 
977
        abranch = bzrdir.BzrDir.create(abranch_url).create_branch()
 
978
        t.delete_tree('arepo/.bzr')
 
979
        self.assertRaises(errors.NoRepositoryPresent,
 
980
            branch.Branch.open, abranch_url)
 
981
        self.make_branch('baz')
 
982
        for actual_bzrdir in bzrdir.BzrDir.find_branches(t):
 
983
            self.assertEndsWith(actual_bzrdir.user_url, '/baz/')
 
984
 
371
985
 
372
986
class TestMeta1DirFormat(TestCaseWithTransport):
373
987
    """Tests specific to the meta1 dir format."""
378
992
        branch_base = t.clone('branch').base
379
993
        self.assertEqual(branch_base, dir.get_branch_transport(None).base)
380
994
        self.assertEqual(branch_base,
381
 
                         dir.get_branch_transport(bzrlib.branch.BzrBranchFormat5()).base)
 
995
                         dir.get_branch_transport(BzrBranchFormat5()).base)
382
996
        repository_base = t.clone('repository').base
383
997
        self.assertEqual(repository_base, dir.get_repository_transport(None).base)
 
998
        repository_format = repository.format_registry.get_default()
384
999
        self.assertEqual(repository_base,
385
 
                         dir.get_repository_transport(repository.RepositoryFormat7()).base)
 
1000
                         dir.get_repository_transport(repository_format).base)
386
1001
        checkout_base = t.clone('checkout').base
387
1002
        self.assertEqual(checkout_base, dir.get_workingtree_transport(None).base)
388
1003
        self.assertEqual(checkout_base,
389
 
                         dir.get_workingtree_transport(workingtree.WorkingTreeFormat3()).base)
 
1004
                         dir.get_workingtree_transport(workingtree_3.WorkingTreeFormat3()).base)
390
1005
 
391
1006
    def test_meta1dir_uses_lockdir(self):
392
1007
        """Meta1 format uses a LockDir to guard the whole directory, not a file."""
394
1009
        t = dir.transport
395
1010
        self.assertIsDirectory('branch-lock', t)
396
1011
 
397
 
        
398
 
class TestFormat5(TestCaseWithTransport):
399
 
    """Tests specific to the version 5 bzrdir format."""
400
 
 
401
 
    def test_same_lockfiles_between_tree_repo_branch(self):
402
 
        # this checks that only a single lockfiles instance is created 
403
 
        # for format 5 objects
404
 
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
405
 
        def check_dir_components_use_same_lock(dir):
406
 
            ctrl_1 = dir.open_repository().control_files
407
 
            ctrl_2 = dir.open_branch().control_files
408
 
            ctrl_3 = dir.open_workingtree()._control_files
409
 
            self.assertTrue(ctrl_1 is ctrl_2)
410
 
            self.assertTrue(ctrl_2 is ctrl_3)
411
 
        check_dir_components_use_same_lock(dir)
412
 
        # and if we open it normally.
413
 
        dir = bzrdir.BzrDir.open(self.get_url())
414
 
        check_dir_components_use_same_lock(dir)
415
 
    
416
 
    def test_can_convert(self):
417
 
        # format 5 dirs are convertable
418
 
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
419
 
        self.assertTrue(dir.can_convert_format())
420
 
    
421
 
    def test_needs_conversion(self):
422
 
        # format 5 dirs need a conversion if they are not the default.
423
 
        # and they start of not the default.
424
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
425
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirFormat5())
426
 
        try:
427
 
            dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
428
 
            self.assertFalse(dir.needs_format_conversion())
429
 
        finally:
430
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
431
 
        self.assertTrue(dir.needs_format_conversion())
432
 
 
433
 
 
434
 
class TestFormat6(TestCaseWithTransport):
435
 
    """Tests specific to the version 6 bzrdir format."""
436
 
 
437
 
    def test_same_lockfiles_between_tree_repo_branch(self):
438
 
        # this checks that only a single lockfiles instance is created 
439
 
        # for format 6 objects
440
 
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
441
 
        def check_dir_components_use_same_lock(dir):
442
 
            ctrl_1 = dir.open_repository().control_files
443
 
            ctrl_2 = dir.open_branch().control_files
444
 
            ctrl_3 = dir.open_workingtree()._control_files
445
 
            self.assertTrue(ctrl_1 is ctrl_2)
446
 
            self.assertTrue(ctrl_2 is ctrl_3)
447
 
        check_dir_components_use_same_lock(dir)
448
 
        # and if we open it normally.
449
 
        dir = bzrdir.BzrDir.open(self.get_url())
450
 
        check_dir_components_use_same_lock(dir)
451
 
    
452
 
    def test_can_convert(self):
453
 
        # format 6 dirs are convertable
454
 
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
455
 
        self.assertTrue(dir.can_convert_format())
456
 
    
457
 
    def test_needs_conversion(self):
458
 
        # format 6 dirs need an conversion if they are not the default.
459
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
460
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
461
 
        try:
462
 
            dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
463
 
            self.assertTrue(dir.needs_format_conversion())
464
 
        finally:
465
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
1012
    def test_comparison(self):
 
1013
        """Equality and inequality behave properly.
 
1014
 
 
1015
        Metadirs should compare equal iff they have the same repo, branch and
 
1016
        tree formats.
 
1017
        """
 
1018
        mydir = controldir.format_registry.make_bzrdir('knit')
 
1019
        self.assertEqual(mydir, mydir)
 
1020
        self.assertFalse(mydir != mydir)
 
1021
        otherdir = controldir.format_registry.make_bzrdir('knit')
 
1022
        self.assertEqual(otherdir, mydir)
 
1023
        self.assertFalse(otherdir != mydir)
 
1024
        otherdir2 = controldir.format_registry.make_bzrdir('development-subtree')
 
1025
        self.assertNotEqual(otherdir2, mydir)
 
1026
        self.assertFalse(otherdir2 == mydir)
 
1027
 
 
1028
    def test_with_features(self):
 
1029
        tree = self.make_branch_and_tree('tree', format='2a')
 
1030
        tree.bzrdir.update_feature_flags({"bar": "required"})
 
1031
        self.assertRaises(errors.MissingFeature, bzrdir.BzrDir.open, 'tree')
 
1032
        bzrdir.BzrDirMetaFormat1.register_feature('bar')
 
1033
        self.addCleanup(bzrdir.BzrDirMetaFormat1.unregister_feature, 'bar')
 
1034
        dir = bzrdir.BzrDir.open('tree')
 
1035
        self.assertEquals("required", dir._format.features.get("bar"))
 
1036
        tree.bzrdir.update_feature_flags({"bar": None, "nonexistant": None})
 
1037
        dir = bzrdir.BzrDir.open('tree')
 
1038
        self.assertEquals({}, dir._format.features)
 
1039
 
 
1040
    def test_needs_conversion_different_working_tree(self):
 
1041
        # meta1dirs need an conversion if any element is not the default.
 
1042
        new_format = controldir.format_registry.make_bzrdir('dirstate')
 
1043
        tree = self.make_branch_and_tree('tree', format='knit')
 
1044
        self.assertTrue(tree.bzrdir.needs_format_conversion(
 
1045
            new_format))
 
1046
 
 
1047
    def test_initialize_on_format_uses_smart_transport(self):
 
1048
        self.setup_smart_server_with_call_log()
 
1049
        new_format = controldir.format_registry.make_bzrdir('dirstate')
 
1050
        transport = self.get_transport('target')
 
1051
        transport.ensure_base()
 
1052
        self.reset_smart_call_log()
 
1053
        instance = new_format.initialize_on_transport(transport)
 
1054
        self.assertIsInstance(instance, remote.RemoteBzrDir)
 
1055
        rpc_count = len(self.hpss_calls)
 
1056
        # This figure represent the amount of work to perform this use case. It
 
1057
        # is entirely ok to reduce this number if a test fails due to rpc_count
 
1058
        # being too low. If rpc_count increases, more network roundtrips have
 
1059
        # become necessary for this use case. Please do not adjust this number
 
1060
        # upwards without agreement from bzr's network support maintainers.
 
1061
        self.assertEqual(2, rpc_count)
466
1062
 
467
1063
 
468
1064
class NonLocalTests(TestCaseWithTransport):
470
1066
 
471
1067
    def setUp(self):
472
1068
        super(NonLocalTests, self).setUp()
473
 
        self.transport_server = MemoryServer
474
 
    
 
1069
        self.vfs_transport_factory = memory.MemoryServer
 
1070
 
475
1071
    def test_create_branch_convenience(self):
476
1072
        # outside a repo the default convenience output is a repo+branch_tree
477
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
478
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
479
 
        try:
480
 
            branch = bzrdir.BzrDir.create_branch_convenience(self.get_url('foo'))
481
 
            self.assertRaises(errors.NoWorkingTree,
482
 
                              branch.bzrdir.open_workingtree)
483
 
            branch.bzrdir.open_repository()
484
 
        finally:
485
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
1073
        format = controldir.format_registry.make_bzrdir('knit')
 
1074
        branch = bzrdir.BzrDir.create_branch_convenience(
 
1075
            self.get_url('foo'), format=format)
 
1076
        self.assertRaises(errors.NoWorkingTree,
 
1077
                          branch.bzrdir.open_workingtree)
 
1078
        branch.bzrdir.open_repository()
486
1079
 
487
1080
    def test_create_branch_convenience_force_tree_not_local_fails(self):
488
1081
        # outside a repo the default convenience output is a repo+branch_tree
489
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
490
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
491
 
        try:
492
 
            self.assertRaises(errors.NotLocalUrl,
493
 
                bzrdir.BzrDir.create_branch_convenience,
494
 
                self.get_url('foo'),
495
 
                force_new_tree=True)
496
 
            t = get_transport(self.get_url('.'))
497
 
            self.assertFalse(t.has('foo'))
498
 
        finally:
499
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
1082
        format = controldir.format_registry.make_bzrdir('knit')
 
1083
        self.assertRaises(errors.NotLocalUrl,
 
1084
            bzrdir.BzrDir.create_branch_convenience,
 
1085
            self.get_url('foo'),
 
1086
            force_new_tree=True,
 
1087
            format=format)
 
1088
        t = self.get_transport()
 
1089
        self.assertFalse(t.has('foo'))
500
1090
 
501
1091
    def test_clone(self):
502
1092
        # clone into a nonlocal path works
503
 
        old_format = bzrdir.BzrDirFormat.get_default_format()
504
 
        bzrdir.BzrDirFormat.set_default_format(bzrdir.BzrDirMetaFormat1())
505
 
        try:
506
 
            branch = bzrdir.BzrDir.create_branch_convenience('local')
507
 
        finally:
508
 
            bzrdir.BzrDirFormat.set_default_format(old_format)
 
1093
        format = controldir.format_registry.make_bzrdir('knit')
 
1094
        branch = bzrdir.BzrDir.create_branch_convenience('local',
 
1095
                                                         format=format)
509
1096
        branch.bzrdir.open_workingtree()
510
1097
        result = branch.bzrdir.clone(self.get_url('remote'))
511
1098
        self.assertRaises(errors.NoWorkingTree,
513
1100
        result.open_branch()
514
1101
        result.open_repository()
515
1102
 
 
1103
    def test_checkout_metadir(self):
 
1104
        # checkout_metadir has reasonable working tree format even when no
 
1105
        # working tree is present
 
1106
        self.make_branch('branch-knit2', format='dirstate-with-subtree')
 
1107
        my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
 
1108
        checkout_format = my_bzrdir.checkout_metadir()
 
1109
        self.assertIsInstance(checkout_format.workingtree_format,
 
1110
                              workingtree_4.WorkingTreeFormat4)
 
1111
 
 
1112
 
 
1113
class TestHTTPRedirections(object):
 
1114
    """Test redirection between two http servers.
 
1115
 
 
1116
    This MUST be used by daughter classes that also inherit from
 
1117
    TestCaseWithTwoWebservers.
 
1118
 
 
1119
    We can't inherit directly from TestCaseWithTwoWebservers or the
 
1120
    test framework will try to create an instance which cannot
 
1121
    run, its implementation being incomplete.
 
1122
    """
 
1123
 
 
1124
    def create_transport_readonly_server(self):
 
1125
        # We don't set the http protocol version, relying on the default
 
1126
        return http_utils.HTTPServerRedirecting()
 
1127
 
 
1128
    def create_transport_secondary_server(self):
 
1129
        # We don't set the http protocol version, relying on the default
 
1130
        return http_utils.HTTPServerRedirecting()
 
1131
 
 
1132
    def setUp(self):
 
1133
        super(TestHTTPRedirections, self).setUp()
 
1134
        # The redirections will point to the new server
 
1135
        self.new_server = self.get_readonly_server()
 
1136
        # The requests to the old server will be redirected
 
1137
        self.old_server = self.get_secondary_server()
 
1138
        # Configure the redirections
 
1139
        self.old_server.redirect_to(self.new_server.host, self.new_server.port)
 
1140
 
 
1141
    def test_loop(self):
 
1142
        # Both servers redirect to each other creating a loop
 
1143
        self.new_server.redirect_to(self.old_server.host, self.old_server.port)
 
1144
        # Starting from either server should loop
 
1145
        old_url = self._qualified_url(self.old_server.host,
 
1146
                                      self.old_server.port)
 
1147
        oldt = self._transport(old_url)
 
1148
        self.assertRaises(errors.NotBranchError,
 
1149
                          bzrdir.BzrDir.open_from_transport, oldt)
 
1150
        new_url = self._qualified_url(self.new_server.host,
 
1151
                                      self.new_server.port)
 
1152
        newt = self._transport(new_url)
 
1153
        self.assertRaises(errors.NotBranchError,
 
1154
                          bzrdir.BzrDir.open_from_transport, newt)
 
1155
 
 
1156
    def test_qualifier_preserved(self):
 
1157
        wt = self.make_branch_and_tree('branch')
 
1158
        old_url = self._qualified_url(self.old_server.host,
 
1159
                                      self.old_server.port)
 
1160
        start = self._transport(old_url).clone('branch')
 
1161
        bdir = bzrdir.BzrDir.open_from_transport(start)
 
1162
        # Redirection should preserve the qualifier, hence the transport class
 
1163
        # itself.
 
1164
        self.assertIsInstance(bdir.root_transport, type(start))
 
1165
 
 
1166
 
 
1167
class TestHTTPRedirections_urllib(TestHTTPRedirections,
 
1168
                                  http_utils.TestCaseWithTwoWebservers):
 
1169
    """Tests redirections for urllib implementation"""
 
1170
 
 
1171
    _transport = HttpTransport_urllib
 
1172
 
 
1173
    def _qualified_url(self, host, port):
 
1174
        result = 'http+urllib://%s:%s' % (host, port)
 
1175
        self.permit_url(result)
 
1176
        return result
 
1177
 
 
1178
 
 
1179
 
 
1180
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
 
1181
                                  TestHTTPRedirections,
 
1182
                                  http_utils.TestCaseWithTwoWebservers):
 
1183
    """Tests redirections for pycurl implementation"""
 
1184
 
 
1185
    def _qualified_url(self, host, port):
 
1186
        result = 'http+pycurl://%s:%s' % (host, port)
 
1187
        self.permit_url(result)
 
1188
        return result
 
1189
 
 
1190
 
 
1191
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
 
1192
                                  http_utils.TestCaseWithTwoWebservers):
 
1193
    """Tests redirections for the nosmart decorator"""
 
1194
 
 
1195
    _transport = NoSmartTransportDecorator
 
1196
 
 
1197
    def _qualified_url(self, host, port):
 
1198
        result = 'nosmart+http://%s:%s' % (host, port)
 
1199
        self.permit_url(result)
 
1200
        return result
 
1201
 
 
1202
 
 
1203
class TestHTTPRedirections_readonly(TestHTTPRedirections,
 
1204
                                    http_utils.TestCaseWithTwoWebservers):
 
1205
    """Tests redirections for readonly decoratror"""
 
1206
 
 
1207
    _transport = ReadonlyTransportDecorator
 
1208
 
 
1209
    def _qualified_url(self, host, port):
 
1210
        result = 'readonly+http://%s:%s' % (host, port)
 
1211
        self.permit_url(result)
 
1212
        return result
 
1213
 
 
1214
 
 
1215
class TestDotBzrHidden(TestCaseWithTransport):
 
1216
 
 
1217
    ls = ['ls']
 
1218
    if sys.platform == 'win32':
 
1219
        ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
 
1220
 
 
1221
    def get_ls(self):
 
1222
        f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
 
1223
            stderr=subprocess.PIPE)
 
1224
        out, err = f.communicate()
 
1225
        self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
 
1226
                         % (self.ls, err))
 
1227
        return out.splitlines()
 
1228
 
 
1229
    def test_dot_bzr_hidden(self):
 
1230
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
1231
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
1232
        b = bzrdir.BzrDir.create('.')
 
1233
        self.build_tree(['a'])
 
1234
        self.assertEquals(['a'], self.get_ls())
 
1235
 
 
1236
    def test_dot_bzr_hidden_with_url(self):
 
1237
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
1238
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
1239
        b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
 
1240
        self.build_tree(['a'])
 
1241
        self.assertEquals(['a'], self.get_ls())
 
1242
 
 
1243
 
 
1244
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
 
1245
    """Test BzrDirFormat implementation for TestBzrDirSprout."""
 
1246
 
 
1247
    def _open(self, transport):
 
1248
        return _TestBzrDir(transport, self)
 
1249
 
 
1250
 
 
1251
class _TestBzrDir(bzrdir.BzrDirMeta1):
 
1252
    """Test BzrDir implementation for TestBzrDirSprout.
 
1253
 
 
1254
    When created a _TestBzrDir already has repository and a branch.  The branch
 
1255
    is a test double as well.
 
1256
    """
 
1257
 
 
1258
    def __init__(self, *args, **kwargs):
 
1259
        super(_TestBzrDir, self).__init__(*args, **kwargs)
 
1260
        self.test_branch = _TestBranch(self.transport)
 
1261
        self.test_branch.repository = self.create_repository()
 
1262
 
 
1263
    def open_branch(self, unsupported=False, possible_transports=None):
 
1264
        return self.test_branch
 
1265
 
 
1266
    def cloning_metadir(self, require_stacking=False):
 
1267
        return _TestBzrDirFormat()
 
1268
 
 
1269
 
 
1270
class _TestBranchFormat(bzrlib.branch.BranchFormat):
 
1271
    """Test Branch format for TestBzrDirSprout."""
 
1272
 
 
1273
 
 
1274
class _TestBranch(bzrlib.branch.Branch):
 
1275
    """Test Branch implementation for TestBzrDirSprout."""
 
1276
 
 
1277
    def __init__(self, transport, *args, **kwargs):
 
1278
        self._format = _TestBranchFormat()
 
1279
        self._transport = transport
 
1280
        self.base = transport.base
 
1281
        super(_TestBranch, self).__init__(*args, **kwargs)
 
1282
        self.calls = []
 
1283
        self._parent = None
 
1284
 
 
1285
    def sprout(self, *args, **kwargs):
 
1286
        self.calls.append('sprout')
 
1287
        return _TestBranch(self._transport)
 
1288
 
 
1289
    def copy_content_into(self, destination, revision_id=None):
 
1290
        self.calls.append('copy_content_into')
 
1291
 
 
1292
    def last_revision(self):
 
1293
        return _mod_revision.NULL_REVISION
 
1294
 
 
1295
    def get_parent(self):
 
1296
        return self._parent
 
1297
 
 
1298
    def _get_config(self):
 
1299
        return config.TransportConfig(self._transport, 'branch.conf')
 
1300
 
 
1301
    def _get_config_store(self):
 
1302
        return config.BranchStore(self)
 
1303
 
 
1304
    def set_parent(self, parent):
 
1305
        self._parent = parent
 
1306
 
 
1307
    def lock_read(self):
 
1308
        return lock.LogicalLockResult(self.unlock)
 
1309
 
 
1310
    def unlock(self):
 
1311
        return
 
1312
 
 
1313
 
 
1314
class TestBzrDirSprout(TestCaseWithMemoryTransport):
 
1315
 
 
1316
    def test_sprout_uses_branch_sprout(self):
 
1317
        """BzrDir.sprout calls Branch.sprout.
 
1318
 
 
1319
        Usually, BzrDir.sprout should delegate to the branch's sprout method
 
1320
        for part of the work.  This allows the source branch to control the
 
1321
        choice of format for the new branch.
 
1322
 
 
1323
        There are exceptions, but this tests avoids them:
 
1324
          - if there's no branch in the source bzrdir,
 
1325
          - or if the stacking has been requested and the format needs to be
 
1326
            overridden to satisfy that.
 
1327
        """
 
1328
        # Make an instrumented bzrdir.
 
1329
        t = self.get_transport('source')
 
1330
        t.ensure_base()
 
1331
        source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
 
1332
        # The instrumented bzrdir has a test_branch attribute that logs calls
 
1333
        # made to the branch contained in that bzrdir.  Initially the test
 
1334
        # branch exists but no calls have been made to it.
 
1335
        self.assertEqual([], source_bzrdir.test_branch.calls)
 
1336
 
 
1337
        # Sprout the bzrdir
 
1338
        target_url = self.get_url('target')
 
1339
        result = source_bzrdir.sprout(target_url, recurse='no')
 
1340
 
 
1341
        # The bzrdir called the branch's sprout method.
 
1342
        self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
 
1343
 
 
1344
    def test_sprout_parent(self):
 
1345
        grandparent_tree = self.make_branch('grandparent')
 
1346
        parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
 
1347
        branch_tree = parent.bzrdir.sprout('branch').open_branch()
 
1348
        self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
 
1349
 
 
1350
 
 
1351
class TestBzrDirHooks(TestCaseWithMemoryTransport):
 
1352
 
 
1353
    def test_pre_open_called(self):
 
1354
        calls = []
 
1355
        bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
 
1356
        transport = self.get_transport('foo')
 
1357
        url = transport.base
 
1358
        self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
 
1359
        self.assertEqual([transport.base], [t.base for t in calls])
 
1360
 
 
1361
    def test_pre_open_actual_exceptions_raised(self):
 
1362
        count = [0]
 
1363
        def fail_once(transport):
 
1364
            count[0] += 1
 
1365
            if count[0] == 1:
 
1366
                raise errors.BzrError("fail")
 
1367
        bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
 
1368
        transport = self.get_transport('foo')
 
1369
        url = transport.base
 
1370
        err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
 
1371
        self.assertEqual('fail', err._preformatted_string)
 
1372
 
 
1373
    def test_post_repo_init(self):
 
1374
        from bzrlib.controldir import RepoInitHookParams
 
1375
        calls = []
 
1376
        bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
 
1377
            calls.append, None)
 
1378
        self.make_repository('foo')
 
1379
        self.assertLength(1, calls)
 
1380
        params = calls[0]
 
1381
        self.assertIsInstance(params, RepoInitHookParams)
 
1382
        self.assertTrue(hasattr(params, 'bzrdir'))
 
1383
        self.assertTrue(hasattr(params, 'repository'))
 
1384
 
 
1385
    def test_post_repo_init_hook_repr(self):
 
1386
        param_reprs = []
 
1387
        bzrdir.BzrDir.hooks.install_named_hook('post_repo_init',
 
1388
            lambda params: param_reprs.append(repr(params)), None)
 
1389
        self.make_repository('foo')
 
1390
        self.assertLength(1, param_reprs)
 
1391
        param_repr = param_reprs[0]
 
1392
        self.assertStartsWith(param_repr, '<RepoInitHookParams for ')
 
1393
 
 
1394
 
 
1395
class TestGenerateBackupName(TestCaseWithMemoryTransport):
 
1396
    # FIXME: This may need to be unified with test_osutils.TestBackupNames or
 
1397
    # moved to per_bzrdir or per_transport for better coverage ?
 
1398
    # -- vila 20100909
 
1399
 
 
1400
    def setUp(self):
 
1401
        super(TestGenerateBackupName, self).setUp()
 
1402
        self._transport = self.get_transport()
 
1403
        bzrdir.BzrDir.create(self.get_url(),
 
1404
            possible_transports=[self._transport])
 
1405
        self._bzrdir = bzrdir.BzrDir.open_from_transport(self._transport)
 
1406
 
 
1407
    def test_new(self):
 
1408
        self.assertEqual("a.~1~", self._bzrdir._available_backup_name("a"))
 
1409
 
 
1410
    def test_exiting(self):
 
1411
        self._transport.put_bytes("a.~1~", "some content")
 
1412
        self.assertEqual("a.~2~", self._bzrdir._available_backup_name("a"))
 
1413
 
 
1414
 
 
1415
class TestMeta1DirColoFormat(TestCaseWithTransport):
 
1416
    """Tests specific to the meta1 dir with colocated branches format."""
 
1417
 
 
1418
    def test_supports_colo(self):
 
1419
        format = bzrdir.BzrDirMetaFormat1Colo()
 
1420
        self.assertTrue(format.colocated_branches)
 
1421
 
 
1422
    def test_upgrade_from_2a(self):
 
1423
        tree = self.make_branch_and_tree('.', format='2a')
 
1424
        format = bzrdir.BzrDirMetaFormat1Colo()
 
1425
        self.assertTrue(tree.bzrdir.needs_format_conversion(format))
 
1426
        converter = tree.bzrdir._format.get_converter(format)
 
1427
        result = converter.convert(tree.bzrdir, None)
 
1428
        self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1Colo)
 
1429
        self.assertFalse(result.needs_format_conversion(format))
 
1430
 
 
1431
    def test_downgrade_to_2a(self):
 
1432
        tree = self.make_branch_and_tree('.', format='development-colo')
 
1433
        format = bzrdir.BzrDirMetaFormat1()
 
1434
        self.assertTrue(tree.bzrdir.needs_format_conversion(format))
 
1435
        converter = tree.bzrdir._format.get_converter(format)
 
1436
        result = converter.convert(tree.bzrdir, None)
 
1437
        self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1)
 
1438
        self.assertFalse(result.needs_format_conversion(format))
 
1439
 
 
1440
    def test_downgrade_to_2a_too_many_branches(self):
 
1441
        tree = self.make_branch_and_tree('.', format='development-colo')
 
1442
        tree.bzrdir.create_branch(name="another-colocated-branch")
 
1443
        converter = tree.bzrdir._format.get_converter(
 
1444
            bzrdir.BzrDirMetaFormat1())
 
1445
        result = converter.convert(tree.bzrdir, bzrdir.BzrDirMetaFormat1())
 
1446
        self.assertIsInstance(result._format, bzrdir.BzrDirMetaFormat1)
 
1447
 
 
1448
    def test_nested(self):
 
1449
        tree = self.make_branch_and_tree('.', format='development-colo')
 
1450
        tree.bzrdir.create_branch(name='foo')
 
1451
        tree.bzrdir.create_branch(name='fool/bla')
 
1452
        self.assertRaises(
 
1453
            errors.ParentBranchExists, tree.bzrdir.create_branch,
 
1454
            name='foo/bar')
 
1455
 
 
1456
    def test_parent(self):
 
1457
        tree = self.make_branch_and_tree('.', format='development-colo')
 
1458
        tree.bzrdir.create_branch(name='fool/bla')
 
1459
        tree.bzrdir.create_branch(name='foo/bar')
 
1460
        self.assertRaises(
 
1461
            errors.AlreadyBranchError, tree.bzrdir.create_branch,
 
1462
            name='foo')
 
1463
 
 
1464
 
 
1465
class SampleBzrFormat(bzrdir.BzrFormat):
 
1466
 
 
1467
    @classmethod
 
1468
    def get_format_string(cls):
 
1469
        return "First line\n"
 
1470
 
 
1471
 
 
1472
class TestBzrFormat(TestCase):
 
1473
    """Tests for BzrFormat."""
 
1474
 
 
1475
    def test_as_string(self):
 
1476
        format = SampleBzrFormat()
 
1477
        format.features = {"foo": "required"}
 
1478
        self.assertEquals(format.as_string(),
 
1479
            "First line\n"
 
1480
            "required foo\n")
 
1481
        format.features["another"] = "optional"
 
1482
        self.assertEquals(format.as_string(),
 
1483
            "First line\n"
 
1484
            "required foo\n"
 
1485
            "optional another\n")
 
1486
 
 
1487
    def test_network_name(self):
 
1488
        # The network string should include the feature info
 
1489
        format = SampleBzrFormat()
 
1490
        format.features = {"foo": "required"}
 
1491
        self.assertEquals(
 
1492
            "First line\nrequired foo\n",
 
1493
            format.network_name())
 
1494
 
 
1495
    def test_from_string_no_features(self):
 
1496
        # No features
 
1497
        format = SampleBzrFormat.from_string(
 
1498
            "First line\n")
 
1499
        self.assertEquals({}, format.features)
 
1500
 
 
1501
    def test_from_string_with_feature(self):
 
1502
        # Proper feature
 
1503
        format = SampleBzrFormat.from_string(
 
1504
            "First line\nrequired foo\n")
 
1505
        self.assertEquals("required", format.features.get("foo"))
 
1506
 
 
1507
    def test_from_string_format_string_mismatch(self):
 
1508
        # The first line has to match the format string
 
1509
        self.assertRaises(AssertionError, SampleBzrFormat.from_string,
 
1510
            "Second line\nrequired foo\n")
 
1511
 
 
1512
    def test_from_string_missing_space(self):
 
1513
        # At least one space is required in the feature lines
 
1514
        self.assertRaises(errors.ParseFormatError, SampleBzrFormat.from_string,
 
1515
            "First line\nfoo\n")
 
1516
 
 
1517
    def test_from_string_with_spaces(self):
 
1518
        # Feature with spaces (in case we add stuff like this in the future)
 
1519
        format = SampleBzrFormat.from_string(
 
1520
            "First line\nrequired foo with spaces\n")
 
1521
        self.assertEquals("required", format.features.get("foo with spaces"))
 
1522
 
 
1523
    def test_eq(self):
 
1524
        format1 = SampleBzrFormat()
 
1525
        format1.features = {"nested-trees": "optional"}
 
1526
        format2 = SampleBzrFormat()
 
1527
        format2.features = {"nested-trees": "optional"}
 
1528
        self.assertEquals(format1, format1)
 
1529
        self.assertEquals(format1, format2)
 
1530
        format3 = SampleBzrFormat()
 
1531
        self.assertNotEquals(format1, format3)
 
1532
 
 
1533
    def test_check_support_status_optional(self):
 
1534
        # Optional, so silently ignore
 
1535
        format = SampleBzrFormat()
 
1536
        format.features = {"nested-trees": "optional"}
 
1537
        format.check_support_status(True)
 
1538
        self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
 
1539
        SampleBzrFormat.register_feature("nested-trees")
 
1540
        format.check_support_status(True)
 
1541
 
 
1542
    def test_check_support_status_required(self):
 
1543
        # Optional, so trigger an exception
 
1544
        format = SampleBzrFormat()
 
1545
        format.features = {"nested-trees": "required"}
 
1546
        self.assertRaises(errors.MissingFeature, format.check_support_status,
 
1547
            True)
 
1548
        self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
 
1549
        SampleBzrFormat.register_feature("nested-trees")
 
1550
        format.check_support_status(True)
 
1551
 
 
1552
    def test_check_support_status_unknown(self):
 
1553
        # treat unknown necessity as required
 
1554
        format = SampleBzrFormat()
 
1555
        format.features = {"nested-trees": "unknown"}
 
1556
        self.assertRaises(errors.MissingFeature, format.check_support_status,
 
1557
            True)
 
1558
        self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
 
1559
        SampleBzrFormat.register_feature("nested-trees")
 
1560
        format.check_support_status(True)
 
1561
 
 
1562
    def test_feature_already_registered(self):
 
1563
        # a feature can only be registered once
 
1564
        self.addCleanup(SampleBzrFormat.unregister_feature, "nested-trees")
 
1565
        SampleBzrFormat.register_feature("nested-trees")
 
1566
        self.assertRaises(errors.FeatureAlreadyRegistered,
 
1567
            SampleBzrFormat.register_feature, "nested-trees")
 
1568
 
 
1569
    def test_feature_with_space(self):
 
1570
        # spaces are not allowed in feature names
 
1571
        self.assertRaises(ValueError, SampleBzrFormat.register_feature,
 
1572
            "nested trees")