~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

  • Committer: Aaron Bentley
  • Date: 2007-12-12 15:17:13 UTC
  • mto: This revision was merged to the branch mainline in revision 3113.
  • Revision ID: abentley@panoramicfeedback.com-20071212151713-ox5n8rlx8m3nsspy
Add support for reconfiguring repositories into branches or trees

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
2
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
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
 
16
 
 
17
"""Tests for the BzrDir facility and any format specific tests.
 
18
 
 
19
For interface contract tests, see tests/bzr_dir_implementations.
 
20
"""
 
21
 
 
22
import os.path
 
23
from StringIO import StringIO
 
24
import subprocess
 
25
import sys
 
26
 
 
27
from bzrlib import (
 
28
    bzrdir,
 
29
    errors,
 
30
    help_topics,
 
31
    repository,
 
32
    symbol_versioning,
 
33
    urlutils,
 
34
    win32utils,
 
35
    workingtree,
 
36
    )
 
37
import bzrlib.branch
 
38
from bzrlib.errors import (NotBranchError,
 
39
                           UnknownFormatError,
 
40
                           UnsupportedFormatError,
 
41
                           )
 
42
from bzrlib.symbol_versioning import (
 
43
    zero_ninetyone,
 
44
    )
 
45
from bzrlib.tests import (
 
46
    TestCase,
 
47
    TestCaseWithTransport,
 
48
    TestSkipped,
 
49
    test_sftp_transport
 
50
    )
 
51
from bzrlib.tests.HttpServer import HttpServer
 
52
from bzrlib.tests.HTTPTestUtil import (
 
53
    TestCaseWithTwoWebservers,
 
54
    HTTPServerRedirecting,
 
55
    )
 
56
from bzrlib.tests.test_http import TestWithTransport_pycurl
 
57
from bzrlib.transport import get_transport
 
58
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
59
from bzrlib.transport.memory import MemoryServer
 
60
from bzrlib.repofmt import knitrepo, weaverepo
 
61
 
 
62
 
 
63
class TestDefaultFormat(TestCase):
 
64
 
 
65
    def test_get_set_default_format(self):
 
66
        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())
 
72
        # creating a bzr dir should now create an instrumented dir.
 
73
        try:
 
74
            result = bzrdir.BzrDir.create('memory:///')
 
75
            self.failUnless(isinstance(result, SampleBzrDir))
 
76
        finally:
 
77
            self.applyDeprecated(symbol_versioning.zero_fourteen,
 
78
                bzrdir.BzrDirFormat.set_default_format, old_format)
 
79
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
 
80
 
 
81
 
 
82
class TestFormatRegistry(TestCase):
 
83
 
 
84
    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',
 
92
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
 
93
            'Format using knits',
 
94
            )
 
95
        my_format_registry.set_default('knit')
 
96
        my_format_registry.register_metadir(
 
97
            'branch6',
 
98
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
 
99
            'Experimental successor to knit.  Use at your own risk.',
 
100
            branch_format='bzrlib.branch.BzrBranchFormat6',
 
101
            experimental=True)
 
102
        my_format_registry.register_metadir(
 
103
            'hidden format',
 
104
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
 
105
            'Experimental successor to knit.  Use at your own risk.',
 
106
            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)
 
113
        return my_format_registry
 
114
 
 
115
    def test_format_registry(self):
 
116
        my_format_registry = self.make_format_registry()
 
117
        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)
 
121
        my_bzrdir = my_format_registry.make_bzrdir('default')
 
122
        self.assertIsInstance(my_bzrdir.repository_format, 
 
123
            knitrepo.RepositoryFormatKnit1)
 
124
        my_bzrdir = my_format_registry.make_bzrdir('knit')
 
125
        self.assertIsInstance(my_bzrdir.repository_format, 
 
126
            knitrepo.RepositoryFormatKnit1)
 
127
        my_bzrdir = my_format_registry.make_bzrdir('branch6')
 
128
        self.assertIsInstance(my_bzrdir.get_branch_format(),
 
129
                              bzrlib.branch.BzrBranchFormat6)
 
130
 
 
131
    def test_get_help(self):
 
132
        my_format_registry = self.make_format_registry()
 
133
        self.assertEqual('Format registered lazily',
 
134
                         my_format_registry.get_help('lazy'))
 
135
        self.assertEqual('Format using knits', 
 
136
                         my_format_registry.get_help('knit'))
 
