← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bzrdir.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-04-09 20:23:07 UTC
  • mfrom: (4265.1.4 bbc-merge)
  • Revision ID: pqm@pqm.ubuntu.com-20090409202307-n0depb16qepoe21o
(jam) Change _fetch_uses_deltas = False for CHK repos until we can
        write a better fix.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bzrdir.py
1
 
    # Copyright (C) 2005, 2006, 2007 Canonical Ltd
2
 
# 
 
1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 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
19
For interface contract tests, see tests/bzr_dir_implementations.
20
20
"""
21
21
 
 
22
import os
22
23
import os.path
23
24
from StringIO import StringIO
 
25
import subprocess
 
26
import sys
24
27
 
25
28
from bzrlib import (
26
29
    bzrdir,
27
30
    errors,
28
31
    help_topics,
29
32
    repository,
 
33
    osutils,
30
34
    symbol_versioning,
 
35
    remote,
31
36
    urlutils,
 
37
    win32utils,
32
38
    workingtree,
33
39
    )
34
40
import bzrlib.branch
38
44
                           )
39
45
from bzrlib.tests import (
40
46
    TestCase,
 
47
    TestCaseWithMemoryTransport,
41
48
    TestCaseWithTransport,
 
49
    TestSkipped,
42
50
    test_sftp_transport
43
51
    )
44
 
from bzrlib.tests.HttpServer import HttpServer
45
 
from bzrlib.tests.HTTPTestUtil import (
46
 
    TestCaseWithTwoWebservers,
47
 
    HTTPServerRedirecting,
 
52
from bzrlib.tests import(
 
53
    http_server,
 
54
    http_utils,
48
55
    )
49
56
from bzrlib.tests.test_http import TestWithTransport_pycurl
50
57
from bzrlib.transport import get_transport
51
58
from bzrlib.transport.http._urllib import HttpTransport_urllib
52
59
from bzrlib.transport.memory import MemoryServer
53
 
from bzrlib.repofmt import knitrepo, weaverepo
 
60
from bzrlib.transport.nosmart import NoSmartTransportDecorator
 
61
from bzrlib.transport.readonly import ReadonlyTransportDecorator
 
62
from bzrlib.repofmt import knitrepo, weaverepo, pack_repo
54
63
 
55
64
 
56
65
class TestDefaultFormat(TestCase):
59
68
        old_format = bzrdir.BzrDirFormat.get_default_format()
60
69
        # default is BzrDirFormat6
61
70
        self.failUnless(isinstance(old_format, bzrdir.BzrDirMetaFormat1))
62
 
        self.applyDeprecated(symbol_versioning.zero_fourteen, 
63
 
                             bzrdir.BzrDirFormat.set_default_format, 
64
 
                             SampleBzrDirFormat())
 
71
        bzrdir.BzrDirFormat._set_default_format(SampleBzrDirFormat())
65
72
        # creating a bzr dir should now create an instrumented dir.
66
73
        try:
67
74
            result = bzrdir.BzrDir.create('memory:///')
68
75
            self.failUnless(isinstance(result, SampleBzrDir))
69
76
        finally:
70
 
            self.applyDeprecated(symbol_versioning.zero_fourteen,
71
 
                bzrdir.BzrDirFormat.set_default_format, old_format)
 
77
            bzrdir.BzrDirFormat._set_default_format(old_format)
72
78
        self.assertEqual(old_format, bzrdir.BzrDirFormat.get_default_format())
73
79
 
74
80
 
79
85
        my_format_registry.register('weave', bzrdir.BzrDirFormat6,
80
86
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
81
87
            ' repositories', deprecated=True)
82
 
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir', 
 
88
        my_format_registry.register_lazy('lazy', 'bzrlib.bzrdir',
83
89
            'BzrDirFormat6', 'Format registered lazily', deprecated=True)
84
90
        my_format_registry.register_metadir('knit',
85
91
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit1',
90
96
            'branch6',
91
97
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
92
98
            'Experimental successor to knit.  Use at your own risk.',
93
 
            branch_format='bzrlib.branch.BzrBranchFormat6')
 
99
            branch_format='bzrlib.branch.BzrBranchFormat6',
 
100
            experimental=True)
94
101
        my_format_registry.register_metadir(
95
102
            'hidden format',
96
103
            'bzrlib.repofmt.knitrepo.RepositoryFormatKnit3',
111
118
        my_bzrdir = my_format_registry.make_bzrdir('weave')
112
119
        self.assertIsInstance(my_bzrdir, bzrdir.BzrDirFormat6)
113
120
        my_bzrdir = my_format_registry.make_bzrdir('default')
114
 
        self.assertIsInstance(my_bzrdir.repository_format, 
 
121
        self.assertIsInstance(my_bzrdir.repository_format,
115
122
            knitrepo.RepositoryFormatKnit1)
116
123
        my_bzrdir = my_format_registry.make_bzrdir('knit')
117
 
        self.assertIsInstance(my_bzrdir.repository_format, 
 
124
        self.assertIsInstance(my_bzrdir.repository_format,
118
125
            knitrepo.RepositoryFormatKnit1)
119
126
        my_bzrdir = my_format_registry.make_bzrdir('branch6')
120
127
        self.assertIsInstance(my_bzrdir.get_branch_format(),
124
131
        my_format_registry = self.make_format_registry()
125
132
        self.assertEqual('Format registered lazily',
126
133
                         my_format_registry.get_help('lazy'))
127
 
        self.assertEqual('Format using knits', 
 
134
        self.assertEqual('Format using knits',
128
135
                         my_format_registry.get_help('knit'))
129
 
        self.assertEqual('Format using knits', 
 
136
        self.assertEqual('Format using knits',
130
137
                         my_format_registry.get_help('default'))
131
138
        self.assertEqual('Pre-0.8 format.  Slower and does not support'
132
 
                         ' checkouts or shared repositories', 
 
139
                         ' checkouts or shared repositories',
133
140
                         my_format_registry.get_help('weave'))
134
 
        
 
141
 
135
142
    def test_help_topic(self):
136
143
        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')
 
144
        registry = self.make_format_registry()
 
145
        topics.register('current-formats', registry.help_topic,
 
146
                        'Current formats')
 
147
        topics.register('other-formats', registry.help_topic,
 
148
                        'Other formats')
 
149
        new = topics.get_detail('current-formats')
 
150
        rest = topics.get_detail('other-formats')
 
151
        experimental, deprecated = rest.split('Deprecated formats')
 
152
        self.assertContainsRe(new, 'bzr help formats')
 
153
        self.assertContainsRe(new,
 
154
                ':knit:\n    \(native\) \(default\) Format using knits\n')
 
155
        self.assertContainsRe(experimental,
 
156
                ':branch6:\n    \(native\) Experimental successor to knit')
 
157
        self.assertContainsRe(deprecated,
 
158
                ':lazy:\n    \(native\) Format registered lazily\n')
146
159
        self.assertNotContainsRe(new, 'hidden')
147
160
 
148
161
    def test_set_default_repository(self):
159
172
        finally:
160
173
            bzrdir.format_registry.set_default_repository(old_default)
161
174
 
 
175
    def test_aliases(self):
 
176
        a_registry = bzrdir.BzrDirFormatRegistry()
 
177
        a_registry.register('weave', bzrdir.BzrDirFormat6,
 
178
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
179
            ' repositories', deprecated=True)
 
180
        a_registry.register('weavealias', bzrdir.BzrDirFormat6,
 
181
            'Pre-0.8 format.  Slower and does not support checkouts or shared'
 
182
            ' repositories', deprecated=True, alias=True)
 
183
        self.assertEqual(frozenset(['weavealias']), a_registry.aliases())
 
184
 
162
185
 
163
186
class SampleBranch(bzrlib.branch.Branch):
164
187
    """A dummy branch for guess what, dummy use."""
167
190
        self.bzrdir = dir
168
191
 
169
192
 
 
193
class SampleRepository(bzrlib.repository.Repository):
 
194
    """A dummy repo."""
 
195
 
 
196
    def __init__(self, dir):
 
197
        self.bzrdir = dir
 
198
 
 
199
 
170
200
class SampleBzrDir(bzrdir.BzrDir):
171
201
    """A sample BzrDir implementation to allow testing static methods."""
172
202
 
176
206
 
177
207
    def open_repository(self):
178
208
        """See BzrDir.open_repository."""
179
 
        return "A repository"
 
209
        return SampleRepository(self)
180
210
 
181
211
    def create_branch(self):
182
212
        """See BzrDir.create_branch."""
190
220
class SampleBzrDirFormat(bzrdir.BzrDirFormat):
191
221
    """A sample format
