~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

Merge up bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2006-2013, 2016 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2005, 2006, 2007 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
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for the BzrDir facility and any format specific tests.
18
18
 
19
 
For interface contract tests, see tests/per_bzr_dir.
 
19
For interface contract tests, see tests/bzr_dir_implementations.
20
20
"""
21
21
 
22
 
import os
 
22
import os.path
 
23
from StringIO import StringIO
23
24
import subprocess
24
25
import sys
25
26
 
26
27
from bzrlib import (
27
 
    branch,
28
28
    bzrdir,
29
 
    config,
30
 
    controldir,
31
29
    errors,
32
30
    help_topics,
33
 
    lock,
34
31
    repository,
35
 
    revision as _mod_revision,
36
 
    osutils,
37
 
    remote,
38
 
    transport as _mod_transport,
 
32
    symbol_versioning,
39
33
    urlutils,
40
34
    win32utils,
41
 
    workingtree_3,
42
 
    workingtree_4,
 
35
    workingtree,
43
36
    )
44
37
import bzrlib.branch
45
 
from bzrlib.branchfmt.fullhistory import BzrBranchFormat5
46
 
from bzrlib.errors import (
47
 
    NotBranchError,
48
 
    NoColocatedBranchSupport,
49
 
    UnknownFormatError,
50
 
    UnsupportedFormatError,
51
 
    )
 
38
from bzrlib.errors import (NotBranchError,
 
39
                           UnknownFormatError,
 
40
                           UnsupportedFormatError,
 
41
                           )
52
42
from bzrlib.tests import (
53
43
    TestCase,
54
 
    TestCaseWithMemoryTransport,
55
44
    TestCaseWithTransport,
56
45
    TestSkipped,
 
46
    test_sftp_transport
57
47
    )
58
 
from bzrlib.tests import(
59
 
    http_server,
60
 
    http_utils,
 
48
from bzrlib.tests.http_server import HttpServer
 
49
from bzrlib.tests.http_utils import (
 
50
    TestCaseWithTwoWebservers,
 
51
    HTTPServerRedirecting,
61
52
    )
62
53
from bzrlib.tests.test_http import TestWithTransport_pycurl
63
 
from bzrlib.transport import (
64
 
    memory,
65
 
    pathfilter,
66
 
    )
 
54
from bzrlib.transport import get_transport
67
55
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
 
56
from bzrlib.transport.memory import MemoryServer
 
57
from bzrlib.repofmt import knitrepo, weaverepo
71
58
 
72
59
 
73
60
class TestDefaultFormat(TestCase):
74
61
 
75
62
    def test_get_set_default_format(self):
76
63
        old_format = bzrdir.BzrDirFormat.get_default_format()
77
 
        # default is BzrDirMetaFormat1
78
 
        self.assertIsInstance(old_format, bzrdir.BzrDirMetaFormat1)
79
 
        controldir.ControlDirFormat._set_default_format(SampleBzrDirFormat())
 
64
        # default is BzrDirFormat6
 
65
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
 
66
        bzrdir.BzrDirFormat._set_default_format(SampleBzrDirFormat())
80
67
        # creating a bzr dir should now create an instrumented dir.
81
68
        try:
82
69
            result = bzrdir.BzrDir.create('memory:///')
83
 
            self.assertIsInstance(result, SampleBzrDir)
 
70
            self.failUnless(isinstance(result, SampleBzrDir))
84
71
        finally:
85
 
            controldir.ControlDirFormat._set_default_format(old_format)
 
72
            bzrdir.BzrDirFormat._set_default_format(old_format)
86
73
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
87
74
 
88
75
 
89
 
class DeprecatedBzrDirFormat(bzrdir.BzrDirFormat):
90
 
    """A deprecated bzr dir format."""
91
 
 
92
 
 
93
76
class TestFormatRegistry(TestCase):
94
77
 
95
78
    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',
 
79
        my_format_registry = bzrdir.BzrDirFormatRegistry()
 
80
        my_format_registry.register('weave', bzrdir.BzrDirFormat6,
 
81
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
82
            ' repositories', deprecated=True)
 
83
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir', 
 
84
            'BzrDirFormat6', 'Format registered lazily', deprecated=True)
 
85
        my_format_registry.register_metadir('knit',
104
86
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
105
87
            'Format using knits',
106
88
            )
107
89
        my_format_registry.set_default('knit')
108
 
        bzrdir.register_metadir(my_format_registry,
 
90
        my_format_registry.register_metadir(
109
91
            'branch6',
110
92
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
111
93
            'Experimental successor to knit.  Use at your own risk.',
112
94
            branch_format='bzrlib.branch.BzrBranchFormat6',
113
95
            experimental=True)
114
 
        bzrdir.register_metadir(my_format_registry,
 
96
        my_format_registry.register_metadir(
115
97
            'hidden format',
116
98
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
117
99
            'Experimental successor to knit.  Use at your own risk.',
118
100
            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)
 
101
        my_format_registry.register('hiddenweave', bzrdir.BzrDirFormat6,
 
102
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
103
            ' repositories', hidden=True)
 
104
        my_format_registry.register_lazy('hiddenlazy', 'bzrlib.bzrdir',
 
105
            'BzrDirFormat6', 'Format registered lazily', deprecated=True,
 
106
            hidden=True)
124
107
        return my_format_registry
125
108
 
126
109
    def test_format_registry(self):
127
110
        my_format_registry = self.make_format_registry()
128
111
        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)
 
112
        self.assertIsInstance(my_bzrdir, bzrdir.BzrDirFormat6)
 
113
        my_bzrdir = my_format_registry.make_bzrdir('weave')
 
114
        self.assertIsInstance(my_bzrdir, bzrdir.BzrDirFormat6)
132
115
        my_bzrdir = my_format_registry.make_bzrdir('default')
133
 
        self.assertIsInstance(my_bzrdir.repository_format,
 
116
        self.assertIsInstance(my_bzrdir.repository_format, 
134
117
            knitrepo.RepositoryFormatKnit1)
135
118
        my_bzrdir = my_format_registry.make_bzrdir('knit')
136
 
        self.assertIsInstance(my_bzrdir.repository_format,
 
119
        self.assertIsInstance(my_bzrdir.repository_format, 
137
120
            knitrepo.RepositoryFormatKnit1)
138
121
        my_bzrdir = my_format_registry.make_bzrdir('branch6')
139
122
        self.assertIsInstance(my_bzrdir.get_branch_format(),
143
126
        my_format_registry = self.make_format_registry()
144
127
        self.assertEqual('Format registered lazily',
145
128
                         my_format_registry.get_help('lazy'))
146
 
        self.assertEqual('Format using knits',
 
129
        self.assertEqual('Format using knits', 
147
130
                         my_format_registry.get_help('knit'))
148
 
        self.assertEqual('Format using knits',
 
131
        self.assertEqual('Format using knits', 
149
132
                         my_format_registry.get_help('default'))
150
 
        self.assertEqual('Some format.  Slower and unawesome and deprecated.',
151
 
                         my_format_registry.get_help('deprecated'))
152
 
 
 
133
        self.assertEqual('Pre-0.8 format.  Slower and does not support'
 
134
                         ' checkouts or shared repositories', 
 
135
                         my_format_registry.get_help('weave'))
 
136
        
153
137
    def test_help_topic(self):
154
138
        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')
 
139
        topics.register('formats', self.make_format_registry().help_topic, 
 
140
                        'Directory formats')
 
141
        topic = topics.get_detail('formats')
 
142
        new, rest = topic.split('Experimental formats')
162
143
        experimental, deprecated = rest.split('Deprecated formats')
163
 
        self.assertContainsRe(new, 'formats-help')
164
 
        self.assertContainsRe(new,
 
144
        self.assertContainsRe(new, 'These formats can be used')
 
145
        self.assertContainsRe(new, 
165
146
                ':knit:\n    \(native\) \(default\) Format using knits\n')
166
 
        self.assertContainsRe(experimental,
 
147
        self.assertContainsRe(experimental, 
167
148
                ':branch6:\n    \(native\) Experimental successor to knit')
168
 
        self.assertContainsRe(deprecated,
 
149
        self.assertContainsRe(deprecated, 
169
150
                ':lazy:\n    \(native\) Format registered lazily\n')
170
151
        self.assertNotContainsRe(new, 'hidden')
171
152
 
172
153
    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()
 
154
        default_factory = bzrdir.format_registry.get('default')
 
155
        old_default = [k for k, v in bzrdir.format_registry.iteritems()
175
156
                       if v == default_factory and k != 'default'][0]
176
 
        controldir.format_registry.set_default_repository('dirstate-with-subtree')
 
157
        bzrdir.format_registry.set_default_repository('dirstate-with-subtree')
177
158
        try:
178
 
            self.assertIs(controldir.format_registry.get('dirstate-with-subtree'),
179
 
                          controldir.format_registry.get('default'))
 
159
            self.assertIs(bzrdir.format_registry.get('dirstate-with-subtree'),
 
160
                          bzrdir.format_registry.get('default'))
180
161
            self.assertIs(
181
 
                repository.format_registry.get_default().__class__,
 
162
                repository.RepositoryFormat.get_default_format().__class__,
182
163
                knitrepo.RepositoryFormatKnit3)
183
164
        finally:
184
 
            controldir.format_registry.set_default_repository(old_default)
 
165
            bzrdir.format_registry.set_default_repository(old_default)
185
166
 
186
167
    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
 
 
 
168
        a_registry = bzrdir.BzrDirFormatRegistry()
 
169
        a_registry.register('weave', bzrdir.BzrDirFormat6,
 
170
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
171
            ' repositories', deprecated=True)
 
172
        a_registry.register('weavealias', bzrdir.BzrDirFormat6,
 
173
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
174
            ' repositories', deprecated=True, alias=True)
 
175
        self.assertEqual(frozenset(['weavealias']), a_registry.aliases())
 
176
    
196
177
 
197
178
class SampleBranch(bzrlib.branch.Branch):
198
179
    """A dummy branch for guess what, dummy use."""
201
182
        self.bzrdir = dir
202
183
 
203
184
 
204
 
class SampleRepository(bzrlib.repository.Repository):
205
 
    """A dummy repo."""
206
 
 
207
 
    def __init__(self, dir):
208
 
        self.bzrdir = dir
209
 
 
210
 
 
211
185
class SampleBzrDir(bzrdir.BzrDir):
212
186
    """A sample BzrDir implementation to allow testing static methods."""
213
187
 
214
188
    def create_repository(self, shared=False):
215
 
        """See ControlDir.create_repository."""
 
189
        """See BzrDir.create_repository."""
216
190
        return "A repository"
217
191
 
218
192
    def open_repository(self):
219
 
        """See ControlDir.open_repository."""
220
 
        return SampleRepository(self)
 
193
        """See BzrDir.open_repository."""
 
194
        return "A repository"
221
195
 
222
 
    def create_branch(self, name=None):
223
 
        """See ControlDir.create_branch."""
224
 
        if name is not None:
225
 
            raise NoColocatedBranchSupport(self)
 
196
    def create_branch(self):
 
197
        """See BzrDir.create_branch."""
226
198
        return SampleBranch(self)
227
199
 
228
200
    def create_workingtree(self):
229
 
        """See ControlDir.create_workingtree."""
 
201
        """See BzrDir.create_workingtree."""
230
202
        return "A tree"
231
203
 
232
204
 
233
205
class SampleBzrDirFormat(bzrdir.BzrDirFormat):
234
206
    """A sample format