137
        self.assertEqual('Format using knits', 
 
138
                         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
        
 
143
    def test_help_topic(self):
 
144
        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')
 
149
        experimental, deprecated = rest.split('Deprecated formats')
 
150
        self.assertContainsRe(new, 'These formats can be used')
 
151
        self.assertContainsRe(new, 
 
152
                ':knit:\n    \(native\) \(default\) Format using knits\n')
 
153
        self.assertContainsRe(experimental, 
 
154
                ':branch6:\n    \(native\) Experimental successor to knit')
 
155
        self.assertContainsRe(deprecated, 
 
156
                ':lazy:\n    \(native\) Format registered lazily\n')
 
157
        self.assertNotContainsRe(new, 'hidden')
 
158
 
 
159
    def test_set_default_repository(self):
 
160
        default_factory = bzrdir.format_registry.get('default')
 
161
        old_default = [k for k, v in bzrdir.format_registry.iteritems()
 
162
                       if v == default_factory and k != 'default'][0]
 
163
        bzrdir.format_registry.set_default_repository('dirstate-with-subtree')
 
164
        try:
 
165
            self.assertIs(bzrdir.format_registry.get('dirstate-with-subtree'),
 
166
                          bzrdir.format_registry.get('default'))
 
167
            self.assertIs(
 
168
                repository.RepositoryFormat.get_default_format().__class__,
 
169
                knitrepo.RepositoryFormatKnit3)
 
170
        finally:
 
171
            bzrdir.format_registry.set_default_repository(old_default)
 
172
 
 
173
 
 
174
class SampleBranch(bzrlib.branch.Branch):
 
175
    """A dummy branch for guess what, dummy use."""
 
176
 
 
177
    def __init__(self, dir):
 
178
        self.bzrdir = dir
 
179
 
 
180
 
 
181
class SampleBzrDir(bzrdir.BzrDir):
 
182
    """A sample BzrDir implementation to allow testing static methods."""
 
183
 
 
184
    def create_repository(self, shared=False):
 
185
        """See BzrDir.create_repository."""
 
186
        return "A repository"
 
187
 
 
188
    def open_repository(self):
 
189
        """See BzrDir.open_repository."""
 
190
        return "A repository"
 
191
 
 
192
    def create_branch(self):
 
193
        """See BzrDir.create_branch."""
 
194
        return SampleBranch(self)
 
195
 
 
196
    def create_workingtree(self):
 
197
        """See BzrDir.create_workingtree."""
 
198
        return "A tree"
 
199
 
 
200
 
 
201
class SampleBzrDirFormat(bzrdir.BzrDirFormat):
 
202
    """A sample format
 
203
 
 
204
    this format is initializable, unsupported to aid in testing the 
 
205
    open and open_downlevel routines.
 
206
    """
 
207
 
 
208
    def get_format_string(self):
 
209
        """See BzrDirFormat.get_format_string()."""
 
210
        return "Sample .bzr dir format."
 
211
 
 
212
    def initialize_on_transport(self, t):
 
213
        """Create a bzr dir."""
 
214
        t.mkdir('.bzr')
 
215
        t.put_bytes('.bzr/branch-format', self.get_format_string())
 
216
        return SampleBzrDir(t, self)
 
217
 
 
218
    def is_supported(self):
 
219
        return False
 
220
 
 
221
    def open(self, transport, _found=None):
 
222
        return "opened branch."
 
223
 
 
224
 
 
225
class TestBzrDirFormat(TestCaseWithTransport):
 
226
    """Tests for the BzrDirFormat facility."""
 
227
 
 
228
    def test_find_format(self):
 
229
        # is the right format object found for a branch?
 
230
        # create a branch with a few known format objects.
 
231
        # this is not quite the same as 
 
232
        t = get_transport(self.get_url())
 
233
        self.build_tree(["foo/", "bar/"], transport=t)
 
234
        def check_format(format, url):
 
235
            format.initialize(url)
 
236
            t = get_transport(url)
 
237
            found_format = bzrdir.BzrDirFormat.find_format(t)
 
238
            self.failUnless(isinstance(found_format, format.__class__))
 
239
        check_format(bzrdir.BzrDirFormat5(), "foo")
 
240
        check_format(bzrdir.BzrDirFormat6(), "bar")
 
241
        
 
242
    def test_find_format_nothing_there(self):
 
243
        self.assertRaises(NotBranchError,
 
244
                          bzrdir.BzrDirFormat.find_format,
 
245
                          get_transport('.'))
 
246
 
 
247
    def test_find_format_unknown_format(self):
 
248
        t = get_transport(self.get_url())
 
249
        t.mkdir('.bzr')
 
250
        t.put_bytes('.bzr/branch-format', '')
 
251
        self.assertRaises(UnknownFormatError,
 
252
                          bzrdir.BzrDirFormat.find_format,
 
253
                          get_transport('.'))
 
254
 
 
255
    def test_register_unregister_format(self):
 
256
        format = SampleBzrDirFormat()
 
257
        url = self.get_url()
 
258
        # make a bzrdir
 
259
        format.initialize(url)
 
260
        # register a format for it.
 
261
        bzrdir.BzrDirFormat.register_format(format)
 
262
        # which bzrdir.Open will refuse (not supported)
 
263
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open, url)
 
264
        # which bzrdir.open_containing will refuse (not supported)
 
265
        self.assertRaises(UnsupportedFormatError, bzrdir.BzrDir.open_containing, url)
 
266
        # but open_downlevel will work
 
267
        t = get_transport(url)
 
268
        self.assertEqual(format.open(t), bzrdir.BzrDir.open_unsupported(url))
 
269
        # unregister the format
 
270
        bzrdir.BzrDirFormat.unregister_format(format)
 