192
222
 
193
 
    this format is initializable, unsupported to aid in testing the 
 
223
    this format is initializable, unsupported to aid in testing the
194
224
    open and open_downlevel routines.
195
225
    """
196
226
 
198
228
        """See BzrDirFormat.get_format_string()."""
199
229
        return "Sample .bzr dir format."
200
230
 
201
 
    def initialize(self, url):
 
231
    def initialize_on_transport(self, t):
202
232
        """Create a bzr dir."""
203
 
        t = get_transport(url)
204
233
        t.mkdir('.bzr')
205
234
        t.put_bytes('.bzr/branch-format', self.get_format_string())
206
235
        return SampleBzrDir(t, self)
218
247
    def test_find_format(self):
219
248
        # is the right format object found for a branch?
220
249
        # create a branch with a few known format objects.
221
 
        # this is not quite the same as 
 
250
        # this is not quite the same as
222
251
        t = get_transport(self.get_url())
223
252
        self.build_tree(["foo/", "bar/"], transport=t)
224
253
        def check_format(format, url):
228
257
            self.failUnless(isinstance(found_format, format.__class__))
229
258
        check_format(bzrdir.BzrDirFormat5(), "foo")
230
259
        check_format(bzrdir.BzrDirFormat6(), "bar")
231
 
        
 
260
 
232
261
    def test_find_format_nothing_there(self):
233
262
        self.assertRaises(NotBranchError,
234
263
                          bzrdir.BzrDirFormat.find_format,
261
290
        # now open_downlevel should fail too.
262
291
        self.assertRaises(UnknownFormatError, bzrdir.BzrDir.open_unsupported, url)
263
292
 
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
293
    def test_create_branch_and_repo_uses_default(self):
290
294
        format = SampleBzrDirFormat()
291
 
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(), 
 
295
        branch = bzrdir.BzrDir.create_branch_and_repo(self.get_url(),
292
296
                                                      format=format)
293
297
        self.assertTrue(isinstance(branch, SampleBranch))
294
298
 
303
307
                          branch.bzrdir.open_repository)
304
308
 
305
309
    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 
 
310
        # creating a branch and repo in a shared repo can be forced to
307
311
        # make a new repo
308
312
        format = bzrdir.format_registry.make_bzrdir('knit')
309
313
        self.make_repository('.', shared=True, format=format)
314
318
 
315
319
    def test_create_standalone_working_tree(self):
316
320
        format = SampleBzrDirFormat()
317
 
        # note this is deliberately readonly, as this failure should 
 
321
        # note this is deliberately readonly, as this failure should
318
322
        # occur before any writes.
319
323
        self.assertRaises(errors.NotLocalUrl,
320
324
                          bzrdir.BzrDir.create_standalone_workingtree,
321
325
                          self.get_readonly_url(), format=format)
322
 
        tree = bzrdir.BzrDir.create_standalone_workingtree('.', 
 
326
        tree = bzrdir.BzrDir.create_standalone_workingtree('.',
323
327
                                                           format=format)
324
328
        self.assertEqual('A tree', tree)
325
329
 
327
331
        # create standalone working tree always makes a repo.
328
332
        format = bzrdir.format_registry.make_bzrdir('knit')
329
333
        self.make_repository('.', shared=True, format=format)
330
 
        # note this is deliberately readonly, as this failure should 
 
334
        # note this is deliberately readonly, as this failure should
331
335
        # occur before any writes.
332
336
        self.assertRaises(errors.NotLocalUrl,
333
337
                          bzrdir.BzrDir.create_standalone_workingtree,
334
338
                          self.get_readonly_url('child'), format=format)
335
 
        tree = bzrdir.BzrDir.create_standalone_workingtree('child', 
 
339
        tree = bzrdir.BzrDir.create_standalone_workingtree('child',
336
340
            format=format)
337
341
        tree.bzrdir.open_repository()
338
342
 
343
347
        branch.bzrdir.open_workingtree()
344
348
        branch.bzrdir.open_repository()
345
349
 
 
350
    def test_create_branch_convenience_possible_transports(self):
 
351
        """Check that the optional 'possible_transports' is recognized"""
 
352
        format = bzrdir.format_registry.make_bzrdir('knit')
 
353
        t = self.get_transport()
 
354
        branch = bzrdir.BzrDir.create_branch_convenience(
 
355
            '.', format=format, possible_transports=[t])
 
356
        branch.bzrdir.open_workingtree()
 
357
        branch.bzrdir.open_repository()
 
358
 
346
359
    def test_create_branch_convenience_root(self):
347
360
        """Creating a branch at the root of a fs should work."""
348
361
        self.vfs_transport_factory = MemoryServer
349
362
        # outside a repo the default convenience output is a repo+branch_tree
350
363
        format = bzrdir.format_registry.make_bzrdir('knit')
351
 
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(), 
 
364
        branch = bzrdir.BzrDir.create_branch_convenience(self.get_url(),
352
365
                                                         format=format)
353
366
        self.assertRaises(errors.NoWorkingTree,
354
367
                          branch.bzrdir.open_workingtree)
364
377
        branch.bzrdir.open_workingtree()
365
378
        self.assertRaises(errors.NoRepositoryPresent,
366
379
                          branch.bzrdir.open_repository)
367
 
            
 
380
 
368
381
    def test_create_branch_convenience_under_shared_repo_force_no_tree(self):
369
382
        # inside a repo the default convenience output is a branch+ follow the
370
383
        # repo tree policy but we can override that
376
389
                          branch.bzrdir.open_workingtree)
377
390
        self.assertRaises(errors.NoRepositoryPresent,
378
391
                          branch.bzrdir.open_repository)
379
 
            
 
392
 
380
393
    def test_create_branch_convenience_under_shared_repo_no_tree_policy(self):
381
394
        # inside a repo the default convenience output is a branch+ follow the
382
395
        # repo tree policy
383
396
        format = bzrdir.format_registry.make_bzrdir('knit')
384
397
        repo = self.make_repository('.', shared=True, format=format)
385
398
        repo.set_make_working_trees(False)
386
 
        branch = bzrdir.BzrDir.create_branch_convenience('child', 
 
399
        branch = bzrdir.BzrDir.create_branch_convenience('child',
387
400
                                                         format=format)
388
401
        self.assertRaises(errors.NoWorkingTree,
389
402
                          branch.bzrdir.open_workingtree)
413
426
        branch.bzrdir.open_workingtree()
414
427
 
415
428
 
 
429
class TestRepositoryAcquisitionPolicy(TestCaseWithTransport):
 
430
 
 
431
    def test_acquire_repository_standalone(self):
 
432
        """The default acquisition policy should create a standalone branch."""
 
433
        my_bzrdir = self.make_bzrdir('.')
 
434
        repo_policy = my_bzrdir.determine_repository_policy()
 
435
        repo, is_new = repo_policy.acquire_repository()
 
436
        self.assertEqual(repo.bzrdir.root_transport.base,
 
437
                         my_bzrdir.root_transport.base)
 
438
        self.assertFalse(repo.is_shared())
 
439
 
 
440
    def test_determine_stacking_policy(self):
 
441
        parent_bzrdir = self.make_bzrdir('.')
 
442
        child_bzrdir = self.make_bzrdir('child')
 
443
        parent_bzrdir.get_config().set_default_stack_on('http://example.org')
 
444
        repo_policy = child_bzrdir.determine_repository_policy()
 
445
        self.assertEqual('http://example.org', repo_policy._stack_on)
 
446
 
 
447
    def test_determine_stacking_policy_relative(self):
 
448
        parent_bzrdir = self.make_bzrdir('.')
 
449
        child_bzrdir = self.make_bzrdir('child')
 
450
        parent_bzrdir.get_config().set_default_stack_on('child2')
 
451
        repo_policy = child_bzrdir.determine_repository_policy()
 
452
        self.assertEqual('child2', repo_policy._stack_on)
 
453
        self.assertEqual(parent_bzrdir.root_transport.base,
 
454
                         repo_policy._stack_on_pwd)
 
455
 
 
456
    def prepare_default_stacking(self, child_format='1.6'):
 
457
        parent_bzrdir = self.make_bzrdir('.')
 
458
        child_branch = self.make_branch('child', format=child_format)
 
459
        parent_bzrdir.get_config().set_default_stack_on(child_branch.base)
 
460
        new_child_transport = parent_bzrdir.transport.clone('child2')
 
461
        return child_branch, new_child_transport
 
462
 
 
463
    def test_clone_on_transport_obeys_stacking_policy(self):
 
464
        child_branch, new_child_transport = self.prepare_default_stacking()
 
465
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
 
466
        self.assertEqual(child_branch.base,
 
467
                         new_child.open_branch().get_stacked_on_url())
 
468
 
 
469
    def test_default_stacking_with_stackable_branch_unstackable_repo(self):
 
470
        # Make stackable source branch with an unstackable repo format.
 
471
        source_bzrdir = self.make_bzrdir('source')
 
472
        pack_repo.RepositoryFormatKnitPack1().initialize(source_bzrdir)
 
473
        source_branch = bzrlib.branch.BzrBranchFormat7().initialize(source_bzrdir)
 
474
        # Make a directory with a default stacking policy
 
475
        parent_bzrdir = self.make_bzrdir('parent')
 
476
        stacked_on = self.make_branch('parent/stacked-on', format='pack-0.92')
 
477
        parent_bzrdir.get_config().set_default_stack_on(stacked_on.base)
 
478
        # Clone source into directory
 
479
        target = source_bzrdir.clone(self.get_url('parent/target'))
 
480
 
 
481
    def test_sprout_obeys_stacking_policy(self):
 
482
        child_branch, new_child_transport = self.prepare_default_stacking()
 
483
        new_child = child_branch.bzrdir.sprout(new_child_transport.base)
 
484
        self.assertEqual(child_branch.base,
 
485
                         new_child.open_branch().get_stacked_on_url())
 
486
 
 
487
    def test_clone_ignores_policy_for_unsupported_formats(self):
 
488
        child_branch, new_child_transport = self.prepare_default_stacking(
 
489
            child_format='pack-0.92')
 
490
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport)
 
491
        self.assertRaises(errors.UnstackableBranchFormat,
 
492
                          new_child.open_branch().get_stacked_on_url)
 
493
 
 
494
    def test_sprout_ignores_policy_for_unsupported_formats(self):
 
495
        child_branch, new_child_transport = self.prepare_default_stacking(
 
496
            child_format='pack-0.92')
 
497
        new_child = child_branch.bzrdir.sprout(new_child_transport.base)
 
498
        self.assertRaises(errors.UnstackableBranchFormat,
 
499
                          new_child.open_branch().get_stacked_on_url)
 
500
 
 
501
    def test_sprout_upgrades_format_if_stacked_specified(self):
 
502
        child_branch, new_child_transport = self.prepare_default_stacking(
 
503
            child_format='pack-0.92')
 
504
        new_child = child_branch.bzrdir.sprout(new_child_transport.base,
 
505
                                               stacked=True)
 
506
        self.assertEqual(child_branch.bzrdir.root_transport.base,
 
507
                         new_child.open_branch().get_stacked_on_url())
 
508
        repo = new_child.open_repository()
 
509
        self.assertTrue(repo._format.supports_external_lookups)
 
510
        self.assertFalse(repo.supports_rich_root())
 
511
 
 
512
    def test_clone_on_transport_upgrades_format_if_stacked_on_specified(self):
 
513
        child_branch, new_child_transport = self.prepare_default_stacking(
 
514
            child_format='pack-0.92')
 
515
        new_child = child_branch.bzrdir.clone_on_transport(new_child_transport,
 
516
            stacked_on=child_branch.bzrdir.root_transport.base)
 
517
        self.assertEqual(child_branch.bzrdir.root_transport.base,
 
518
                         new_child.open_branch().get_stacked_on_url())
 
519
        repo = new_child.open_repository()
 
520
        self.assertTrue(repo._format.supports_external_lookups)
 
521
        self.assertFalse(repo.supports_rich_root())
 
522
 
 
523
    def test_sprout_upgrades_to_rich_root_format_if_needed(self):
 
524
        child_branch, new_child_transport = self.prepare_default_stacking(
 
525
            child_format='rich-root-pack')
 
526
        new_child = child_branch.bzrdir.sprout(new_child_transport.base,
 
527
                                               stacked=True)
 
528
        repo = new_child.open_repository()
 
529
        self.assertTrue(repo._format.supports_external_lookups)
 
530
        self.assertTrue(repo.supports_rich_root())
 
531
 
 
532
    def test_add_fallback_repo_handles_absolute_urls(self):
 
533
        stack_on = self.make_branch('stack_on', format='1.6')
 
534
        repo = self.make_repository('repo', format='1.6')
 
535
        policy = bzrdir.UseExistingRepository(repo, stack_on.base)
 
536
        policy._add_fallback(repo)
 
537
 
 
538
    def test_add_fallback_repo_handles_relative_urls(self):
 
539
        stack_on = self.make_branch('stack_on', format='1.6')
 
540
        repo = self.make_repository('repo', format='1.6')
 
541
        policy = bzrdir.UseExistingRepository(repo, '.', stack_on.base)
 
542
        policy._add_fallback(repo)
 
543
 
 
544
    def test_configure_relative_branch_stacking_url(self):
 
545
        stack_on = self.make_branch('stack_on', format='1.6')
 
546
        stacked = self.make_branch('stack_on/stacked', format='1.6')
 
547
        policy = bzrdir.UseExistingRepository(stacked.repository,
 
548
            '.', stack_on.base)
 
549
        policy.configure_branch(stacked)
 
550
        self.assertEqual('..', stacked.get_stacked_on_url())
 
551
 
 
552
    def test_relative_branch_stacking_to_absolute(self):
 
553
        stack_on = self.make_branch('stack_on', format='1.6')
 
554
        stacked = self.make_branch('stack_on/stacked', format='1.6')
 
555
        policy = bzrdir.UseExistingRepository(stacked.repository,
 
556
            '.', self.get_readonly_url('stack_on'))
 
557
        policy.configure_branch(stacked)
 
558
        self.assertEqual(self.get_readonly_url('stack_on'),
 
559
                         stacked.get_stacked_on_url())
 
560
 
 
561
 
416
562
class ChrootedTests(TestCaseWithTransport):
417
563
    """A support class that provides readonly urls outside the local namespace.