235
207
 
236
 
    this format is initializable, unsupported to aid in testing the
 
208
    this format is initializable, unsupported to aid in testing the 
237
209
    open and open_downlevel routines.
238
210
    """
239
211
 
253
225
    def open(self, transport, _found=None):
254
226
        return "opened branch."
255
227
 
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
 
 
274
228
 
275
229
class TestBzrDirFormat(TestCaseWithTransport):
276
230
    """Tests for the BzrDirFormat facility."""
278
232
    def test_find_format(self):
279
233
        # is the right format object found for a branch?
280
234
        # create a branch with a few known format objects.
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()
 
235
        # this is not quite the same as 
 
236
        t = get_transport(self.get_url())
290
237
        self.build_tree(["foo/", "bar/"], transport=t)
291
238
        def check_format(format, url):
292
239
            format.initialize(url)
293
 
            t = _mod_transport.get_transport_from_path(url)
 
240
            t = get_transport(url)
294
241
            found_format = bzrdir.BzrDirFormat.find_format(t)
295
 
            self.assertIsInstance(found_format, format.__class__)
296
 
        check_format(BzrDirFormatTest1(), "foo")
297
 
        check_format(BzrDirFormatTest2(), "bar")
298
 
 
 
242
            self.failUnless(isinstance(found_format, format.__class__))
 
243
        check_format(bzrdir.BzrDirFormat5(), "foo")
 
244
        check_format(bzrdir.BzrDirFormat6(), "bar")
 
245
        
299
246
    def test_find_format_nothing_there(self):
300
247
        self.assertRaises(NotBranchError,
301
248
                          bzrdir.BzrDirFormat.find_format,
302
 
                          _mod_transport.get_transport_from_path('.'))
 
249
                          get_transport('.'))
303
250
 
304
251
    def test_find_format_unknown_format(self):
305
 
        t = self.get_transport()
 
252
        t = get_transport(self.get_url())
306
253
        t.mkdir('.bzr')
307
254
        t.put_bytes('.bzr/branch-format', '')
308
255
        self.assertRaises(UnknownFormatError,
309
256
                          bzrdir.BzrDirFormat.find_format,
310
 
                          _mod_transport.get_transport_from_path('.'))
 
257
                          get_transport('.'))
311
258
 
312
259
    def test_register_unregister_format(self):
313
260
        format = SampleBzrDirFormat()
315
262
        # make a bzrdir
316
263
        format.initialize(url)
317
264
        # register a format for it.
318
 
        bzrdir.BzrProber.formats.register(format.get_format_string(), format)
 
265
        bzrdir.BzrDirFormat.register_format(format)
319
266
        # which bzrdir.Open will refuse (not supported)
320
267
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open, url)
321
268
        # which bzrdir.open_containing will refuse (not supported)
322
269
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open_containing, url)
323
270
        # but open_downlevel will work
324
 
        t = _mod_transport.get_transport_from_url(url)
 
271
        t = get_transport(url)
325
272
        self.assertEqual(format.open(t), bzrdir.BzrDir.open_unsupported(url))
326
273
        # unregister the format
327
 
        bzrdir.BzrProber.formats.remove(format.get_format_string())
 
274
        bzrdir.BzrDirFormat.unregister_format(format)
328
275
        # now open_downlevel should fail too.
329
276
        self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
330
277
 
337
284
    def test_create_branch_and_repo_under_shared(self):
338
285
        # creating a branch and repo in a shared repo uses the
339
286
        # shared repository
340
 
        format = controldir.format_registry.make_bzrdir('knit')
 
287
        format = bzrdir.format_registry.make_bzrdir('knit')
341
288
        self.make_repository('.', shared=True, format=format)
342
289
        branch = bzrdir.BzrDir.create_branch_and_repo(
343
290
            self.get_url('child'), format=format)
345
292
                          branch.bzrdir.open_repository)
346
293
 
347
294
    def test_create_branch_and_repo_under_shared_force_new(self):
348
 
        # creating a branch and repo in a shared repo can be forced to
 
295
        # creating a branch and repo in a shared repo can be forced to 
349
296
        # make a new repo
350
 
        format = controldir.format_registry.make_bzrdir('knit')
 
297
        format = bzrdir.format_registry.make_bzrdir('knit')
351
298
        self.make_repository('.', shared=True, format=format)
352
299
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
353
300
                                                      force_new_repo=True,
356
303
 
357
304
    def test_create_standalone_working_tree(self):
358
305
        format = SampleBzrDirFormat()
359
 
        # note this is deliberately readonly, as this failure should
 
306
        # note this is deliberately readonly, as this failure should 
360
307
        # occur before any writes.
361
308
        self.assertRaises(errors.NotLocalUrl,
362
309
                          bzrdir.BzrDir.create_standalone_workingtree,
363
310
                          self.get_readonly_url(), format=format)
364
 
        tree = bzrdir.BzrDir.create_standalone_workingtree('.',
 
311
        tree = bzrdir.BzrDir.create_standalone_workingtree('.', 
365
312
                                                           format=format)
366
313
        self.assertEqual('A tree', tree)
367
314
 
368
315
    def test_create_standalone_working_tree_under_shared_repo(self):
369
316
        # create standalone working tree always makes a repo.
370
 
        format = controldir.format_registry.make_bzrdir('knit')
 
317
        format = bzrdir.format_registry.make_bzrdir('knit')
371
318
        self.make_repository('.', shared=True, format=format)
372
 
        # note this is deliberately readonly, as this failure should
 
319
        # note this is deliberately readonly, as this failure should 
373
320
        # occur before any writes.
374
321
        self.assertRaises(errors.NotLocalUrl,
375
322
                          bzrdir.BzrDir.create_standalone_workingtree,
376
323
                          self.get_readonly_url('child'), format=format)
377
 
        tree = bzrdir.BzrDir.create_standalone_workingtree('child',
 
324
        tree = bzrdir.BzrDir.create_standalone_workingtree('child', 
378
325
            format=format)
379
326
        tree.bzrdir.open_repository()
380
327
 
381
328
    def test_create_branch_convenience(self):
382
329
        # outside a repo the default convenience output is a repo+branch_tree
383
 
        format = controldir.format_registry.make_bzrdir('knit')
 
330
        format = bzrdir.format_registry.make_bzrdir('knit')
384
331
        branch = bzrdir.BzrDir.create_branch_convenience('.', format=format)
385
332
        branch.bzrdir.open_workingtree()
386
333
        branch.bzrdir.open_repository()
387
334
 
388
335
    def test_create_branch_convenience_possible_transports(self):
389
336
        """Check that the optional 'possible_transports' is recognized"""
390
 
        format = controldir.format_registry.make_bzrdir('knit')
 
337
        format = bzrdir.format_registry.make_bzrdir('knit')
391
338
        t = self.get_transport()
392
339
        branch = bzrdir.BzrDir.create_branch_convenience(
393
340
            '.', format=format, possible_transports=[t])
396
343
 
397
344
    def test_create_branch_convenience_root(self):
398
345
        """Creating a branch at the root of a fs should work."""
399
 
        self.vfs_transport_factory = memory.MemoryServer
 
346
        self.vfs_transport_factory = MemoryServer
400
347
        # 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(),
 
348
        format = bzrdir.format_registry.make_bzrdir('knit')
 
349
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(), 
403
350
                                                         format=format)
404
351
        self.assertRaises(errors.NoWorkingTree,
405
352
                          branch.bzrdir.open_workingtree)
408
355
    def test_create_branch_convenience_under_shared_repo(self):
409
356
        # inside a repo the default convenience output is a branch+ follow the
410
357
        # repo tree policy
411
 
        format = controldir.format_registry.make_bzrdir('knit')
 
358
        format = bzrdir.format_registry.make_bzrdir('knit')
412
359
        self.make_repository('.', shared=True, format=format)
413
360
        branch = bzrdir.BzrDir.create_branch_convenience('child',
414
361
            format=format)
415
362
        branch.bzrdir.open_workingtree()
416
363
        self.assertRaises(errors.NoRepositoryPresent,
417
364
                          branch.bzrdir.open_repository)
418
 
 
 
365
            
419
366
    def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
420
367
        # inside a repo the default convenience output is a branch+ follow the
421
368
        # repo tree policy but we can override that
422
 
        format = controldir.format_registry.make_bzrdir('knit')
 
369
        format = bzrdir.format_registry.make_bzrdir('knit')
423
370
        self.make_repository('.', shared=True, format=format)
424
371
        branch = bzrdir.BzrDir.create_branch_convenience('child',
425
372
            force_new_tree=False, format=format)
427
374
                          branch.bzrdir.open_workingtree)
428
375
        self.assertRaises(errors.NoRepositoryPresent,
429
376
                          branch.bzrdir.open_repository)
430
 
 
 
377
            
431
378
    def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
432
379
        # inside a repo the default convenience output is a branch+ follow the
433
380
        # repo tree policy
434
 
        format = controldir.format_registry.make_bzrdir('knit')
 
381
        format = bzrdir.format_registry.make_bzrdir('knit')
435
382
        repo = self.make_repository('.', shared=True, format=format)
436
383
        repo.set_make_working_trees(False)
437
 
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
384
        branch = bzrdir.BzrDir.create_branch_convenience('child', 
438
385
                                                         format=format)
439
386
        self.assertRaises(errors.NoWorkingTree,
440
387
                          branch.bzrdir.open_workingtree)
444
391
    def test_create_branch_convenience_under_shared_repo_no_tree_policy_force_tree(self):
445
392
        # inside a repo the default convenience output is a branch+ follow the
446
393
        # repo tree policy but we can override that
447
 
        format = controldir.format_registry.make_bzrdir('knit')
 
394
        format = bzrdir.format_registry.make_bzrdir('knit')
448
395
        repo = self.make_repository('.', shared=True, format=format)
449
396
        repo.set_make_working_trees(False)
450
397
        branch = bzrdir.BzrDir.create_branch_convenience('child',
456
403
    def test_create_branch_convenience_under_shared_repo_force_new_repo(self):
457
404
        # inside a repo the default convenience output is overridable to give
458
405
        # repo+branch+tree
459
 
        format = controldir.format_registry.make_bzrdir('knit')
 
406
        format = bzrdir.format_registry.make_bzrdir('knit')
460
407
        self.make_repository('.', shared=True, format=format)
461
408
        branch = bzrdir.BzrDir.create_branch_convenience('child',
462
409
            force_new_repo=True, format=format)
470
417
        """The default acquisition policy should create a standalone branch."""
471
418
        my_bzrdir = self.make_bzrdir('.')
472
419
        repo_policy = my_bzrdir.determine_repository_policy()
473
 
        repo, is_new = repo_policy.acquire_repository()
 
420
        repo = repo_policy.acquire_repository()
474
421
        self.assertEqual(repo.bzrdir.root_transport.base,
475
422
                         my_bzrdir.root_transport.base)
476
423
        self.assertFalse(repo.is_shared())
477
424
 
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())
628
 
 
629
425
 
630
426
class ChrootedTests(TestCaseWithTransport):
631
427
    """A support class that provides readonly urls outside the local namespace.