271
        # now open_downlevel should fail too.
 
272
        self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
 
273
 
 
274
    def test_create_repository_deprecated(self):
 
275
        # new interface is to make the bzrdir, then a repository within that.
 
276
        format = SampleBzrDirFormat()
 
277
        repo = self.applyDeprecated(zero_ninetyone,
 
278
                bzrdir.BzrDir.create_repository,
 
279
                self.get_url(), format=format)
 
280
        self.assertEqual('A repository', repo)
 
281
 
 
282
    def test_create_repository_shared(self):
 
283
        # new interface is to make the bzrdir, then a repository within that.
 
284
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
285
        repo = self.applyDeprecated(zero_ninetyone,
 
286
                bzrdir.BzrDir.create_repository,
 
287
                '.', shared=True)
 
288
        self.assertTrue(repo.is_shared())
 
289
 
 
290
    def test_create_repository_nonshared(self):
 
291
        # new interface is to make the bzrdir, then a repository within that.
 
292
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
293
        repo = self.applyDeprecated(zero_ninetyone,
 
294
                bzrdir.BzrDir.create_repository,
 
295
                '.')
 
296
        self.assertFalse(repo.is_shared())
 
297
 
 
298
    def test_create_repository_under_shared(self):
 
299
        # an explicit create_repository always does so.
 
300
        # we trust the format is right from the 'create_repository test'
 
301
        # new interface is to make the bzrdir, then a repository within that.
 
302
        format = bzrdir.format_registry.make_bzrdir('knit')
 
303
        self.make_repository('.', shared=True, format=format)
 
304
        repo = self.applyDeprecated(zero_ninetyone,
 
305
                bzrdir.BzrDir.create_repository,
 
306
                self.get_url('child'),
 
307
                format=format)
 
308
        self.assertTrue(isinstance(repo, repository.Repository))
 
309
        self.assertTrue(repo.bzrdir.root_transport.base.endswith('child/'))
 
310
 
 
311
    def test_create_branch_and_repo_uses_default(self):
 
312
        format = SampleBzrDirFormat()
 
313
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(),
 
314
                                                      format=format)
 
315
        self.assertTrue(isinstance(branch, SampleBranch))
 
316
 
 
317
    def test_create_branch_and_repo_under_shared(self):
 
318
        # creating a branch and repo in a shared repo uses the
 
319
        # shared repository
 
320
        format = bzrdir.format_registry.make_bzrdir('knit')
 
321
        self.make_repository('.', shared=True, format=format)
 
322
        branch = bzrdir.BzrDir.create_branch_and_repo(
 
323
            self.get_url('child'), format=format)
 
324
        self.assertRaises(errors.NoRepositoryPresent,
 
325
                          branch.bzrdir.open_repository)
 
326
 
 
327
    def test_create_branch_and_repo_under_shared_force_new(self):
 
328
        # creating a branch and repo in a shared repo can be forced to 
 
329
        # make a new repo
 
330
        format = bzrdir.format_registry.make_bzrdir('knit')
 
331
        self.make_repository('.', shared=True, format=format)
 
332
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url('child'),
 
333
                                                      force_new_repo=True,
 
334
                                                      format=format)
 
335
        branch.bzrdir.open_repository()
 
336
 
 
337
    def test_create_standalone_working_tree(self):
 
338
        format = SampleBzrDirFormat()
 
339
        # note this is deliberately readonly, as this failure should 
 
340
        # occur before any writes.
 
341
        self.assertRaises(errors.NotLocalUrl,
 
342
                          bzrdir.BzrDir.create_standalone_workingtree,
 
343
                          self.get_readonly_url(), format=format)
 
344
        tree = bzrdir.BzrDir.create_standalone_workingtree('.', 
 
345
                                                           format=format)
 
346
        self.assertEqual('A tree', tree)
 
347
 
 
348
    def test_create_standalone_working_tree_under_shared_repo(self):
 
349
        # create standalone working tree always makes a repo.
 
350
        format = bzrdir.format_registry.make_bzrdir('knit')
 
351
        self.make_repository('.', shared=True, format=format)
 
352
        # note this is deliberately readonly, as this failure should 
 
353
        # occur before any writes.
 
354
        self.assertRaises(errors.NotLocalUrl,
 
355
                          bzrdir.BzrDir.create_standalone_workingtree,
 
356
                          self.get_readonly_url('child'), format=format)
 
357
        tree = bzrdir.BzrDir.create_standalone_workingtree('child', 
 
358
            format=format)
 
359
        tree.bzrdir.open_repository()
 
360
 
 
361
    def test_create_branch_convenience(self):
 
362
        # outside a repo the default convenience output is a repo+branch_tree
 
363
        format = bzrdir.format_registry.make_bzrdir('knit')
 
364
        branch = bzrdir.BzrDir.create_branch_convenience('.', format=format)
 
365
        branch.bzrdir.open_workingtree()
 
366
        branch.bzrdir.open_repository()
 
367
 
 
368
    def test_create_branch_convenience_possible_transports(self):
 