418
564
 
424
570
    def setUp(self):
425
571
        super(ChrootedTests, self).setUp()
426
572
        if not self.vfs_transport_factory == MemoryServer:
427
 
            self.transport_readonly_server = HttpServer
 
573
            self.transport_readonly_server = http_server.HttpServer
 
574
 
 
575
    def local_branch_path(self, branch):
 
576
         return os.path.realpath(urlutils.local_path_from_url(branch.base))
428
577
 
429
578
    def test_open_containing(self):
430
579
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing,
437
586
        branch, relpath = bzrdir.BzrDir.open_containing(self.get_readonly_url('g/p/q'))
438
587
        self.assertEqual('g/p/q', relpath)
439
588
 
 
589
    def test_open_containing_tree_branch_or_repository_empty(self):
 
590
        self.assertRaises(errors.NotBranchError,
 
591
            bzrdir.BzrDir.open_containing_tree_branch_or_repository,
 
592
            self.get_readonly_url(''))
 
593
 
 
594
    def test_open_containing_tree_branch_or_repository_all(self):
 
595
        self.make_branch_and_tree('topdir')
 
596
        tree, branch, repo, relpath = \
 
597
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
598
                'topdir/foo')
 
599
        self.assertEqual(os.path.realpath('topdir'),
 
600
                         os.path.realpath(tree.basedir))
 