637
433
 
638
434
    def setUp(self):
639
435
        super(ChrootedTests, self).setUp()
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))
 
436
        if not self.vfs_transport_factory == MemoryServer:
 
437
            self.transport_readonly_server = HttpServer
645
438
 
646
439
    def test_open_containing(self):
647
440
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
654
447
        branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
655
448
        self.assertEqual('g/p/q', relpath)
656
449
 
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
 
 
744
450
    def test_open_containing_from_transport(self):
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')))
 
451
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
 
452
                          get_transport(self.get_readonly_url('')))
 
453
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
 
454
                          get_transport(self.get_readonly_url('g/p/q')))
752
455
        control = bzrdir.BzrDir.create(self.get_url())
753
456
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
754
 
            _mod_transport.get_transport_from_url(
755
 
                self.get_readonly_url('')))
 
457
            get_transport(self.get_readonly_url('')))
756
458
        self.assertEqual('', relpath)
757
459
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
758
 
            _mod_transport.get_transport_from_url(
759
 
                self.get_readonly_url('g/p/q')))
 
460
            get_transport(self.get_readonly_url('g/p/q')))
760
461
        self.assertEqual('g/p/q', relpath)
761
462
 
762
463
    def test_open_containing_tree_or_branch(self):
 
464
        def local_branch_path(branch):
 