369
        """Check that the optional 'possible_transports' is recognized"""
 
370
        format = bzrdir.format_registry.make_bzrdir('knit')
 
371
        t = self.get_transport()
 
372
        branch = bzrdir.BzrDir.create_branch_convenience(
 
373
            '.', format=format, possible_transports=[t])
 
374
        branch.bzrdir.open_workingtree()
 
375
        branch.bzrdir.open_repository()
 
376
 
 
377
    def test_create_branch_convenience_root(self):
 
378
        """Creating a branch at the root of a fs should work."""
 
379
        self.vfs_transport_factory = MemoryServer
 
380
        # outside a repo the default convenience output is a repo+branch_tree
 
381
        format = bzrdir.format_registry.make_bzrdir('knit')
 
382
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(), 
 
383
                                                         format=format)
 
384
        self.assertRaises(errors.NoWorkingTree,
 
385
                          branch.bzrdir.open_workingtree)
 
386
        branch.bzrdir.open_repository()
 
387
 
 
388
    def test_create_branch_convenience_under_shared_repo(self):
 
389
        # inside a repo the default convenience output is a branch+ follow the
 
390
        # repo tree policy
 
391
        format = bzrdir.format_registry.make_bzrdir('knit')
 
392
        self.make_repository('.', shared=True, format=format)
 
393
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
394
            format=format)
 
395
        branch.bzrdir.open_workingtree()
 
396
        self.assertRaises(errors.NoRepositoryPresent,
 
397
                          branch.bzrdir.open_repository)
 
398
            
 
399
    def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
 
400
        # inside a repo the default convenience output is a branch+ follow the
 
401
        # repo tree policy but we can override that
 
402
        format = bzrdir.format_registry.make_bzrdir('knit')
 
403
        self.make_repository('.', shared=True, format=format)
 
404
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
405
            force_new_tree=False, format=format)
 
406
        self.assertRaises(errors.NoWorkingTree,
 
407
                          branch.bzrdir.open_workingtree)
 
408
        self.assertRaises(errors.NoRepositoryPresent,
 
409
                          branch.bzrdir.open_repository)
 
410
            
 
411
    def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
 
412
        # inside a repo the default convenience output is a branch+ follow the
 
413
        # repo tree policy
 
414
        format = bzrdir.format_registry.make_bzrdir('knit')
 
415
        repo = self.make_repository('.', shared=True, format=format)
 
416
        repo.set_make_working_trees(False)
 
417
        branch = bzrdir.BzrDir.create_branch_convenience('child', 
 
418
                                                         format=format)
 
419
        self.assertRaises(errors.NoWorkingTree,
 
420
                          branch.bzrdir.open_workingtree)
 
421
        self.assertRaises(errors.NoRepositoryPresent,
 
422
                          branch.bzrdir.open_repository)
 
423
 
 
424
    def test_create_branch_convenience_under_shared_repo_no_tree_policy_force_tree(self):
 
425
        # inside a repo the default convenience output is a branch+ follow the
 
426
        # repo tree policy but we can override that
 
427
        format = bzrdir.format_registry.make_bzrdir('knit')
 
428
        repo = self.make_repository('.', shared=True, format=format)
 
429
        repo.set_make_working_trees(False)
 
430
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
431
            force_new_tree=True, format=format)
 
432
        branch.bzrdir.open_workingtree()
 
433
        self.assertRaises(errors.NoRepositoryPresent,
 
434
                          branch.bzrdir.open_repository)
 
435
 
 
436
    def test_create_branch_convenience_under_shared_repo_force_new_repo(self):
 
437
        # inside a repo the default convenience output is overridable to give
 
438
        # repo+branch+tree
 
439
        format = bzrdir.format_registry.make_bzrdir('knit')
 
440
        self.make_repository('.', shared=True, format=format)
 
441
        branch = bzrdir.BzrDir.create_branch_convenience('child',
 
442
            force_new_repo=True, format=format)
 
443
        branch.bzrdir.open_repository()
 
444
        branch.bzrdir.open_workingtree()
 
445
 
 
446
 
 
447
class ChrootedTests(TestCaseWithTransport):
 
448
    """A support class that provides readonly urls outside the local namespace.
 
449
 
 
450
    This is done by checking if self.transport_server is a MemoryServer. if it
 
451
    is then we are chrooted already, if it is not then an HttpServer is used
 
452
    for readonly urls.
 
453
    """
 
454
 
 
455
    def setUp(self):
 
456
        super(ChrootedTests, self).setUp()
 
457
        if not self.vfs_transport_factory == MemoryServer:
 
458
            self.transport_readonly_server = HttpServer
 
459
 
 
460
    def test_open_containing(self):
 
461
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
 
462
                          self.get_readonly_url(''))
 
463
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
 
464
                          self.get_readonly_url('g/p/q'))
 
465
        control = bzrdir.BzrDir.create(self.get_url())
 
466
        branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url(''))
 
467
        self.assertEqual('', relpath)
 
468
        branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
 
469
        self.assertEqual('g/p/q', relpath)
 