601
        self.assertEqual(os.path.realpath('topdir'),
 
602
                         self.local_branch_path(branch))
 
603
        self.assertEqual(
 
604
            osutils.realpath(os.path.join('topdir', '.bzr', 'repository')),
 
605
            repo.bzrdir.transport.local_abspath('repository'))
 
606
        self.assertEqual(relpath, 'foo')
 
607
 
 
608
    def test_open_containing_tree_branch_or_repository_no_tree(self):
 
609
        self.make_branch('branch')
 
610
        tree, branch, repo, relpath = \
 
611
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
612
                'branch/foo')
 
613
        self.assertEqual(tree, None)
 
614
        self.assertEqual(os.path.realpath('branch'),
 
615
                         self.local_branch_path(branch))
 
616
        self.assertEqual(
 
617
            osutils.realpath(os.path.join('branch', '.bzr', 'repository')),
 
618
            repo.bzrdir.transport.local_abspath('repository'))
 
619
        self.assertEqual(relpath, 'foo')
 
620
 
 
621
    def test_open_containing_tree_branch_or_repository_repo(self):
 
622
        self.make_repository('repo')
 
623
        tree, branch, repo, relpath = \
 
624
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
625
                'repo')
 
626
        self.assertEqual(tree, None)
 
627
        self.assertEqual(branch, None)
 
628
        self.assertEqual(
 
629
            osutils.realpath(os.path.join('repo', '.bzr', 'repository')),
 
630
            repo.bzrdir.transport.local_abspath('repository'))
 
631
        self.assertEqual(relpath, '')
 
632
 
 
633
    def test_open_containing_tree_branch_or_repository_shared_repo(self):
 
634
        self.make_repository('shared', shared=True)
 
635
        bzrdir.BzrDir.create_branch_convenience('shared/branch',
 
636
                                                force_new_tree=False)
 
637
        tree, branch, repo, relpath = \
 
638
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
639
                'shared/branch')
 
640
        self.assertEqual(tree, None)
 