465
             return os.path.realpath(
 
466
                urlutils.local_path_from_url(branch.base))
 
467
 
763
468
        self.make_branch_and_tree('topdir')
764
469
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
765
470
            'topdir/foo')
766
471
        self.assertEqual(os.path.realpath('topdir'),
767
472
                         os.path.realpath(tree.basedir))
768
473
        self.assertEqual(os.path.realpath('topdir'),
769
 
                         self.local_branch_path(branch))
 
474
                         local_branch_path(branch))
770
475
        self.assertIs(tree.bzrdir, branch.bzrdir)
771
476
        self.assertEqual('foo', relpath)
772
477
        # opening from non-local should not return the tree
780
485
            'topdir/foo')
781
486
        self.assertIs(tree, None)
782
487
        self.assertEqual(os.path.realpath('topdir/foo'),
783
 
                         self.local_branch_path(branch))
 
488
                         local_branch_path(branch))
784
489
        self.assertEqual('', relpath)
785
490
 
786
491
    def test_open_tree_or_branch(self):
 
492
        def local_branch_path(branch):
 
493
             return os.path.realpath(
 
494
                urlutils.local_path_from_url(branch.base))
 
495
 
787
496
        self.make_branch_and_tree('topdir')
788
497
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir')
789
498
        self.assertEqual(os.path.realpath('topdir'),
790
499
                         os.path.realpath(tree.basedir))