470
 
 
471
    def test_open_containing_from_transport(self):
 
472
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
 
473
                          get_transport(self.get_readonly_url('')))
 
474
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
 
475
                          get_transport(self.get_readonly_url('g/p/q')))
 
476
        control = bzrdir.BzrDir.create(self.get_url())
 
477
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
 
478
            get_transport(self.get_readonly_url('')))
 
479
        self.assertEqual('', relpath)
 
480
        branch, relpath = bzrdir.BzrDir.open_containing_from_transport(
 
481
            get_transport(self.get_readonly_url('g/p/q')))
 
482
        self.assertEqual('g/p/q', relpath)
 
483
 
 
484
    def test_open_containing_tree_or_branch(self):
 
485
        def local_branch_path(branch):
 
486
             return os.path.realpath(
 
487
                urlutils.local_path_from_url(branch.base))
 
488
 
 
489
        self.make_branch_and_tree('topdir')
 
490
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
491
            'topdir/foo')
 
492
        self.assertEqual(os.path.realpath('topdir'),
 
493
                         os.path.realpath(tree.basedir))
 
494
        self.assertEqual(os.path.realpath('topdir'),
 
495
                         local_branch_path(branch))
 
496
        self.assertIs(tree.bzrdir, branch.bzrdir)
 
497
        self.assertEqual('foo', relpath)
 
498
        # opening from non-local should not return the tree
 
499
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
500
            self.get_readonly_url('topdir/foo'))
 
501
        self.assertEqual(None, tree)
 
502
        self.assertEqual('foo', relpath)
 
503
        # without a tree:
 
504
        self.make_branch('topdir/foo')
 
505
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
 
506
            'topdir/foo')
 
507
        self.assertIs(tree, None)
 
508
        self.assertEqual(os.path.realpath('topdir/foo'),
 
509
                         local_branch_path(branch))
 
510
        self.assertEqual('', relpath)
 
511
 
 
512
    def test_open_from_transport(self):
 
513
        # transport pointing at bzrdir should give a bzrdir with root transport
 
514
        # set to the given transport
 
515
        control = bzrdir.BzrDir.create(self.get_url())
 
516
        transport = get_transport(self.get_url())
 
517
        opened_bzrdir = bzrdir.BzrDir.open_from_transport(transport)
 
518
        self.assertEqual(transport.base, opened_bzrdir.root_transport.base)
 
519
        self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
 
520
        
 
521
    def test_open_from_transport_no_bzrdir(self):
 
522
        transport = get_transport(self.get_url())
 
523
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
 
524
                          transport)
 
525
 
 
526
    def test_open_from_transport_bzrdir_in_parent(self):
 
527
        control = bzrdir.BzrDir.create(self.get_url())
 
528
        transport = get_transport(self.get_url())
 
529
        transport.mkdir('subdir')
 
530
        transport = transport.clone('subdir')
 
531
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
 
532
                          transport)
 
533
 
 
534
    def test_sprout_recursive(self):
 
535
        tree = self.make_branch_and_tree('tree1', format='dirstate-with-subtree')
 
536
        sub_tree = self.make_branch_and_tree('tree1/subtree',
 
537
            format='dirstate-with-subtree')
 
538
        tree.add_reference(sub_tree)
 
539
        self.build_tree(['tree1/subtree/file'])
 
540
        sub_tree.add('file')
 
541
        tree.commit('Initial commit')
 
542
        tree.bzrdir.sprout('tree2')
 
543
        self.failUnlessExists('tree2/subtree/file')
 
544
 
 
545
    def test_cloning_metadir(self):
 
546
        """Ensure that cloning metadir is suitable"""
 
547
        bzrdir = self.make_bzrdir('bzrdir')
 
548
        bzrdir.cloning_metadir()
 
549
        branch = self.make_branch('branch', format='knit')
 
550
        format = branch.bzrdir.cloning_metadir()
 
551
        self.assertIsInstance(format.workingtree_format,
 
552
            workingtree.WorkingTreeFormat3)
 
553
 
 
554
    def test_sprout_recursive_treeless(self):
 
555
        tree = self.make_branch_and_tree('tree1',
 
556
            format='dirstate-with-subtree')
 
557
        sub_tree = self.make_branch_and_tree('tree1/subtree',
 
558
            format='dirstate-with-subtree')
 
559
        tree.add_reference(sub_tree)
 
560
        self.build_tree(['tree1/subtree/file'])
 
561
        sub_tree.add('file')
 
562
        tree.commit('Initial commit')
 
563
        tree.bzrdir.destroy_workingtree()
 
564
        repo = self.make_repository('repo', shared=True,
 
565
            format='dirstate-with-subtree')
 
566
        repo.set_make_working_trees(False)
 
567
        tree.bzrdir.sprout('repo/tree2')
 
568
        self.failUnlessExists('repo/tree2/subtree')
 
569
        self.failIfExists('repo/tree2/subtree/file')
 
570
 
 
571
 
 
572
class TestMeta1DirFormat(TestCaseWithTransport):
 
573
    """Tests specific to the meta1 dir format."""
 