641
        self.assertEqual(os.path.realpath('shared/branch'),
 
642
                         self.local_branch_path(branch))
 
643
        self.assertEqual(
 
644
            osutils.realpath(os.path.join('shared', '.bzr', 'repository')),
 
645
            repo.bzrdir.transport.local_abspath('repository'))
 
646
        self.assertEqual(relpath, '')
 
647
 
 
648
    def test_open_containing_tree_branch_or_repository_branch_subdir(self):
 
649
        self.make_branch_and_tree('foo')
 
650
        self.build_tree(['foo/bar/'])
 
651
        tree, branch, repo, relpath = \
 
652
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
653
                'foo/bar')
 
654
        self.assertEqual(os.path.realpath('foo'),
 
655
                         os.path.realpath(tree.basedir))
 
656
        self.assertEqual(os.path.realpath('foo'),
 
657
                         self.local_branch_path(branch))
 
658
        self.assertEqual(
 
659
            osutils.realpath(os.path.join('foo', '.bzr', 'repository')),
 
660
            repo.bzrdir.transport.local_abspath('repository'))
 
661
        self.assertEqual(relpath, 'bar')
 
662
 
 
663
    def test_open_containing_tree_branch_or_repository_repo_subdir(self):
 
664
        self.make_repository('bar')
 
665
        self.build_tree(['bar/baz/'])
 
666
        tree, branch, repo, relpath = \
 
667
            bzrdir.BzrDir.open_containing_tree_branch_or_repository(
 
668
                'bar/baz')
 
669
        self.assertEqual(tree, None)
 
670
        self.assertEqual(branch, None)
 
671
        self.assertEqual(
 
672
            osutils.realpath(os.path.join('bar', '.bzr', 'repository')),
 
673
            repo.bzrdir.transport.local_abspath('repository'))
 
674
        self.assertEqual(relpath, 'baz')
 
675
 
440
676
    def test_open_containing_from_transport(self):
441
677
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_containing_from_transport,
442
678
                          get_transport(self.get_readonly_url('')))
451
687
        self.assertEqual('g/p/q', relpath)
452
688
 
453
689
    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
690
        self.make_branch_and_tree('topdir')
459
691
        tree, branch, relpath = bzrdir.BzrDir.open_containing_tree_or_branch(
460
692
            'topdir/foo')
461
693
        self.assertEqual(os.path.realpath('topdir'),
462
694
                         os.path.realpath(tree.basedir))
463
695
        self.assertEqual(os.path.realpath('topdir'),
464
 
                         local_branch_path(branch))
 
696
                         self.local_branch_path(branch))
465
697
        self.assertIs(tree.bzrdir, branch.bzrdir)
466
698
        self.assertEqual('foo', relpath)
467
699
        # opening from non-local should not return the tree
475
707
            'topdir/foo')
476
708
        self.assertIs(tree, None)
477
709
        self.assertEqual(os.path.realpath('topdir/foo'),
478
 
                         local_branch_path(branch))
 
710
                         self.local_branch_path(branch))
479
711
        self.assertEqual('', relpath)
480
712
 
 
713
    def test_open_tree_or_branch(self):
 
714
        self.make_branch_and_tree('topdir')
 
715
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir')
 
716
        self.assertEqual(os.path.realpath('topdir'),
 
717
                         os.path.realpath(tree.basedir))
 
718
        self.assertEqual(os.path.realpath('topdir'),
 
719
                         self.local_branch_path(branch))
 
720
        self.assertIs(tree.bzrdir, branch.bzrdir)
 
721
        # opening from non-local should not return the tree
 
722
        tree, branch = bzrdir.BzrDir.open_tree_or_branch(
 
723
            self.get_readonly_url('topdir'))
 
724
        self.assertEqual(None, tree)
 
725
        # without a tree:
 
726
        self.make_branch('topdir/foo')
 
727
        tree, branch = bzrdir.BzrDir.open_tree_or_branch('topdir/foo')
 
728
        self.assertIs(tree, None)
 
729
        self.assertEqual(os.path.realpath('topdir/foo'),
 
730
                         self.local_branch_path(branch))
 
731
 
481
732
    def test_open_from_transport(self):
482
733
        # transport pointing at bzrdir should give a bzrdir with root transport
483
734
        # set to the given transport
486
737
        opened_bzrdir = bzrdir.BzrDir.open_from_transport(transport)
487
738
        self.assertEqual(transport.base, opened_bzrdir.root_transport.base)
488
739
        self.assertIsInstance(opened_bzrdir, bzrdir.BzrDir)
489
 
        
 
740
 
490
741
    def test_open_from_transport_no_bzrdir(self):
491
742
        transport = get_transport(self.get_url())
492
743
        self.assertRaises(NotBranchError, bzrdir.BzrDir.open_from_transport,
501
752
                          transport)
502
753
 
503
754
    def test_sprout_recursive(self):
504
 
        tree = self.make_branch_and_tree('tree1', format='dirstate-with-subtree')
 
755
        tree = self.make_branch_and_tree('tree1',
 
756
                                         format='dirstate-with-subtree')
505
757
        sub_tree = self.make_branch_and_tree('tree1/subtree',
506
758
            format='dirstate-with-subtree')
 
759
        sub_tree.set_root_id('subtree-root')
507
760
        tree.add_reference(sub_tree)
508
761
        self.build_tree(['tree1/subtree/file'])
509
762
        sub_tree.add('file')
510
763
        tree.commit('Initial commit')
511
 
        tree.bzrdir.sprout('tree2')
 
764
        tree2 = tree.bzrdir.sprout('tree2').open_workingtree()
 
765
        tree2.lock_read()
 
766
        self.addCleanup(tree2.unlock)
512
767
        self.failUnlessExists('tree2/subtree/file')
 
768
        self.assertEqual('tree-reference', tree2.kind('subtree-root'))
513
769
 
514
770
    def test_cloning_metadir(self):
515
771
        """Ensure that cloning metadir is suitable"""
537
793
        self.failUnlessExists('repo/tree2/subtree')
538
794
        self.failIfExists('repo/tree2/subtree/file')
539
795
 
 
796
    def make_foo_bar_baz(self):
 
797
        foo = bzrdir.BzrDir.create_branch_convenience('foo').bzrdir
 
798
        bar = self.make_branch('foo/bar').bzrdir
 
799
        baz = self.make_branch('baz').bzrdir
 
800
        return foo, bar, baz
 
801
 
 
802
    def test_find_bzrdirs(self):
 
803
        foo, bar, baz = self.make_foo_bar_baz()
 
804
        transport = get_transport(self.get_url())
 