791
500
        self.assertEqual(os.path.realpath('topdir'),
792
 
                         self.local_branch_path(branch))
 
501
                         local_branch_path(branch))
793
502
        self.assertIs(tree.bzrdir, branch.bzrdir)
794
503
        # opening from non-local should not return the tree
795
504
        tree, branch = bzrdir.BzrDir.open_tree_or_branch(
800
509
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir/foo')
801
510
        self.assertIs(tree, None)
802
511
        self.assertEqual(os.path.realpath('topdir/foo'),
803
 
                         self.local_branch_path(branch))
 
512
                         local_branch_path(branch))
804
513
 
805
514
    def test_open_from_transport(self):
806
515
        # transport pointing at bzrdir should give a bzrdir with root transport
807
516
        # set to the given transport
808
517
        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)
 
518
        transport = get_transport(self.get_url())
 
519
        opened_bzrdir = bzrdir.BzrDir.open_from_transport(transport)
 
520
        self.assertEqual(transport.base, opened_bzrdir.root_transport.base)
812
521
        self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
813
 
 
 
522
        
814
523
    def test_open_from_transport_no_bzrdir(self):
815
 
        t = self.get_transport()
816
 
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport, t)
 
524
        transport = get_transport(self.get_url())
 
525
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
 
526
                          transport)
817
527
 
818
528
    def test_open_from_transport_bzrdir_in_parent(self):
819
529
        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)
 
530
        transport = get_transport(self.get_url())
 
531
        transport.mkdir('subdir')
 
532
        transport = transport.clone('subdir')
 
533
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
 
534
                          transport)
824
535
 
825
536
    def test_sprout_recursive(self):
826
 
        tree = self.make_branch_and_tree('tree1',
827
 
                                         format='development-subtree')
 
537
        tree = self.make_branch_and_tree('tree1', format='dirstate-with-subtree')
828
538
        sub_tree = self.make_branch_and_tree('tree1/subtree',
829
 
            format='development-subtree')
830
 
        sub_tree.set_root_id('subtree-root')
 
539
            format='dirstate-with-subtree')
831
540
        tree.add_reference(sub_tree)
832
541
        self.build_tree(['tree1/subtree/file'])
833
542
        sub_tree.add('file')
834
543
        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'))
 
544
        tree.bzrdir.sprout('tree2')
 
545
        self.failUnlessExists('tree2/subtree/file')
840
546
 
841
547
    def test_cloning_metadir(self):
842
548
        """Ensure that cloning metadir is suitable"""
845
551
        branch = self.make_branch('branch', format='knit')
846
552
        format = branch.bzrdir.cloning_metadir()
847
553
        self.assertIsInstance(format.workingtree_format,
848
 
            workingtree_4.WorkingTreeFormat6)
 
554
            workingtree.WorkingTreeFormat3)
849
555
 
850
556
    def test_sprout_recursive_treeless(self):
851
557
        tree = self.make_branch_and_tree('tree1',
852
 
            format='development-subtree')
 
558
            format='dirstate-with-subtree')
853
559
        sub_tree = self.make_branch_and_tree('tree1/subtree',
854
 
            format='development-subtree')
 
560
            format='dirstate-with-subtree')
855
561
        tree.add_reference(sub_tree)
856
562
        self.build_tree(['tree1/subtree/file'])
857
563
        sub_tree.add('file')
858
564
        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
565
        tree.bzrdir.destroy_workingtree()
863
 
        # FIXME: subtree/.bzr is left here which allows the test to pass (or
864
 
        # fail :-( ) -- vila 20100909
865
566
        repo = self.make_repository('repo', shared=True,
866
 
            format='development-subtree')
 
567
            format='dirstate-with-subtree')
867
568
        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')
 
569
        tree.bzrdir.sprout('repo/tree2')
 
570
        self.failUnlessExists('repo/tree2/subtree')
 
571
        self.failIfExists('repo/tree2/subtree/file')
880
572
 
881
573
    def make_foo_bar_baz(self):
882
574
        foo = bzrdir.BzrDir.create_branch_convenience('foo').bzrdir
886
578
 
887
579
    def test_find_bzrdirs(self):
888
580
        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))
 
581
        transport = get_transport(self.get_url())
 
582
        self.assertEqualBzrdirs([baz, foo, bar],
 
583
                                bzrdir.BzrDir.find_bzrdirs(transport))
923
584
 
924
585
    def test_find_bzrdirs_list_current(self):
925
586
        def list_current(transport):
926
587
            return [s for s in transport.list_dir('') if s != 'baz']
927
588
 
928
589
        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))
 
590
        transport = get_transport(self.get_url())
 
591
        self.assertEqualBzrdirs([foo, bar],
 
592
                                bzrdir.BzrDir.find_bzrdirs(transport,
 
593
                                    list_current=list_current))
 
594
 