574
 
 
575
    def test_right_base_dirs(self):
 
576
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
577
        t = dir.transport
 
578
        branch_base = t.clone('branch').base
 
579
        self.assertEqual(branch_base, dir.get_branch_transport(None).base)
 
580
        self.assertEqual(branch_base,
 
581
                         dir.get_branch_transport(bzrlib.branch.BzrBranchFormat5()).base)
 
582
        repository_base = t.clone('repository').base
 
583
        self.assertEqual(repository_base, dir.get_repository_transport(None).base)
 
584
        self.assertEqual(repository_base,
 
585
                         dir.get_repository_transport(weaverepo.RepositoryFormat7()).base)
 
586
        checkout_base = t.clone('checkout').base
 
587
        self.assertEqual(checkout_base, dir.get_workingtree_transport(None).base)
 
588
        self.assertEqual(checkout_base,
 
589
                         dir.get_workingtree_transport(workingtree.WorkingTreeFormat3()).base)
 
590
 
 
591
    def test_meta1dir_uses_lockdir(self):
 
592
        """Meta1 format uses a LockDir to guard the whole directory, not a file."""
 
593
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
594
        t = dir.transport
 
595
        self.assertIsDirectory('branch-lock', t)
 
596
 
 
597
    def test_comparison(self):
 
598
        """Equality and inequality behave properly.
 
599
 
 
600
        Metadirs should compare equal iff they have the same repo, branch and
 
601
        tree formats.
 
602
        """
 
603
        mydir = bzrdir.format_registry.make_bzrdir('knit')
 
604
        self.assertEqual(mydir, mydir)
 
605
        self.assertFalse(mydir != mydir)
 
606
        otherdir = bzrdir.format_registry.make_bzrdir('knit')
 
607
        self.assertEqual(otherdir, mydir)
 
608
        self.assertFalse(otherdir != mydir)
 
609
        otherdir2 = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
 
610
        self.assertNotEqual(otherdir2, mydir)
 
611
        self.assertFalse(otherdir2 == mydir)
 
612
 
 
613
    def test_needs_conversion_different_working_tree(self):
 
614
        # meta1dirs need an conversion if any element is not the default.
 
615
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
616
        # test with 
 
617
        new_default = bzrdir.format_registry.make_bzrdir('dirstate')
 
618
        bzrdir.BzrDirFormat._set_default_format(new_default)
 
619
        try:
 
620
            tree = self.make_branch_and_tree('tree', format='knit')
 
621
            self.assertTrue(tree.bzrdir.needs_format_conversion())
 
622
        finally:
 
623
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
624
 
 
625
 
 
626
class TestFormat5(TestCaseWithTransport):
 
627
    """Tests specific to the version 5 bzrdir format."""
 
628
 
 
629
    def test_same_lockfiles_between_tree_repo_branch(self):
 
630
        # this checks that only a single lockfiles instance is created 
 
631
        # for format 5 objects
 
632
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
633
        def check_dir_components_use_same_lock(dir):
 
634
            ctrl_1 = dir.open_repository().control_files
 
635
            ctrl_2 = dir.open_branch().control_files
 
636
            ctrl_3 = dir.open_workingtree()._control_files
 
637
            self.assertTrue(ctrl_1 is ctrl_2)
 
638
            self.assertTrue(ctrl_2 is ctrl_3)
 
639
        check_dir_components_use_same_lock(dir)
 
640
        # and if we open it normally.
 
641
        dir = bzrdir.BzrDir.open(self.get_url())
 
642
        check_dir_components_use_same_lock(dir)
 
643
    
 
644
    def test_can_convert(self):
 
645
        # format 5 dirs are convertable
 
646
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
647
        self.assertTrue(dir.can_convert_format())
 
648
    
 
649
    def test_needs_conversion(self):
 
650
        # format 5 dirs need a conversion if they are not the default.
 
651
        # and they start of not the default.
 
652
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
653
        bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirFormat5())
 
654
        try:
 
655
            dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
656
            self.assertFalse(dir.needs_format_conversion())
 
657
        finally:
 
658
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
659
        self.assertTrue(dir.needs_format_conversion())
 
660
 
 
661
 
 
662
class TestFormat6(TestCaseWithTransport):
 
663
    """Tests specific to the version 6 bzrdir format."""
 
664
 
 
665
    def test_same_lockfiles_between_tree_repo_branch(self):
 
666
        # this checks that only a single lockfiles instance is created 
 
667
        # for format 6 objects
 
668
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
669
        def check_dir_components_use_same_lock(dir):
 
670
            ctrl_1 = dir.open_repository().control_files
 
671
            ctrl_2 = dir.open_branch().control_files
 
672
            ctrl_3 = dir.open_workingtree()._control_files
 
673
            self.assertTrue(ctrl_1 is ctrl_2)
 
674
            self.assertTrue(ctrl_2 is ctrl_3)
 
675
        check_dir_components_use_same_lock(dir)
 
676
        # and if we open it normally.
 
677
        dir = bzrdir.BzrDir.open(self.get_url())
 