805
        self.assertEqualBzrdirs([baz, foo, bar],
 
806
                                bzrdir.BzrDir.find_bzrdirs(transport))
 
807
 
 
808
    def test_find_bzrdirs_list_current(self):
 
809
        def list_current(transport):
 
810
            return [s for s in transport.list_dir('') if s != 'baz']
 
811
 
 
812
        foo, bar, baz = self.make_foo_bar_baz()
 
813
        transport = get_transport(self.get_url())
 
814
        self.assertEqualBzrdirs([foo, bar],
 
815
                                bzrdir.BzrDir.find_bzrdirs(transport,
 
816
                                    list_current=list_current))
 
817
 
 
818
 
 
819
    def test_find_bzrdirs_evaluate(self):
 
820
        def evaluate(bzrdir):
 
821
            try:
 
822
                repo = bzrdir.open_repository()
 
823
            except NoRepositoryPresent:
 
824
                return True, bzrdir.root_transport.base
 
825
            else:
 
826
                return False, bzrdir.root_transport.base
 
827
 
 
828
        foo, bar, baz = self.make_foo_bar_baz()
 
829
        transport = get_transport(self.get_url())
 
830
        self.assertEqual([baz.root_transport.base, foo.root_transport.base],
 
831
                         list(bzrdir.BzrDir.find_bzrdirs(transport,
 
832
                                                         evaluate=evaluate)))
 
833
 
 
834
    def assertEqualBzrdirs(self, first, second):
 
835
        first = list(first)
 
836
        second = list(second)
 
837
        self.assertEqual(len(first), len(second))
 
838
        for x, y in zip(first, second):
 
839
            self.assertEqual(x.root_transport.base, y.root_transport.base)
 
840
 
 
841
    def test_find_branches(self):
 
842
        root = self.make_repository('', shared=True)
 
843
        foo, bar, baz = self.make_foo_bar_baz()
 
844
        qux = self.make_bzrdir('foo/qux')
 
845
        transport = get_transport(self.get_url())
 
846
        branches = bzrdir.BzrDir.find_branches(transport)
 
847
        self.assertEqual(baz.root_transport.base, branches[0].base)
 
848
        self.assertEqual(foo.root_transport.base, branches[1].base)
 
849
        self.assertEqual(bar.root_transport.base, branches[2].base)
 
850
 
 
851
        # ensure this works without a top-level repo
 
852
        branches = bzrdir.BzrDir.find_branches(transport.clone('foo'))
 
853
        self.assertEqual(foo.root_transport.base, branches[0].base)
 
854
        self.assertEqual(bar.root_transport.base, branches[1].base)
 
855
 
540
856
 
541
857
class TestMeta1DirFormat(TestCaseWithTransport):
542
858
    """Tests specific to the meta1 dir format."""
581
897
 
582
898
    def test_needs_conversion_different_working_tree(self):
583
899
        # 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)
 
900
        new_format = bzrdir.format_registry.make_bzrdir('dirstate')
 
901
        tree = self.make_branch_and_tree('tree', format='knit')
 
902
        self.assertTrue(tree.bzrdir.needs_format_conversion(
 
903
            new_format))
 
904
 
 
905
    def test_initialize_on_format_uses_smart_transport(self):
 
906
        self.setup_smart_server_with_call_log()
 
907
        new_format = bzrdir.format_registry.make_bzrdir('dirstate')
 
908
        transport = self.get_transport('target')
 
909
        transport.ensure_base()
 
910
        self.reset_smart_call_log()
 
911
        instance = new_format.initialize_on_transport(transport)
 
912
        self.assertIsInstance(instance, remote.RemoteBzrDir)
 
913
        rpc_count = len(self.hpss_calls)
 
914
        # This figure represent the amount of work to perform this use case. It
 
915
        # is entirely ok to reduce this number if a test fails due to rpc_count
 
916
        # being too low. If rpc_count increases, more network roundtrips have
 
917
        # become necessary for this use case. Please do not adjust this number
 
918
        # upwards without agreement from bzr's network support maintainers.
 
919
        self.assertEqual(2, rpc_count)
593
920
 
594
921
 
595
922
class TestFormat5(TestCaseWithTransport):
596
923
    """Tests specific to the version 5 bzrdir format."""
597
924
 
598
925
    def test_same_lockfiles_between_tree_repo_branch(self):
599
 
        # this checks that only a single lockfiles instance is created 
 
926
        # this checks that only a single lockfiles instance is created
600
927
        # for format 5 objects
601
928
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
602
929
        def check_dir_components_use_same_lock(dir):
609
936
        # and if we open it normally.
610
937
        dir = bzrdir.BzrDir.open(self.get_url())
611
938
        check_dir_components_use_same_lock(dir)
612
 
    
 
939
 
613
940
    def test_can_convert(self):
614
941
        # format 5 dirs are convertable
615
942
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
616
943
        self.assertTrue(dir.can_convert_format())
617
 
    
 
944
 
618
945
    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())
 
946
        # format 5 dirs need a conversion if they are not the default,
 
947
        # and they aren't
 
948
        dir = bzrdir.BzrDirFormat5().initialize(self.get_url())
 
949
        # don't need to convert it to itself
 
950
        self.assertFalse(dir.needs_format_conversion(bzrdir.BzrDirFormat5()))
 
951
        # do need to convert it to the current default
 
952
        self.assertTrue(dir.needs_format_conversion(
 
953
            bzrdir.BzrDirFormat.get_default_format()))
629
954
 
630
955
 
631
956
class TestFormat6(TestCaseWithTransport):
632
957
    """Tests specific to the version 6 bzrdir format."""
633
958
 
634
959
    def test_same_lockfiles_between_tree_repo_branch(self):
635
 
        # this checks that only a single lockfiles instance is created 
 
960
        # this checks that only a single lockfiles instance is created
636
961
        # for format 6 objects
637
962
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
638
963
        def check_dir_components_use_same_lock(dir):
645
970
        # and if we open it normally.
646
971
        dir = bzrdir.BzrDir.open(self.get_url())
647
972
        check_dir_components_use_same_lock(dir)
648
 
    
 
973
 
649
974
    def test_can_convert(self):
650
975
        # format 6 dirs are convertable
651
976
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
652
977
        self.assertTrue(dir.can_convert_format())
653
 
    
 
978
 
654
979
    def test_needs_conversion(self):
655
980
        # 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)
 
981
        dir = bzrdir.BzrDirFormat6().initialize(self.get_url())
 
982
        self.assertTrue(dir.needs_format_conversion(
 
983
            bzrdir.BzrDirFormat.get_default_format()))
663
984
 
664
985
 