933
595
 
934
596
    def test_find_bzrdirs_evaluate(self):
935
597
        def evaluate(bzrdir):
936
598
            try:
937
599
                repo = bzrdir.open_repository()
938
 
            except errors.NoRepositoryPresent:
 
600
            except NoRepositoryPresent:
939
601
                return True, bzrdir.root_transport.base
940
602
            else:
941
603
                return False, bzrdir.root_transport.base
942
604
 
943
605
        foo, bar, baz = self.make_foo_bar_baz()
944
 
        t = self.get_transport()
 
606
        transport = get_transport(self.get_url())
945
607
        self.assertEqual([baz.root_transport.base, foo.root_transport.base],
946
 
                         list(bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate)))
 
608
                         list(bzrdir.BzrDir.find_bzrdirs(transport,
 
609
                                                         evaluate=evaluate)))
947
610
 
948
611
    def assertEqualBzrdirs(self, first, second):
949
612
        first = list(first)
956
619
        root = self.make_repository('', shared=True)
957
620
        foo, bar, baz = self.make_foo_bar_baz()
958
621
        qux = self.make_bzrdir('foo/qux')
959
 
        t = self.get_transport()
960
 
        branches = bzrdir.BzrDir.find_branches(t)
 
622
        transport = get_transport(self.get_url())
 
623
        branches = bzrdir.BzrDir.find_branches(transport)
961
624
        self.assertEqual(baz.root_transport.base, branches[0].base)
962
625
        self.assertEqual(foo.root_transport.base, branches[1].base)
963
626
        self.assertEqual(bar.root_transport.base, branches[2].base)
964
627
 
965
628
        # ensure this works without a top-level repo
966
 
        branches = bzrdir.BzrDir.find_branches(t.clone('foo'))
 
629
        branches = bzrdir.BzrDir.find_branches(transport.clone('foo'))
967
630
        self.assertEqual(foo.root_transport.base, branches[0].base)
968
631
        self.assertEqual(bar.root_transport.base, branches[1].base)
969
632
 
970
633
 
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
 
 
985
 
 
986
634
class TestMeta1DirFormat(TestCaseWithTransport):
987
635
    """Tests specific to the meta1 dir format."""
988
636
 
992
640
        branch_base = t.clone('branch').base
993
641
        self.assertEqual(branch_base, dir.get_branch_transport(None).base)
994
642
        self.assertEqual(branch_base,
995
 
                         dir.get_branch_transport(BzrBranchFormat5()).base)
 
643
                         dir.get_branch_transport(bzrlib.branch.BzrBranchFormat5()).base)
996
644
        repository_base = t.clone('repository').base
997
645
        self.assertEqual(repository_base, dir.get_repository_transport(None).base)
998
 
        repository_format = repository.format_registry.get_default()
999
646
        self.assertEqual(repository_base,
1000
 
                         dir.get_repository_transport(repository_format).base)
 
647
                         dir.get_repository_transport(weaverepo.RepositoryFormat7()).base)
1001
648
        checkout_base = t.clone('checkout').base
1002
649
        self.assertEqual(checkout_base, dir.get_workingtree_transport(None).base)
1003
650
        self.assertEqual(checkout_base,
1004
 
                         dir.get_workingtree_transport(workingtree_3.WorkingTreeFormat3()).base)
 
651
                         dir.get_workingtree_transport(workingtree.WorkingTreeFormat3()).base)
1005
652
 
1006
653
    def test_meta1dir_uses_lockdir(self):
1007
654
        """Meta1 format uses a LockDir to guard the whole directory, not a file."""
1015
662
        Metadirs should compare equal iff they have the same repo, branch and
1016
663
        tree formats.