678
        check_dir_components_use_same_lock(dir)
 
679
    
 
680
    def test_can_convert(self):
 
681
        # format 6 dirs are convertable
 
682
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
683
        self.assertTrue(dir.can_convert_format())
 
684
    
 
685
    def test_needs_conversion(self):
 
686
        # format 6 dirs need an conversion if they are not the default.
 
687
        old_format = bzrdir.BzrDirFormat.get_default_format()
 
688
        bzrdir.BzrDirFormat._set_default_format(bzrdir.BzrDirMetaFormat1())
 
689
        try:
 
690
            dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
691
            self.assertTrue(dir.needs_format_conversion())
 
692
        finally:
 
693
            bzrdir.BzrDirFormat._set_default_format(old_format)
 
694
 
 
695
 
 
696
class NotBzrDir(bzrlib.bzrdir.BzrDir):
 
697
    """A non .bzr based control directory."""
 
698
 
 
699
    def __init__(self, transport, format):
 
700
        self._format = format
 
701
        self.root_transport = transport
 
702
        self.transport = transport.clone('.not')
 
703
 
 
704
 
 
705
class NotBzrDirFormat(bzrlib.bzrdir.BzrDirFormat):
 
706
    """A test class representing any non-.bzr based disk format."""
 
707
 
 
708
    def initialize_on_transport(self, transport):
 
709
        """Initialize a new .not dir in the base directory of a Transport."""
 
710
        transport.mkdir('.not')
 
711
        return self.open(transport)
 
712
 
 
713
    def open(self, transport):
 
714
        """Open this directory."""
 
715
        return NotBzrDir(transport, self)
 
716
 
 
717
    @classmethod
 
718
    def _known_formats(self):
 
719
        return set([NotBzrDirFormat()])
 
720
 
 
721
    @classmethod
 
722
    def probe_transport(self, transport):
 
723
        """Our format is present if the transport ends in '.not/'."""
 
724
        if transport.has('.not'):
 
725
            return NotBzrDirFormat()
 
726
 
 
727
 
 
728
class TestNotBzrDir(TestCaseWithTransport):
 
729
    """Tests for using the bzrdir api with a non .bzr based disk format.
 
730
    
 
731
    If/when one of these is in the core, we can let the implementation tests
 
732
    verify this works.
 
733
    """
 
734
 
 
735
    def test_create_and_find_format(self):
 
736
        # create a .notbzr dir 
 
737
        format = NotBzrDirFormat()
 
738
        dir = format.initialize(self.get_url())
 
739
        self.assertIsInstance(dir, NotBzrDir)
 
740
        # now probe for it.
 
741
        bzrlib.bzrdir.BzrDirFormat.register_control_format(format)
 
742
        try:
 
743
            found = bzrlib.bzrdir.BzrDirFormat.find_format(
 
744
                get_transport(self.get_url()))
 
745
            self.assertIsInstance(found, NotBzrDirFormat)
 
746
        finally:
 
747
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(format)
 
748
 
 
749
    def test_included_in_known_formats(self):
 
750
        bzrlib.bzrdir.BzrDirFormat.register_control_format(NotBzrDirFormat)
 
751
        try:
 
752
            formats = bzrlib.bzrdir.BzrDirFormat.known_formats()
 
753
            for format in formats:
 
754
                if isinstance(format, NotBzrDirFormat):
 
755
                    return
 
756
            self.fail("No NotBzrDirFormat in %s" % formats)
 
757
        finally:
 
758
            bzrlib.bzrdir.BzrDirFormat.unregister_control_format(NotBzrDirFormat)
 
759
 
 
760
 
 
761
class NonLocalTests(TestCaseWithTransport):
 
762
    """Tests for bzrdir static behaviour on non local paths."""
 
763
 
 
764
    def setUp(self):
 
765
        super(NonLocalTests, self).setUp()
 
766
        self.vfs_transport_factory = MemoryServer
 
767
    
 
768
    def test_create_branch_convenience(self):
 
769
        # outside a repo the default convenience output is a repo+branch_tree
 
770
        format = bzrdir.format_registry.make_bzrdir('knit')
 
771
        branch = bzrdir.BzrDir.create_branch_convenience(
 
772
            self.get_url('foo'), format=format)
 
773
        self.assertRaises(errors.NoWorkingTree,
 
774
                          branch.bzrdir.open_workingtree)
 
775
        branch.bzrdir.open_repository()
 
776
 
 
777
    def test_create_branch_convenience_force_tree_not_local_fails(self):
 
778
        # outside a repo the default convenience output is a repo+branch_tree
 
779
        format = bzrdir.format_registry.make_bzrdir('knit')
 
780
        self.assertRaises(errors.NotLocalUrl,
 
781
            bzrdir.BzrDir.create_branch_convenience,
 
782
            self.get_url('foo'),
 
783
            force_new_tree=True,
 
784
            format=format)
 
785
        t = get_transport(self.get_url('.'))
 
786
        self.assertFalse(t.has('foo'))
 
787
 
 
788
    def test_clone(self):
 
789
        # clone into a nonlocal path works
 