665
986
class NotBzrDir(bzrlib.bzrdir.BzrDir):
696
1017
 
697
1018
class TestNotBzrDir(TestCaseWithTransport):
698
1019
    """Tests for using the bzrdir api with a non .bzr based disk format.
699
 
    
 
1020
 
700
1021
    If/when one of these is in the core, we can let the implementation tests
701
1022
    verify this works.
702
1023
    """
703
1024
 
704
1025
    def test_create_and_find_format(self):
705
 
        # create a .notbzr dir 
 
1026
        # create a .notbzr dir
706
1027
        format = NotBzrDirFormat()
707
1028
        dir = format.initialize(self.get_url())
708
1029
        self.assertIsInstance(dir, NotBzrDir)
733
1054
    def setUp(self):
734
1055
        super(NonLocalTests, self).setUp()
735
1056
        self.vfs_transport_factory = MemoryServer
736
 
    
 
1057
 
737
1058
    def test_create_branch_convenience(self):
738
1059
        # outside a repo the default convenience output is a repo+branch_tree
739
1060
        format = bzrdir.format_registry.make_bzrdir('knit')
776
1097
                              workingtree.WorkingTreeFormat3)
777
1098
 
778
1099
 
779
 
class TestHTTPRedirectionLoop(object):
780
 
    """Test redirection loop between two http servers.
 
1100
class TestHTTPRedirections(object):
 
1101
    """Test redirection between two http servers.
781
1102
 
782
1103
    This MUST be used by daughter classes that also inherit from
783
1104
    TestCaseWithTwoWebservers.
784
1105
 
785
1106
    We can't inherit directly from TestCaseWithTwoWebservers or the
786
1107
    test framework will try to create an instance which cannot
787
 
    run, its implementation being incomplete. 
 
1108
    run, its implementation being incomplete.
788
1109
    """
789
1110
 
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
1111
    def create_transport_readonly_server(self):
798
 
        return HTTPServerRedirecting()
 
1112
        return http_utils.HTTPServerRedirecting()
799
1113
 
800
1114
    def create_transport_secondary_server(self):
801
 
        return HTTPServerRedirecting()
 
1115
        return http_utils.HTTPServerRedirecting()
802
1116
 
803
1117
    def setUp(self):
804
 
        # Both servers redirect to each server creating a loop
805
 
        super(TestHTTPRedirectionLoop, self).setUp()
 
1118
        super(TestHTTPRedirections, self).setUp()
806
1119
        # The redirections will point to the new server
807
1120
        self.new_server = self.get_readonly_server()
808
1121
        # The requests to the old server will be redirected
809
1122
        self.old_server = self.get_secondary_server()
810
1123
        # Configure the redirections
811
1124
        self.old_server.redirect_to(self.new_server.host, self.new_server.port)
 
1125
 
 
1126
    def test_loop(self):
 
1127
        # Both servers redirect to each other creating a loop
812
1128
        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
1129
        # Starting from either server should loop
819
 
        old_url = self._qualified_url(self.old_server.host, 
 
1130
        old_url = self._qualified_url(self.old_server.host,
820
1131
                                      self.old_server.port)
821
1132
        oldt = self._transport(old_url)
822
1133
        self.assertRaises(errors.NotBranchError,
823
1134
                          bzrdir.BzrDir.open_from_transport, oldt)
824
 
        new_url = self._qualified_url(self.new_server.host, 
 
1135
        new_url = self._qualified_url(self.new_server.host,
825
1136
                                      self.new_server.port)
826
1137
        newt = self._transport(new_url)
827
1138
        self.assertRaises(errors.NotBranchError,
828
1139
                          bzrdir.BzrDir.open_from_transport, newt)
829
1140
 
830
 
 
831
 
class TestHTTPRedirections_urllib(TestHTTPRedirectionLoop,
832
 
                                  TestCaseWithTwoWebservers):
 
1141
    def test_qualifier_preserved(self):
 
1142
        wt = self.make_branch_and_tree('branch')
 
1143
        old_url = self._qualified_url(self.old_server.host,
 
1144
                                      self.old_server.port)
 
1145
        start = self._transport(old_url).clone('branch')
 
1146
        bdir = bzrdir.BzrDir.open_from_transport(start)
 
1147
        # Redirection should preserve the qualifier, hence the transport class
 
1148
        # itself.
 
1149
        self.assertIsInstance(bdir.root_transport, type(start))
 
1150
 
 
1151
 
 
1152
class TestHTTPRedirections_urllib(TestHTTPRedirections,
 
1153
                                  http_utils.TestCaseWithTwoWebservers):
833
1154
    """Tests redirections for urllib implementation"""
834
1155
 
835
 
    _qualifier = 'urllib'
836
1156
    _transport = HttpTransport_urllib
837
1157
 
 
1158
    def _qualified_url(self, host, port):
 
1159
        return 'http+urllib://%s:%s' % (host, port)
 
1160
 
838
1161
 
839
1162
 
840
1163
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
841
 
                                  TestHTTPRedirectionLoop,
842
 
                                  TestCaseWithTwoWebservers):
 
1164
                                  TestHTTPRedirections,
 
1165
                                  http_utils.TestCaseWithTwoWebservers):
843
1166
    """Tests redirections for pycurl implementation"""
844
1167
 
845
 
    _qualifier = 'pycurl'
 
1168
    def _qualified_url(self, host, port):
 
1169
        return 'http+pycurl://%s:%s' % (host, port)
 
1170
 
 
1171
 
 
1172
class TestHTTPRedirections_nosmart(TestHTTPRedirections,
 
1173
                                  http_utils.TestCaseWithTwoWebservers):
 
1174
    """Tests redirections for the nosmart decorator"""
 
1175
 
 
1176
    _transport = NoSmartTransportDecorator
 
1177
 
 
1178
    def _qualified_url(self, host, port):
 
1179
        return 'nosmart+http://%s:%s' % (host, port)
 
1180
 
 
1181
 
 
1182
class TestHTTPRedirections_readonly(TestHTTPRedirections,
 
1183
                                    http_utils.TestCaseWithTwoWebservers):
 
1184
    """Tests redirections for readonly decoratror"""
 
1185
 
 
1186
    _transport = ReadonlyTransportDecorator
 
1187
 
 
1188
    def _qualified_url(self, host, port):
 
1189
        return 'readonly+http://%s:%s' % (host, port)
 
1190
 
 
1191
 
 
1192
class TestDotBzrHidden(TestCaseWithTransport):
 
1193
 
 
1194
    ls = ['ls']
 
1195
    if sys.platform == 'win32':
 
1196
        ls = [os.environ['COMSPEC'], '/C', 'dir', '/B']
 
1197
 
 
1198
    def get_ls(self):
 
1199
        f = subprocess.Popen(self.ls, stdout=subprocess.PIPE,
 
1200
            stderr=subprocess.PIPE)
 
1201
        out, err = f.communicate()
 
1202
        self.assertEqual(0, f.returncode, 'Calling %s failed: %s'
 
1203
                         % (self.ls, err))
 
1204
        return out.splitlines()
 
1205
 
 
1206
    def test_dot_bzr_hidden(self):
 
1207
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
1208
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
1209
        b = bzrdir.BzrDir.create('.')
 
1210
        self.build_tree(['a'])
 
1211
        self.assertEquals(['a'], self.get_ls())
 
1212
 
 
1213
    def test_dot_bzr_hidden_with_url(self):
 
1214
        if sys.platform == 'win32' and not win32utils.has_win32file:
 
1215
            raise TestSkipped('unable to make file hidden without pywin32 library')
 
1216
        b = bzrdir.BzrDir.create(urlutils.local_path_to_url('.'))
 
1217
        self.build_tree(['a'])
 
1218
        self.assertEquals(['a'], self.get_ls())
 
1219
 
 
1220
 
 
1221
class _TestBzrDirFormat(bzrdir.BzrDirMetaFormat1):
 
1222
    """Test BzrDirFormat implementation for TestBzrDirSprout."""
 
1223
 
 
1224
    def _open(self, transport):
 
1225
        return _TestBzrDir(transport, self)
 
1226
 
 
1227
 
 
1228
class _TestBzrDir(bzrdir.BzrDirMeta1):
 
1229
    """Test BzrDir implementation for TestBzrDirSprout.
 
