~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

(jelmer) Support upgrading between the 2a and development-colo formats.
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

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