790
        format = bzrdir.format_registry.make_bzrdir('knit')
 
791
        branch = bzrdir.BzrDir.create_branch_convenience('local',
 
792
                                                         format=format)
 
793
        branch.bzrdir.open_workingtree()
 
794
        result = branch.bzrdir.clone(self.get_url('remote'))
 
795
        self.assertRaises(errors.NoWorkingTree,
 
796
                          result.open_workingtree)
 
797
        result.open_branch()
 
798
        result.open_repository()
 
799
 
 
800
    def test_checkout_metadir(self):
 
801
        # checkout_metadir has reasonable working tree format even when no
 
802
        # working tree is present
 
803
        self.make_branch('branch-knit2', format='dirstate-with-subtree')
 
804
        my_bzrdir = bzrdir.BzrDir.open(self.get_url('branch-knit2'))
 
805
        checkout_format = my_bzrdir.checkout_metadir()
 
806
        self.assertIsInstance(checkout_format.workingtree_format,
 
807
                              workingtree.WorkingTreeFormat3)
 
808
 
 
809
 
 
810
class TestHTTPRedirectionLoop(object):
 
811
    """Test redirection loop between two http servers.
 
812
 
 
813
    This MUST be used by daughter classes that also inherit from
 
814
    TestCaseWithTwoWebservers.
 
815
 
 
816
    We can't inherit directly from TestCaseWithTwoWebservers or the
 
817
    test framework will try to create an instance which cannot
 
818
    run, its implementation being incomplete. 
 
819
    """
 
820
 
 
821
    # Should be defined by daughter classes to ensure redirection
 
822
    # still use the same transport implementation (not currently
 
823
    # enforced as it's a bit tricky to get right (see the FIXME
 
824
    # in BzrDir.open_from_transport for the unique use case so
 
825
    # far)
 
826
    _qualifier = None
 
827
 
 
828
    def create_transport_readonly_server(self):
 
829
        return HTTPServerRedirecting()
 
830
 
 
831
    def create_transport_secondary_server(self):
 
832
        return HTTPServerRedirecting()
 
833
 
 
834
    def setUp(self):
 
835
        # Both servers redirect to each server creating a loop
 
836
        super(TestHTTPRedirectionLoop, self).setUp()
 
837
        # The redirections will point to the new server
 
838
        self.new_server = self.get_readonly_server()
 
839
        # The requests to the old server will be redirected
 
840
        self.old_server = self.get_secondary_server()
 
841
        # Configure the redirections
 
842
        self.old_server.redirect_to(self.new_server.host, self.new_server.port)
 
843
        self.new_server.redirect_to(self.old_server.host, self.old_server.port)
 
844
 
 
845
    def _qualified_url(self, host, port):
 
846
        return 'http+%s://%s:%s' % (self._qualifier, host, port)
 
847
 
 
848
    def test_loop(self):
 
849
        # Starting from either server should loop
 
850
        old_url = self._qualified_url(self.old_server.host, 
 
851
                                      self.old_server.port)
 
852
        oldt = self._transport(old_url)
 
853
        self.assertRaises(errors.NotBranchError,
 
854
                          bzrdir.BzrDir.open_from_transport, oldt)
 
855
        new_url = self._qualified_url(self.new_server.host, 
 
856
                                      self.new_server.port)
 
857
        newt = self._transport(new_url)
 
858
        self.assertRaises(errors.NotBranchError,
 
859
                          bzrdir.BzrDir.open_from_transport, newt)
 
860
 
 
861
 
 
862
class TestHTTPRedirections_urllib(TestHTTPRedirectionLoop,
 
863
                                  TestCaseWithTwoWebservers):
 
864
    """Tests redirections for urllib implementation"""
 
865
 
 
866
    _qualifier = 'urllib'
 
867
    _transport = HttpTransport_urllib
 
868
 
 
869
 
 
870
 
 
871
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
 
872
                                  TestHTTPRedirectionLoop,
 
873
                                  TestCaseWithTwoWebservers):
 
874
    """Tests redirections for pycurl implementation"""
 
875
 
 
876
    _qualifier = 'pycurl'
 
877
 
 
878
 
 
879
class TestDotBzrHidden(TestCaseWithTransport):
 
880
 
 
881
    ls = ['ls']
 
882
    if sys.platform == 'win32':
 
883
        ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
 
884
 
 
885
    def get_ls(self):
 
886
        f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
 
887
            stderr=subprocess.PIPE)
 
888
        out, err = f.communicate()
 
889
        self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
 
890
                         % (self.ls, err))
 
891
        return out.splitlines()
 
892
 
 
893
    def test_dot_bzr_hidden(self):
 
894
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
895
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
896
        b = bzrdir.BzrDir.create('.')
 
897
        self.build_tree('a')
 
898
        self.assertEquals(['a'], self.get_ls())
 
899
 
 
900
    def test_dot_bzr_hidden_with_url(self):
 
901
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
902
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
903
        b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
 
904
        self.build_tree('a')
 
905
        self.assertEquals(['a'], self.get_ls())