1230
 
 
1231
    When created a _TestBzrDir already has repository and a branch.  The branch
 
1232
    is a test double as well.
 
1233
    """
 
1234
 
 
1235
    def __init__(self, *args, **kwargs):
 
1236
        super(_TestBzrDir, self).__init__(*args, **kwargs)
 
1237
        self.test_branch = _TestBranch()
 
1238
        self.test_branch.repository = self.create_repository()
 
1239
 
 
1240
    def open_branch(self, unsupported=False):
 
1241
        return self.test_branch
 
1242
 
 
1243
    def cloning_metadir(self, require_stacking=False):
 
1244
        return _TestBzrDirFormat()
 
1245
 
 
1246
 
 
1247
class _TestBranchFormat(bzrlib.branch.BranchFormat):
 
1248
    """Test Branch format for TestBzrDirSprout."""
 
1249
 
 
1250
 
 
1251
class _TestBranch(bzrlib.branch.Branch):
 
1252
    """Test Branch implementation for TestBzrDirSprout."""
 
1253
 
 
1254
    def __init__(self, *args, **kwargs):
 
1255
        self._format = _TestBranchFormat()
 
1256
        super(_TestBranch, self).__init__(*args, **kwargs)
 
1257
        self.calls = []
 
1258
        self._parent = None
 
1259
 
 
1260
    def sprout(self, *args, **kwargs):
 
1261
        self.calls.append('sprout')
 
1262
        return _TestBranch()
 
1263
 
 
1264
    def copy_content_into(self, destination, revision_id=None):
 
1265
        self.calls.append('copy_content_into')
 
1266
 
 
1267
    def get_parent(self):
 
1268
        return self._parent
 
1269
 
 
1270
    def set_parent(self, parent):
 
1271
        self._parent = parent
 
1272
 
 
1273
 
 
1274
class TestBzrDirSprout(TestCaseWithMemoryTransport):
 
1275
 
 
1276
    def test_sprout_uses_branch_sprout(self):
 
1277
        """BzrDir.sprout calls Branch.sprout.
 
1278
 
 
1279
        Usually, BzrDir.sprout should delegate to the branch's sprout method
 
1280
        for part of the work.  This allows the source branch to control the
 
1281
        choice of format for the new branch.
 
1282
 
 
1283
        There are exceptions, but this tests avoids them:
 
1284
          - if there's no branch in the source bzrdir,
 
1285
          - or if the stacking has been requested and the format needs to be
 
1286
            overridden to satisfy that.
 
1287
        """
 
1288
        # Make an instrumented bzrdir.
 
1289
        t = self.get_transport('source')
 
1290
        t.ensure_base()
 
1291
        source_bzrdir = _TestBzrDirFormat().initialize_on_transport(t)
 
1292
        # The instrumented bzrdir has a test_branch attribute that logs calls
 
1293
        # made to the branch contained in that bzrdir.  Initially the test
 
1294
        # branch exists but no calls have been made to it.
 
1295
        self.assertEqual([], source_bzrdir.test_branch.calls)
 
1296
 
 
1297
        # Sprout the bzrdir
 
1298
        target_url = self.get_url('target')
 
1299
        result = source_bzrdir.sprout(target_url, recurse='no')
 
1300
 
 
1301
        # The bzrdir called the branch's sprout method.
 
1302
        self.assertSubset(['sprout'], source_bzrdir.test_branch.calls)
 
1303
 
 
1304
    def test_sprout_parent(self):
 
1305
        grandparent_tree = self.make_branch('grandparent')
 
1306
        parent = grandparent_tree.bzrdir.sprout('parent').open_branch()
 
1307
        branch_tree = parent.bzrdir.sprout('branch').open_branch()
 
1308
        self.assertContainsRe(branch_tree.get_parent(), '/parent/$')
 
1309
 
 
1310
 
 
1311
class TestBzrDirHooks(TestCaseWithMemoryTransport):
 
1312
 
 
1313
    def test_pre_open_called(self):
 
1314
        calls = []
 
1315
        bzrdir.BzrDir.hooks.install_named_hook('pre_open', calls.append, None)
 
1316
        transport = self.get_transport('foo')
 
1317
        url = transport.base
 
1318
        self.assertRaises(errors.NotBranchError, bzrdir.BzrDir.open, url)
 
1319
        self.assertEqual([transport.base], [t.base for t in calls])
 
1320
 
 
1321
    def test_pre_open_actual_exceptions_raised(self):
 
1322
        count = [0]
 
1323
        def fail_once(transport):
 
1324
            count[0] += 1
 
1325
            if count[0] == 1:
 
1326
                raise errors.BzrError("fail")
 
1327
        bzrdir.BzrDir.hooks.install_named_hook('pre_open', fail_once, None)
 
1328
        transport = self.get_transport('foo')
 
1329
        url = transport.base
 
1330
        err = self.assertRaises(errors.BzrError, bzrdir.BzrDir.open, url)
 
1331
        self.assertEqual('fail', err._preformatted_string)