← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_branch.py

Merge sftp-leaks into catch-them-all resolving conflicts

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_branch.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2006-2010 Canonical Ltd
2
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
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 Branch facility that are not interface  tests.
18
18
 
19
 
For interface tests see tests/branch_implementations/*.py.
 
19
For interface tests see tests/per_branch/*.py.
20
20
 
21
21
For concrete class tests see this file, and for meta-branch tests
22
22
also see this file.
23
23
"""
24
24
 
25
 
from StringIO import StringIO
26
 
 
27
 
import bzrlib.branch
28
 
from bzrlib.branch import (BzrBranch5, 
29
 
                           BzrBranchFormat5)
30
 
import bzrlib.bzrdir as bzrdir
31
 
from bzrlib.bzrdir import (BzrDirMetaFormat1, BzrDirMeta1, 
32
 
                           BzrDir, BzrDirFormat)
33
 
from bzrlib.errors import (NotBranchError,
34
 
                           UnknownFormatError,
35
 
                           UnsupportedFormatError,
36
 
                           )
37
 
 
38
 
from bzrlib.tests import TestCase, TestCaseWithTransport
39
 
from bzrlib.transport import get_transport
40
 
 
41
 
class TestDefaultFormat(TestCase):
 
25
from cStringIO import StringIO
 
26
 
 
27
from bzrlib import (
 
28
    branch as _mod_branch,
 
29
    bzrdir,
 
30
    config,
 
31
    errors,
 
32
    tests,
 
33
    trace,
 
34
    transport,
 
35
    urlutils,
 
36
    )
 
37
 
 
38
 
 
39
class TestDefaultFormat(tests.TestCase):
 
40
 
 
41
    def test_default_format(self):
 
42
        # update this if you change the default branch format
 
43
        self.assertIsInstance(_mod_branch.BranchFormat.get_default_format(),
 
44
                _mod_branch.BzrBranchFormat7)
 
45
 
 
46
    def test_default_format_is_same_as_bzrdir_default(self):
 
47
        # XXX: it might be nice if there was only one place the default was
 
48
        # set, but at the moment that's not true -- mbp 20070814 --
 
49
        # https://bugs.launchpad.net/bzr/+bug/132376
 
50
        self.assertEqual(
 
51
            _mod_branch.BranchFormat.get_default_format(),
 
52
            bzrdir.BzrDirFormat.get_default_format().get_branch_format())
42
53
 
43
54
    def test_get_set_default_format(self):
44
 
        old_format = bzrlib.branch.BranchFormat.get_default_format()
45
 
        # default is 5
46
 
        self.assertTrue(isinstance(old_format, bzrlib.branch.BzrBranchFormat5))
47
 
        bzrlib.branch.BranchFormat.set_default_format(SampleBranchFormat())
 
55
        # set the format and then set it back again
 
56
        old_format = _mod_branch.BranchFormat.get_default_format()
 
57
        _mod_branch.BranchFormat.set_default_format(SampleBranchFormat())
48
58
        try:
49
59
            # the default branch format is used by the meta dir format
50
60
            # which is not the default bzrdir format at this point
51
 
            dir = BzrDirMetaFormat1().initialize('memory:///')
 
61
            dir = bzrdir.BzrDirMetaFormat1().initialize('memory:///')
52
62
            result = dir.create_branch()
53
63
            self.assertEqual(result, 'A branch')
54
64
        finally:
55
 
            bzrlib.branch.BranchFormat.set_default_format(old_format)
56
 
        self.assertEqual(old_format, bzrlib.branch.BranchFormat.get_default_format())
57
 
 
58
 
 
59
 
class TestBranchFormat5(TestCaseWithTransport):
 
65
            _mod_branch.BranchFormat.set_default_format(old_format)
 
66
        self.assertEqual(old_format,
 
67
                         _mod_branch.BranchFormat.get_default_format())
 
68
 
 
69
 
 
70
class TestBranchFormat5(tests.TestCaseWithTransport):
60
71
    """Tests specific to branch format 5"""
61
72
 
62
73
    def test_branch_format_5_uses_lockdir(self):
63
74
        url = self.get_url()
64
 
        bzrdir = BzrDirMetaFormat1().initialize(url)
65
 
        bzrdir.create_repository()
66
 
        branch = bzrdir.create_branch()
 
75
        bdir = bzrdir.BzrDirMetaFormat1().initialize(url)
 
76
        bdir.create_repository()
 
77
        branch = bdir.create_branch()
67
78
        t = self.get_transport()
68
79
        self.log("branch instance is %r" % branch)
69
 
        self.assert_(isinstance(branch, BzrBranch5))
 
80
        self.assert_(isinstance(branch, _mod_branch.BzrBranch5))
70
81
        self.assertIsDirectory('.', t)
71
82
        self.assertIsDirectory('.bzr/branch', t)
72
83
        self.assertIsDirectory('.bzr/branch/lock', t)
73
84
        branch.lock_write()
 
85
        self.addCleanup(branch.unlock)
 
86
        self.assertIsDirectory('.bzr/branch/lock/held', t)
 
87
 
 
88
    def test_set_push_location(self):
 
89
        from bzrlib.config import (locations_config_filename,
 
90
                                   ensure_config_dir_exists)
 
91
        ensure_config_dir_exists()
 
92
        fn = locations_config_filename()
 
93
        # write correct newlines to locations.conf
 
94
        # by default ConfigObj uses native line-endings for new files
 
95
        # but uses already existing line-endings if file is not empty
 
96
        f = open(fn, 'wb')
74
97
        try:
75
 
            self.assertIsDirectory('.bzr/branch/lock/held', t)
 
98
            f.write('# comment\n')
76
99
        finally:
77
 
            branch.unlock()
78
 
 
79
 
 
80
 
class TestBranchEscaping(TestCaseWithTransport):
81
 
    """Test a branch can be correctly stored and used on a vfat-like transport
82
 
    
83
 
    Makes sure we have proper escaping of invalid characters, etc.
84
 
 
85
 
    It'd be better to test all operations on the FakeVFATTransportDecorator,
86
 
    but working trees go straight to the os not through the Transport layer.
87
 
    Therefore we build some history first in the regular way and then 
88
 
    check it's safe to access for vfat.
89
 
    """
90
 
 
91
 
    FOO_ID = 'foo<:>ID'
92
 
    REV_ID = 'revid-1'
93
 
 
94
 
    def setUp(self):
95
 
        super(TestBranchEscaping, self).setUp()
96
 
        from bzrlib.repository import RepositoryFormatKnit1
97
 
        bzrdir = BzrDirMetaFormat1().initialize(self.get_url())
98
 
        repo = RepositoryFormatKnit1().initialize(bzrdir)
99
 
        branch = bzrdir.create_branch()
100
 
        wt = bzrdir.create_workingtree()
101
 
        self.build_tree_contents([("foo", "contents of foo")])
102
 
        # add file with id containing wierd characters
103
 
        wt.add(['foo'], [self.FOO_ID])
104
 
        wt.commit('this is my new commit', rev_id=self.REV_ID)
105
 
 
106
 
    def test_branch_on_vfat(self):
107
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
108
 
        # now access over vfat; should be safe
109
 
        transport = FakeVFATTransportDecorator('vfat+' + self.get_url())
110
 
        bzrdir, junk = BzrDir.open_containing_from_transport(transport)
111
 
        branch = bzrdir.open_branch()
112
 
        revtree = branch.repository.revision_tree(self.REV_ID)
113
 
        contents = revtree.get_file_text(self.FOO_ID)
114
 
        self.assertEqual(contents, 'contents of foo')
115
 
 
116
 
 
117
 
class SampleBranchFormat(bzrlib.branch.BranchFormat):
 
100
            f.close()
 
101
 
 
102
        branch = self.make_branch('.', format='knit')
 
103
        branch.set_push_location('foo')
 
104
        local_path = urlutils.local_path_from_url(branch.base[:-1])
 
105
        self.assertFileEqual("# comment\n"
 
106
                             "[%s]\n"
 
107
                             "push_location = foo\n"
 
108
                             "push_location:policy = norecurse\n" % local_path,
 
109
                             fn)
 
110
 
 
111
    # TODO RBC 20051029 test getting a push location from a branch in a
 
112
    # recursive section - that is, it appends the branch name.
 
113
 
 
114
 
 
115
class SampleBranchFormat(_mod_branch.BranchFormat):
118
116
    """A sample format
119
117
 
120
 
    this format is initializable, unsupported to aid in testing the 
 
118
    this format is initializable, unsupported to aid in testing the
121
119
    open and open_downlevel routines.
122
120
    """
123
121
 
125
123
        """See BzrBranchFormat.get_format_string()."""
126
124
        return "Sample branch format."
127
125
 
128
 
    def initialize(self, a_bzrdir):
 
126
    def initialize(self, a_bzrdir, name=None):
129
127
        """Format 4 branches cannot be created."""
130
 
        t = a_bzrdir.get_branch_transport(self)
131
 
        t.put('format', StringIO(self.get_format_string()))
 
128
        t = a_bzrdir.get_branch_transport(self, name=name)
 
129
        t.put_bytes('format', self.get_format_string())
132
130
        return 'A branch'
133
131
 
134
132
    def is_supported(self):
135
133
        return False
136
134
 
137
 
    def open(self, transport, _found=False):
 
135
    def open(self, transport, name=None, _found=False, ignore_fallbacks=False):
138
136
        return "opened branch."
139
137
 
140
138
 
141
 
class TestBzrBranchFormat(TestCaseWithTransport):
 
139
# Demonstrating how lazy loading is often implemented:
 
140
# A constant string is created.
 
141
SampleSupportedBranchFormatString = "Sample supported branch format."
 
142
 
 
143
# And the format class can then reference the constant to avoid skew.
 
144
class SampleSupportedBranchFormat(_mod_branch.BranchFormat):
 
145
    """A sample supported format."""
 
146
 
 
147
    def get_format_string(self):
 
148
        """See BzrBranchFormat.get_format_string()."""
 
149
        return SampleSupportedBranchFormatString
 
150
 
 
151
    def initialize(self, a_bzrdir, name=None):
 
152
        t = a_bzrdir.get_branch_transport(self, name=name)
 
153
        t.put_bytes('format', self.get_format_string())
 
154
        return 'A branch'
 
155
 
 
156
    def open(self, transport, name=None, _found=False, ignore_fallbacks=False):
 
157
        return "opened supported branch."
 
158
 
 
159
 
 
160
class TestBzrBranchFormat(tests.TestCaseWithTransport):
142
161
    """Tests for the BzrBranchFormat facility."""
143
162
 
144
163
    def test_find_format(self):
145
164
        # is the right format object found for a branch?
146
165
        # create a branch with a few known format objects.
147
 
        # this is not quite the same as 
 
166
        # this is not quite the same as
148
167
        self.build_tree(["foo/", "bar/"])
149
168
        def check_format(format, url):
150
169
            dir = format._matchingbzrdir.initialize(url)
151
170
            dir.create_repository()
152
171
            format.initialize(dir)
153
 
            found_format = bzrlib.branch.BranchFormat.find_format(dir)
 
172
            found_format = _mod_branch.BranchFormat.find_format(dir)
154
173
            self.failUnless(isinstance(found_format, format.__class__))
155
 
        check_format(bzrlib.branch.BzrBranchFormat5(), "bar")
156
 
        
 
174
        check_format(_mod_branch.BzrBranchFormat5(), "bar")
 
175
 
 
176
    def test_find_format_factory(self):
 
177
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
178
        SampleSupportedBranchFormat().initialize(dir)
 
179
        factory = _mod_branch.MetaDirBranchFormatFactory(
 
180
            SampleSupportedBranchFormatString,
 
181
            "bzrlib.tests.test_branch", "SampleSupportedBranchFormat")
 
182
        _mod_branch.BranchFormat.register_format(factory)
 
183
        self.addCleanup(_mod_branch.BranchFormat.unregister_format, factory)
 
184
        b = _mod_branch.Branch.open(self.get_url())
 
185
        self.assertEqual(b, "opened supported branch.")
 
186
 
157
187
    def test_find_format_not_branch(self):
158
188
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
159
 
        self.assertRaises(NotBranchError,
160
 
                          bzrlib.branch.BranchFormat.find_format,
 
189
        self.assertRaises(errors.NotBranchError,
 
190
                          _mod_branch.BranchFormat.find_format,
161
191
                          dir)
162
192
 
163
193
    def test_find_format_unknown_format(self):
164
194
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
165
195
        SampleBranchFormat().initialize(dir)
166
 
        self.assertRaises(UnknownFormatError,
167
 
                          bzrlib.branch.BranchFormat.find_format,
 
196
        self.assertRaises(errors.UnknownFormatError,
 
197
                          _mod_branch.BranchFormat.find_format,
168
198
                          dir)
169
199
 
170
200
    def test_register_unregister_format(self):
174
204
        # make a branch
175
205
        format.initialize(dir)
176
206
        # register a format for it.
177
 
        bzrlib.branch.BranchFormat.register_format(format)
 
207
        _mod_branch.BranchFormat.register_format(format)
178
208
        # which branch.Open will refuse (not supported)
179
 
        self.assertRaises(UnsupportedFormatError, bzrlib.branch.Branch.open, self.get_url())
 
209
        self.assertRaises(errors.UnsupportedFormatError,
 
210
                          _mod_branch.Branch.open, self.get_url())
 
211
        self.make_branch_and_tree('foo')
180
212
        # but open_downlevel will work
181
 
        self.assertEqual(format.open(dir), bzrdir.BzrDir.open(self.get_url()).open_branch(unsupported=True))
 
213
        self.assertEqual(
 
214
            format.open(dir),
 
215
            bzrdir.BzrDir.open(self.get_url()).open_branch(unsupported=True))
182
216
        # unregister the format
183
 
        bzrlib.branch.BranchFormat.unregister_format(format)
184
 
 
185
 
 
186
 
class TestBranchReference(TestCaseWithTransport):
 
217
        _mod_branch.BranchFormat.unregister_format(format)
 
218
        self.make_branch_and_tree('bar')
 
219
 
 
220
 
 
221
#Used by TestMetaDirBranchFormatFactory 
 
222
FakeLazyFormat = None
 
223
 
 
224
 
 
225
class TestMetaDirBranchFormatFactory(tests.TestCase):
 
226
 
 
227
    def test_get_format_string_does_not_load(self):
 
228
        """Formats have a static format string."""
 
229
        factory = _mod_branch.MetaDirBranchFormatFactory("yo", None, None)
 
230
        self.assertEqual("yo", factory.get_format_string())
 
231
 
 
232
    def test_call_loads(self):
 
233
        # __call__ is used by the network_format_registry interface to get a
 
234
        # Format.
 
235
        global FakeLazyFormat
 
236
        del FakeLazyFormat
 
237
        factory = _mod_branch.MetaDirBranchFormatFactory(None,
 
238
            "bzrlib.tests.test_branch", "FakeLazyFormat")
 
239
        self.assertRaises(AttributeError, factory)
 
240
 
 
241
    def test_call_returns_call_of_referenced_object(self):
 
242
        global FakeLazyFormat
 
243
        FakeLazyFormat = lambda:'called'
 
244
        factory = _mod_branch.MetaDirBranchFormatFactory(None,
 
245
            "bzrlib.tests.test_branch", "FakeLazyFormat")
 
246
        self.assertEqual('called', factory())
 
247
 
 
248
 
 
249
class TestBranch67(object):
 
250
    """Common tests for both branch 6 and 7 which are mostly the same."""
 
251
 
 
252
    def get_format_name(self):
 
253
        raise NotImplementedError(self.get_format_name)
 
254
 
 
255
    def get_format_name_subtree(self):
 
256
        raise NotImplementedError(self.get_format_name)
 
257
 
 
258
    def get_class(self):
 
259
        raise NotImplementedError(self.get_class)
 
260
 
 
261
    def test_creation(self):
 
262
        format = bzrdir.BzrDirMetaFormat1()
 
263
        format.set_branch_format(_mod_branch.BzrBranchFormat6())
 
264
        branch = self.make_branch('a', format=format)
 
265
        self.assertIsInstance(branch, self.get_class())
 
266
        branch = self.make_branch('b', format=self.get_format_name())
 
267
        self.assertIsInstance(branch, self.get_class())
 
268
        branch = _mod_branch.Branch.open('a')
 
269
        self.assertIsInstance(branch, self.get_class())
 
270
 
 
271
    def test_layout(self):
 
272
        branch = self.make_branch('a', format=self.get_format_name())
 
273
        self.failUnlessExists('a/.bzr/branch/last-revision')
 
274
        self.failIfExists('a/.bzr/branch/revision-history')
 
275
        self.failIfExists('a/.bzr/branch/references')
 
276
 
 
277
    def test_config(self):
 
278
        """Ensure that all configuration data is stored in the branch"""
 
279
        branch = self.make_branch('a', format=self.get_format_name())
 
280
        branch.set_parent('http://bazaar-vcs.org')
 
281
        self.failIfExists('a/.bzr/branch/parent')
 
282
        self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
 
283
        branch.set_push_location('sftp://bazaar-vcs.org')
 
284
        config = branch.get_config()._get_branch_data_config()
 
285
        self.assertEqual('sftp://bazaar-vcs.org',
 
286
                         config.get_user_option('push_location'))
 
287
        branch.set_bound_location('ftp://bazaar-vcs.org')
 
288
        self.failIfExists('a/.bzr/branch/bound')
 
289
        self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
 
290
 
 
291
    def test_set_revision_history(self):
 
292
        builder = self.make_branch_builder('.', format=self.get_format_name())
 
293
        builder.build_snapshot('foo', None,
 
294
            [('add', ('', None, 'directory', None))],
 
295
            message='foo')
 
296
        builder.build_snapshot('bar', None, [], message='bar')
 
297
        branch = builder.get_branch()
 
298
        branch.lock_write()
 
299
        self.addCleanup(branch.unlock)
 
300
        branch.set_revision_history(['foo', 'bar'])
 
301
        branch.set_revision_history(['foo'])
 
302
        self.assertRaises(errors.NotLefthandHistory,
 
303
                          branch.set_revision_history, ['bar'])
 
304
 
 
305
    def do_checkout_test(self, lightweight=False):
 
306
        tree = self.make_branch_and_tree('source',
 
307
            format=self.get_format_name_subtree())
 
308
        subtree = self.make_branch_and_tree('source/subtree',
 
309
            format=self.get_format_name_subtree())
 
310
        subsubtree = self.make_branch_and_tree('source/subtree/subsubtree',
 
311
            format=self.get_format_name_subtree())
 
312
        self.build_tree(['source/subtree/file',
 
313
                         'source/subtree/subsubtree/file'])
 
314
        subsubtree.add('file')
 
315
        subtree.add('file')
 
316
        subtree.add_reference(subsubtree)
 
317
        tree.add_reference(subtree)
 
318
        tree.commit('a revision')
 
319
        subtree.commit('a subtree file')
 
320
        subsubtree.commit('a subsubtree file')
 
321
        tree.branch.create_checkout('target', lightweight=lightweight)
 
322
        self.failUnlessExists('target')
 
323
        self.failUnlessExists('target/subtree')
 
324
        self.failUnlessExists('target/subtree/file')
 
325
        self.failUnlessExists('target/subtree/subsubtree/file')
 
326
        subbranch = _mod_branch.Branch.open('target/subtree/subsubtree')
 
327
        if lightweight:
 
328
            self.assertEndsWith(subbranch.base, 'source/subtree/subsubtree/')
 
329
        else:
 
330
            self.assertEndsWith(subbranch.base, 'target/subtree/subsubtree/')
 
331
 
 
332
    def test_checkout_with_references(self):
 
333
        self.do_checkout_test()
 
334
 
 
335
    def test_light_checkout_with_references(self):
 
336
        self.do_checkout_test(lightweight=True)
 
337
 
 
338
    def test_set_push(self):
 
339
        branch = self.make_branch('source', format=self.get_format_name())
 
340
        branch.get_config().set_user_option('push_location', 'old',
 
341
            store=config.STORE_LOCATION)
 
342
        warnings = []
 
343
        def warning(*args):
 
344
            warnings.append(args[0] % args[1:])
 
345
        _warning = trace.warning
 
346
        trace.warning = warning
 
347
        try:
 
348
            branch.set_push_location('new')
 
349
        finally:
 
350
            trace.warning = _warning
 
351
        self.assertEqual(warnings[0], 'Value "new" is masked by "old" from '
 
352
                         'locations.conf')
 
353
 
 
354
 
 
355
class TestBranch6(TestBranch67, tests.TestCaseWithTransport):
 
356
 
 
357
    def get_class(self):
 
358
        return _mod_branch.BzrBranch6
 
359
 
 
360
    def get_format_name(self):
 
361
        return "dirstate-tags"
 
362
 
 
363
    def get_format_name_subtree(self):
 
364
        return "dirstate-with-subtree"
 
365
 
 
366
    def test_set_stacked_on_url_errors(self):
 
367
        branch = self.make_branch('a', format=self.get_format_name())
 
368
        self.assertRaises(errors.UnstackableBranchFormat,
 
369
            branch.set_stacked_on_url, None)
 
370
 
 
371
    def test_default_stacked_location(self):
 
372
        branch = self.make_branch('a', format=self.get_format_name())
 
373
        self.assertRaises(errors.UnstackableBranchFormat, branch.get_stacked_on_url)
 
374
 
 
375
 
 
376
class TestBranch7(TestBranch67, tests.TestCaseWithTransport):
 
377
 
 
378
    def get_class(self):
 
379
        return _mod_branch.BzrBranch7
 
380
 
 
381
    def get_format_name(self):
 
382
        return "1.9"
 
383
 
 
384
    def get_format_name_subtree(self):
 
385
        return "development-subtree"
 
386
 
 
387
    def test_set_stacked_on_url_unstackable_repo(self):
 
388
        repo = self.make_repository('a', format='dirstate-tags')
 
389
        control = repo.bzrdir
 
390
        branch = _mod_branch.BzrBranchFormat7().initialize(control)
 
391
        target = self.make_branch('b')
 
392
        self.assertRaises(errors.UnstackableRepositoryFormat,
 
393
            branch.set_stacked_on_url, target.base)
 
394
 
 
395
    def test_clone_stacked_on_unstackable_repo(self):
 
396
        repo = self.make_repository('a', format='dirstate-tags')
 
397
        control = repo.bzrdir
 
398
        branch = _mod_branch.BzrBranchFormat7().initialize(control)
 
399
        # Calling clone should not raise UnstackableRepositoryFormat.
 
400
        cloned_bzrdir = control.clone('cloned')
 
401
 
 
402
    def _test_default_stacked_location(self):
 
403
        branch = self.make_branch('a', format=self.get_format_name())
 
404
        self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
 
405
 
 
406
    def test_stack_and_unstack(self):
 
407
        branch = self.make_branch('a', format=self.get_format_name())
 
408
        target = self.make_branch_and_tree('b', format=self.get_format_name())
 
409
        branch.set_stacked_on_url(target.branch.base)
 
410
        self.assertEqual(target.branch.base, branch.get_stacked_on_url())
 
411
        revid = target.commit('foo')
 
412
        self.assertTrue(branch.repository.has_revision(revid))
 
413
        branch.set_stacked_on_url(None)
 
414
        self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
 
415
        self.assertFalse(branch.repository.has_revision(revid))
 
416
 
 
417
    def test_open_opens_stacked_reference(self):
 
418
        branch = self.make_branch('a', format=self.get_format_name())
 
419
        target = self.make_branch_and_tree('b', format=self.get_format_name())
 
420
        branch.set_stacked_on_url(target.branch.base)
 
421
        branch = branch.bzrdir.open_branch()
 
422
        revid = target.commit('foo')
 
423
        self.assertTrue(branch.repository.has_revision(revid))
 
424
 
 
425
 
 
426
class BzrBranch8(tests.TestCaseWithTransport):
 
427
 
 
428
    def make_branch(self, location, format=None):
 
429
        if format is None:
 
430
            format = bzrdir.format_registry.make_bzrdir('1.9')
 
431
            format.set_branch_format(_mod_branch.BzrBranchFormat8())
 
432
        return tests.TestCaseWithTransport.make_branch(
 
433
            self, location, format=format)
 
434
 
 
435
    def create_branch_with_reference(self):
 
436
        branch = self.make_branch('branch')
 
437
        branch._set_all_reference_info({'file-id': ('path', 'location')})
 
438
        return branch
 
439
 
 
440
    @staticmethod
 
441
    def instrument_branch(branch, gets):
 
442
        old_get = branch._transport.get
 
443
        def get(*args, **kwargs):
 
444
            gets.append((args, kwargs))
 
445
            return old_get(*args, **kwargs)
 
446
        branch._transport.get = get
 
447
 
 
448
    def test_reference_info_caching_read_locked(self):
 
449
        gets = []
 
450
        branch = self.create_branch_with_reference()
 
451
        branch.lock_read()
 
452
        self.addCleanup(branch.unlock)
 
453
        self.instrument_branch(branch, gets)
 
454
        branch.get_reference_info('file-id')
 
455
        branch.get_reference_info('file-id')
 
456
        self.assertEqual(1, len(gets))
 
457
 
 
458
    def test_reference_info_caching_read_unlocked(self):
 
459
        gets = []
 
460
        branch = self.create_branch_with_reference()
 
461
        self.instrument_branch(branch, gets)
 
462
        branch.get_reference_info('file-id')
 
463
        branch.get_reference_info('file-id')
 
464
        self.assertEqual(2, len(gets))
 
465
 
 
466
    def test_reference_info_caching_write_locked(self):
 
467
        gets = []
 
468
        branch = self.make_branch('branch')
 
469
        branch.lock_write()
 
470
        self.instrument_branch(branch, gets)
 
471
        self.addCleanup(branch.unlock)
 
472
        branch._set_all_reference_info({'file-id': ('path2', 'location2')})
 
473
        path, location = branch.get_reference_info('file-id')
 
474
        self.assertEqual(0, len(gets))
 
475
        self.assertEqual('path2', path)
 
476
        self.assertEqual('location2', location)
 
477
 
 
478
    def test_reference_info_caches_cleared(self):
 
479
        branch = self.make_branch('branch')
 
480
        branch.lock_write()
 
481
        branch.set_reference_info('file-id', 'path2', 'location2')
 
482
        branch.unlock()
 
483
        doppelganger = _mod_branch.Branch.open('branch')
 
484
        doppelganger.set_reference_info('file-id', 'path3', 'location3')
 
485
        self.assertEqual(('path3', 'location3'),
 
486
                         branch.get_reference_info('file-id'))
 
487
 
 
488
class TestBranchReference(tests.TestCaseWithTransport):
187
489
    """Tests for the branch reference facility."""
188
490
 
189
491
    def test_create_open_reference(self):
190
492
        bzrdirformat = bzrdir.BzrDirMetaFormat1()
191
 
        t = get_transport(self.get_url('.'))
 
493
        t = transport.get_transport(self.get_url('.'))
192
494
        t.mkdir('repo')
193
495
        dir = bzrdirformat.initialize(self.get_url('repo'))
194
496
        dir.create_repository()
195
497
        target_branch = dir.create_branch()
196
498
        t.mkdir('branch')
197
499
        branch_dir = bzrdirformat.initialize(self.get_url('branch'))
198
 
        made_branch = bzrlib.branch.BranchReferenceFormat().initialize(branch_dir, target_branch)
 
500
        made_branch = _mod_branch.BranchReferenceFormat().initialize(
 
501
            branch_dir, target_branch=target_branch)
199
502
        self.assertEqual(made_branch.base, target_branch.base)
200
503
        opened_branch = branch_dir.open_branch()
201
504
        self.assertEqual(opened_branch.base, target_branch.base)
 
505
 
 
506
    def test_get_reference(self):
 
507
        """For a BranchReference, get_reference should reutrn the location."""
 
508
        branch = self.make_branch('target')
 
509
        checkout = branch.create_checkout('checkout', lightweight=True)
 
510
        reference_url = branch.bzrdir.root_transport.abspath('') + '/'
 
511
        # if the api for create_checkout changes to return different checkout types
 
512
        # then this file read will fail.
 
513
        self.assertFileEqual(reference_url, 'checkout/.bzr/branch/location')
 
514
        self.assertEqual(reference_url,
 
515
            _mod_branch.BranchReferenceFormat().get_reference(checkout.bzrdir))
 
516
 
 
517
 
 
518
class TestHooks(tests.TestCaseWithTransport):
 
519
 
 
520
    def test_constructor(self):
 
521
        """Check that creating a BranchHooks instance has the right defaults."""
 
522
        hooks = _mod_branch.BranchHooks()
 
523
        self.assertTrue("set_rh" in hooks, "set_rh not in %s" % hooks)
 
524
        self.assertTrue("post_push" in hooks, "post_push not in %s" % hooks)
 
525
        self.assertTrue("post_commit" in hooks, "post_commit not in %s" % hooks)
 
526
        self.assertTrue("pre_commit" in hooks, "pre_commit not in %s" % hooks)
 
527
        self.assertTrue("post_pull" in hooks, "post_pull not in %s" % hooks)
 
528
        self.assertTrue("post_uncommit" in hooks,
 
529
                        "post_uncommit not in %s" % hooks)
 
530
        self.assertTrue("post_change_branch_tip" in hooks,
 
531
                        "post_change_branch_tip not in %s" % hooks)
 
532
        self.assertTrue("post_branch_init" in hooks,
 
533
                        "post_branch_init not in %s" % hooks)
 
534
        self.assertTrue("post_switch" in hooks,
 
535
                        "post_switch not in %s" % hooks)
 
536
 
 
537
    def test_installed_hooks_are_BranchHooks(self):
 
538
        """The installed hooks object should be a BranchHooks."""
 
539
        # the installed hooks are saved in self._preserved_hooks.
 
540
        self.assertIsInstance(self._preserved_hooks[_mod_branch.Branch][1],
 
541
                              _mod_branch.BranchHooks)
 
542
 
 
543
    def test_post_branch_init_hook(self):
 
544
        calls = []
 
545
        _mod_branch.Branch.hooks.install_named_hook('post_branch_init',
 
546
            calls.append, None)
 
547
        self.assertLength(0, calls)
 
548
        branch = self.make_branch('a')
 
549
        self.assertLength(1, calls)
 
550
        params = calls[0]
 
551
        self.assertIsInstance(params, _mod_branch.BranchInitHookParams)
 
552
        self.assertTrue(hasattr(params, 'bzrdir'))
 
553
        self.assertTrue(hasattr(params, 'branch'))
 
554
 
 
555
    def test_post_switch_hook(self):
 
556
        from bzrlib import switch
 
557
        calls = []
 
558
        _mod_branch.Branch.hooks.install_named_hook('post_switch',
 
559
            calls.append, None)
 
560
        tree = self.make_branch_and_tree('branch-1')
 
561
        self.build_tree(['branch-1/file-1'])
 
562
        tree.add('file-1')
 
563
        tree.commit('rev1')
 
564
        to_branch = tree.bzrdir.sprout('branch-2').open_branch()
 
565
        self.build_tree(['branch-1/file-2'])
 
566
        tree.add('file-2')
 
567
        tree.remove('file-1')
 
568
        tree.commit('rev2')
 
569
        checkout = tree.branch.create_checkout('checkout')
 
570
        self.assertLength(0, calls)
 
571
        switch.switch(checkout.bzrdir, to_branch)
 
572
        self.assertLength(1, calls)
 
573
        params = calls[0]
 
574
        self.assertIsInstance(params, _mod_branch.SwitchHookParams)
 
575
        self.assertTrue(hasattr(params, 'to_branch'))
 
576
        self.assertTrue(hasattr(params, 'revision_id'))
 
577
 
 
578
 
 
579
class TestBranchOptions(tests.TestCaseWithTransport):
 
580
 
 
581
    def setUp(self):
 
582
        super(TestBranchOptions, self).setUp()
 
583
        self.branch = self.make_branch('.')
 
584
        self.config = self.branch.get_config()
 
585
 
 
586
    def check_append_revisions_only(self, expected_value, value=None):
 
587
        """Set append_revisions_only in config and check its interpretation."""
 
588
        if value is not None:
 
589
            self.config.set_user_option('append_revisions_only', value)
 
590
        self.assertEqual(expected_value,
 
591
                         self.branch._get_append_revisions_only())
 
592
 
 
593
    def test_valid_append_revisions_only(self):
 
594
        self.assertEquals(None,
 
595
                          self.config.get_user_option('append_revisions_only'))
 
596
        self.check_append_revisions_only(None)
 
597
        self.check_append_revisions_only(False, 'False')
 
598
        self.check_append_revisions_only(True, 'True')
 
599
        # The following values will cause compatibility problems on projects
 
600
        # using older bzr versions (<2.2) but are accepted
 
601
        self.check_append_revisions_only(False, 'false')
 
602
        self.check_append_revisions_only(True, 'true')
 
603
 
 
604
    def test_invalid_append_revisions_only(self):
 
605
        """Ensure warning is noted on invalid settings"""
 
606
        self.warnings = []
 
607
        def warning(*args):
 
608
            self.warnings.append(args[0] % args[1:])
 
609
        self.overrideAttr(trace, 'warning', warning)
 
610
        self.check_append_revisions_only(None, 'not-a-bool')
 
611
        self.assertLength(1, self.warnings)
 
612
        self.assertEqual(
 
613
            'Value "not-a-bool" is not a boolean for "append_revisions_only"',
 
614
            self.warnings[0])
 
615
 
 
616
 
 
617
class TestPullResult(tests.TestCase):
 
618
 
 
619
    def test_pull_result_to_int(self):
 
620
        # to support old code, the pull result can be used as an int
 
621
        r = _mod_branch.PullResult()
 
622
        r.old_revno = 10
 
623
        r.new_revno = 20
 
624
        # this usage of results is not recommended for new code (because it
 
625
        # doesn't describe very well what happened), but for api stability
 
626
        # it's still supported
 
627
        a = "%d revisions pulled" % r
 
628
        self.assertEqual(a, "10 revisions pulled")
 
629
 
 
630
    def test_report_changed(self):
 
631
        r = _mod_branch.PullResult()
 
632
        r.old_revid = "old-revid"
 
633
        r.old_revno = 10
 
634
        r.new_revid = "new-revid"
 
635
        r.new_revno = 20
 
636
        f = StringIO()
 
637
        r.report(f)
 
638
        self.assertEqual("Now on revision 20.\n", f.getvalue())
 
639
 
 
640
    def test_report_unchanged(self):
 
641
        r = _mod_branch.PullResult()
 
642
        r.old_revid = "same-revid"
 
643
        r.new_revid = "same-revid"
 
644
        f = StringIO()
 
645
        r.report(f)
 
646
        self.assertEqual("No revisions to pull.\n", f.getvalue())
 
647
 
 
648
 
 
649
class _StubLockable(object):
 
650
    """Helper for TestRunWithWriteLockedTarget."""
 
651
 
 
652
    def __init__(self, calls, unlock_exc=None):
 
653
        self.calls = calls
 
654
        self.unlock_exc = unlock_exc
 
655
 
 
656
    def lock_write(self):
 
657
        self.calls.append('lock_write')
 
658
 
 
659
    def unlock(self):
 
660
        self.calls.append('unlock')
 
661
        if self.unlock_exc is not None:
 
662
            raise self.unlock_exc
 
663
 
 
664
 
 
665
class _ErrorFromCallable(Exception):
 
666
    """Helper for TestRunWithWriteLockedTarget."""
 
667
 
 
668
 
 
669
class _ErrorFromUnlock(Exception):
 
670
    """Helper for TestRunWithWriteLockedTarget."""
 
671
 
 
672
 
 
673
class TestRunWithWriteLockedTarget(tests.TestCase):
 
674
    """Tests for _run_with_write_locked_target."""
 
675
 
 
676
    def setUp(self):
 
677
        tests.TestCase.setUp(self)
 
678
        self._calls = []
 
679
 
 
680
    def func_that_returns_ok(self):
 
681
        self._calls.append('func called')
 
682
        return 'ok'
 
683
 
 
684
    def func_that_raises(self):
 
685
        self._calls.append('func called')
 
686
        raise _ErrorFromCallable()
 
687
 
 
688
    def test_success_unlocks(self):
 
689
        lockable = _StubLockable(self._calls)
 
690
        result = _mod_branch._run_with_write_locked_target(
 
691
            lockable, self.func_that_returns_ok)
 
692
        self.assertEqual('ok', result)
 
693
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
694
 
 
695
    def test_exception_unlocks_and_propagates(self):
 
696
        lockable = _StubLockable(self._calls)
 
697
        self.assertRaises(_ErrorFromCallable,
 
698
                          _mod_branch._run_with_write_locked_target,
 
699
                          lockable, self.func_that_raises)
 
700
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
701
 
 
702
    def test_callable_succeeds_but_error_during_unlock(self):
 
703
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
704
        self.assertRaises(_ErrorFromUnlock,
 
705
                          _mod_branch._run_with_write_locked_target,
 
706
                          lockable, self.func_that_returns_ok)
 
707
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
708
 
 
709
    def test_error_during_unlock_does_not_mask_original_error(self):
 
710
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
711
        self.assertRaises(_ErrorFromCallable,
 
712
                          _mod_branch._run_with_write_locked_target,
 
713
                          lockable, self.func_that_raises)
 
714
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
715
 
 
716