~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

  • Committer: Martin Packman
  • Date: 2011-12-23 19:38:22 UTC
  • mto: This revision was merged to the branch mainline in revision 6405.
  • Revision ID: martin.packman@canonical.com-20111223193822-hesheea4o8aqwexv
Accept and document passing the medium rather than transport for smart connections

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