1017
664
        """
1018
 
        mydir = controldir.format_registry.make_bzrdir('knit')
 
665
        mydir = bzrdir.format_registry.make_bzrdir('knit')
1019
666
        self.assertEqual(mydir, mydir)
1020
667
        self.assertFalse(mydir != mydir)
1021
 
        otherdir = controldir.format_registry.make_bzrdir('knit')
 
668
        otherdir = bzrdir.format_registry.make_bzrdir('knit')
1022
669
        self.assertEqual(otherdir, mydir)
1023
670
        self.assertFalse(otherdir != mydir)
1024
 
        otherdir2 = controldir.format_registry.make_bzrdir('development-subtree')
 
671
        otherdir2 = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
1025
672
        self.assertNotEqual(otherdir2, mydir)
1026
673
        self.assertFalse(otherdir2 == mydir)
1027
674
 
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.assertEqual("required", dir._format.features.get("bar"))
1036
 
        tree.bzrdir.update_feature_flags({"bar": None, "nonexistant": None})
1037
 
        dir = bzrdir.BzrDir.open('tree')
1038
 
        self.assertEqual({}, dir._format.features)
1039
 
 
1040
675
    def test_needs_conversion_different_working_tree(self):
1041
676
        # 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)
 
677
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
678
        # test with 
 
679
        new_default = bzrdir.format_registry.make_bzrdir('dirstate')
 
680
        bzrdir.BzrDirFormat._set_default_format(new_default)
 
681
        try:
 
682
            tree = self.make_branch_and_tree('tree', format='knit')
 
683
            self.assertTrue(tree.bzrdir.needs_format_conversion())
 
684
        finally:
 
685
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
686
 
 
687
 
 
688
class TestFormat5(TestCaseWithTransport):
 
689
    """Tests specific to the version 5 bzrdir format."""
 
690
 
 
691
    def test_same_lockfiles_between_tree_repo_branch(self):
 
692
        # this checks that only a single lockfiles instance is created 
 
693
        # for format 5 objects
 
694
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
695
        def check_dir_components_use_same_lock(dir):
 
696
            ctrl_1 = dir.open_repository().control_files
 
697
            ctrl_2 = dir.open_branch().control_files
 
698
            ctrl_3 = dir.open_workingtree()._control_files
 
699
            self.assertTrue(ctrl_1 is ctrl_2)
 
700
            self.assertTrue(ctrl_2 is ctrl_3)
 
701
        check_dir_components_use_same_lock(dir)
 
702
        # and if we open it normally.
 
703
        dir = bzrdir.BzrDir.open(self.get_url())
 
704
        check_dir_components_use_same_lock(dir)
 
705
    
 
706
    def test_can_convert(self):
 
707
        # format 5 dirs are convertable
 
708
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
709
        self.assertTrue(dir.can_convert_format())
 
710
    
 
711
    def test_needs_conversion(self):
 
712
        # format 5 dirs need a conversion if they are not the default.
 
713
        # and they start of not the default.
 
714
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
715
        bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirFormat5())
 
716
        try:
 
717
            dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
718
            self.assertFalse(dir.needs_format_conversion())
 
719
        finally:
 
720
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
721
        self.assertTrue(dir.needs_format_conversion())
 
722
 
 
723
 
 
724
class TestFormat6(TestCaseWithTransport):
 
725
    """Tests specific to the version 6 bzrdir format."""
 
726
 
 
727
    def test_same_lockfiles_between_tree_repo_branch(self):
 
728
        # this checks that only a single lockfiles instance is created 
 
729
        # for format 6 objects
 
730
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
731
        def check_dir_components_use_same_lock(dir):
 
732
            ctrl_1 = dir.open_repository().control_files
 
733
            ctrl_2 = dir.open_branch().control_files
 
734
            ctrl_3 = dir.open_workingtree()._control_files
 
735
            self.assertTrue(ctrl_1 is ctrl_2)
 
736
            self.assertTrue(ctrl_2 is ctrl_3)
 
737
        check_dir_components_use_same_lock(dir)
 
738
        # and if we open it normally.
 
739
        dir = bzrdir.BzrDir.open(self.get_url())
 
740
        check_dir_components_use_same_lock(dir)
 
741
    
 
742
    def test_can_convert(self):
 
743
        # format 6 dirs are convertable
 
744
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
745
        self.assertTrue(dir.can_convert_format())
 
746
    
 
747
    def test_needs_conversion(self):
 
748
        # format 6 dirs need an conversion if they are not the default.
 
749
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
750
        bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirMetaFormat1())
 
751
        try:
 
752
            dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
753
            self.assertTrue(dir.needs_format_conversion())
 
754
        finally:
 
755
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
756
 
 
757
 
 
758
class NotBzrDir(bzrlib.bzrdir.BzrDir):
 
759
    """A non .bzr based control directory."""
 
760
 
 
761
    def __init__(self, transport, format):
 
762
        self._format = format
 
763
        self.root_transport = transport
 
764
        self.transport = transport.clone('.not')
 
765
 
 
766
 
 
767
class NotBzrDirFormat(bzrlib.bzrdir.BzrDirFormat):
 
768
    """A test class representing any non-.bzr based disk format."""
 
769
 
 
770
    def initialize_on_transport(self, transport):
 
771
        """Initialize a new .not dir in the base directory of a Transport."""
 
772
        transport.mkdir('.not')
 
773
        return self.open(transport)
 
774
 
 
775
    def open(self, transport):
 
776
        """Open this directory."""
 
777
        return NotBzrDir(transport, self)
 
778
 
 
779
    @classmethod
 
780
    def _known_formats(self):
 
781
        return set([NotBzrDirFormat()])
 
782
 
 
783
    @classmethod
 
784
    def probe_transport(self, transport):
 
785
        """Our format is present if the transport ends in '.not/'."""
 
786
        if transport.has('.not'):
 
787
            return NotBzrDirFormat()
 
788
 
 
789
 
 
790
class TestNotBzrDir(TestCaseWithTransport):
 
791
    """Tests for using the bzrdir api with a non .bzr based disk format.
 
792
    
 
793
    If/when one of these is in the core, we can let the implementation tests
 
794
    verify this works.
 
795
    """
 
796
 
 
797
    def test_create_and_find_format(self):
 
798
        # create a .notbzr dir 
 
799
        format = NotBzrDirFormat()
 
800
        dir = format.initialize(self.get_url())
 
801
        self.assertIsInstance(dir, NotBzrDir)
 
802
        # now probe for it.
 
803
        bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
 
804
        try:
 
805
            found = bzrlib.bzrdir.BzrDirFormat.find_format(
 
806
                get_transport(self.get_url()))
 
807
            self.assertIsInstance(found, NotBzrDirFormat)
 
808
        finally:
 
809
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
 
810
 
 
811
    def test_included_in_known_formats(self):
 
812
        bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
 
813
        try:
 
814
            formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
 
815
            for format in formats:
 
816
                if isinstance(format, NotBzrDirFormat):
 
817
                    return
 
818
            self.fail("No NotBzrDirFormat in %s" % formats)
 
819
        finally:
 
820
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
1062
821
 
1063
822
 
1064
823
class NonLocalTests(TestCaseWithTransport):
1066
825
 
1067
826
    def setUp(self):
1068
827
        super(NonLocalTests, self).setUp()
1069
 
        self.vfs_transport_factory = memory.MemoryServer
1070
 
 
 
828
        self.vfs_transport_factory = MemoryServer
 
829
    
1071
830
    def test_create_branch_convenience(self):
1072
831
        # outside a repo the default convenience output is a repo+branch_tree
1073
 
        format = controldir.format_registry.make_bzrdir('knit')
 
832
        format = bzrdir.format_registry.make_bzrdir('knit')
1074
833
        branch = bzrdir.BzrDir.create_branch_convenience(
1075
834
            self.get_url('foo'), format=format)
1076
835
        self.assertRaises(errors.NoWorkingTree,
1079
838
 
1080
839
    def test_create_branch_convenience_force_tree_not_local_fails(self):
1081
840
        # outside a repo the default convenience output is a repo+branch_tree
1082
 
        format = controldir.format_registry.make_bzrdir('knit')
 
841
        format = bzrdir.format_registry.make_bzrdir('knit')
1083
842
        self.assertRaises(errors.NotLocalUrl,
1084
843
            bzrdir.BzrDir.create_branch_convenience,
1085
844
            self.get_url('foo'),
1086
845
            force_new_tree=True,
1087
846
            format=format)
1088
 
        t = self.get_transport()
 
847
        t = get_transport(self.get_url('.'))
1089
848
        self.assertFalse(t.has('foo'))
1090
849
 
1091
850
    def test_clone(self):
1092
851
        # clone into a nonlocal path works
1093
 
        format = controldir.format_registry.make_bzrdir('knit')
 
852
        format = bzrdir.format_registry.make_bzrdir('knit')
1094
853
        branch = bzrdir.BzrDir.create_branch_convenience('local',
1095
854
                                                         format=format)
1096
855
        branch.bzrdir.open_workingtree()
1107
866
        my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
1108
867
        checkout_format = my_bzrdir.checkout_metadir()
1109
868
        self.assertIsInstance(checkout_format.workingtree_format,
1110
 
                              workingtree_4.WorkingTreeFormat4)
1111
 
 
1112
 
 
1113
 
class TestHTTPRedirections(object):
1114
 
    """Test redirection between two http servers.
 
869
                              workingtree.WorkingTreeFormat3)
 
870
 
 
871
 
 
872
class TestHTTPRedirectionLoop(object):
 
873
    """Test redirection loop between two http servers.
1115
874
 
1116
875
    This MUST be used by daughter classes that also inherit from
1117
876
    TestCaseWithTwoWebservers.
1118
877
 
1119
878
    We can't inherit directly from TestCaseWithTwoWebservers or the
1120
879
    test framework will try to create an instance which cannot
1121
 
    run, its implementation being incomplete.
 
880
    run, its implementation being incomplete. 
1122
881
    """
1123
882
 
 
883
    # Should be defined by daughter classes to ensure redirection
 
884
    # still use the same transport implementation (not currently
 
885
    # enforced as it's a bit tricky to get right (see the FIXME
 
886
    # in BzrDir.open_from_transport for the unique use case so
 
887
    # far)
 
888
    _qualifier = None
 
889
 
1124
890
    def create_transport_readonly_server(self):
1125
 
        # We don't set the http protocol version, relying on the default
1126
 
        return http_utils.HTTPServerRedirecting()
 
891
        return HTTPServerRedirecting()
1127
892
 
1128
893
    def create_transport_secondary_server(self):
1129
 
        # We don't set the http protocol version, relying on the default
1130
 
        return http_utils.HTTPServerRedirecting()
 
894
        return HTTPServerRedirecting()
1131
895
 
1132
896
    def setUp(self):
1133
 
        super(TestHTTPRedirections, self).setUp()
 
897
        # Both servers redirect to each server creating a loop
 
898
        super(TestHTTPRedirectionLoop, self).setUp()
1134
899
        # The redirections will point to the new server
1135
900
        self.new_server = self.get_readonly_server()
1136
901
        # The requests to the old server will be redirected
1137
902
        self.old_server = self.get_secondary_server()
1138
903
        # Configure the redirections
1139
904
        self.old_server.redirect_to(self.new_server.host, self.new_server.port)
 
905
        self.new_server.redirect_to(self.old_server.host, self.old_server.port)
 
906
 
 
907
    def _qualified_url(self, host, port):
 
908
        return 'http+%s://%s:%s' % (self._qualifier, host, port)
1140
909
 
1141
910
    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
911
        # Starting from either server should loop
1145
 
        old_url = self._qualified_url(self.old_server.host,
 
912
        old_url = self._qualified_url(self.old_server.host, 
1146
913
                                      self.old_server.port)
1147
914
        oldt = self._transport(old_url)
1148
915
        self.assertRaises(errors.NotBranchError,
1149
916
                          bzrdir.BzrDir.open_from_transport, oldt)
1150
 
        new_url = self._qualified_url(self.new_server.host,
 
917
        new_url = self._qualified_url(self.new_server.host, 
1151
918
                                      self.new_server.port)
1152
919
        newt = self._transport(new_url)
1153
920
        self.assertRaises(errors.NotBranchError,
1154
921
                          bzrdir.BzrDir.open_from_transport, newt)
1155
922
 
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):
 
923
 
 
924
class TestHTTPRedirections_urllib(TestHTTPRedirectionLoop,
 
925
                                  TestCaseWithTwoWebservers):
1169
926
    """Tests redirections for urllib implementation"""
1170
927
 
 
928
    _qualifier = 'urllib'
1171
929
    _transport = HttpTransport_urllib
1172
930
 
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
931
 
1179
932
 
1180
933
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1181
 
                                  TestHTTPRedirections,
1182
 
                                  http_utils.TestCaseWithTwoWebservers):
 
934
                                  TestHTTPRedirectionLoop,
 
935
                                  TestCaseWithTwoWebservers):
1183
936
    """Tests redirections for pycurl implementation"""
1184
937
 
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
 
938
    _qualifier = 'pycurl'
1213
939
 
1214
940
 
1215
941
class TestDotBzrHidden(TestCaseWithTransport):
1231
957
            raise TestSkipped('unable to make file hidden without pywin32 library')
1232
958
        b = bzrdir.BzrDir.create('.')
1233
959
        self.build_tree(['a'])
1234
 
        self.assertEqual(['a'], self.get_ls())
 
960
        self.assertEquals(['a'], self.get_ls())
1235
961
 
1236
962
    def test_dot_bzr_hidden_with_url(self):
1237
963
        if sys.platform == 'win32' and not win32utils.has_win32file:
1238
964
            raise TestSkipped('unable to make file hidden without pywin32 library')
1239
965
        b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
1240
966
        self.build_tree(['a'])
1241
 
        self.assertEqual(['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.assertEqual(format.as_string(),
1479
 
            "First line\n"
1480
 
            "required foo\n")
1481
 
        format.features["another"] = "optional"
1482
 
        self.assertEqual(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.assertEqual(
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.assertEqual({}, 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.assertEqual("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.assertEqual("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.assertEqual(format1, format1)
1529
 
        self.assertEqual(format1, format2)
1530
 
        format3 = SampleBzrFormat()
1531
 
        self.assertNotEqual(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")
 
967
        self.assertEquals(['a'], self.get_ls())