~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

  • Committer: Martin Pool
  • Date: 2005-07-21 21:32:13 UTC
  • Revision ID: mbp@sourcefrog.net-20050721213213-c6ac0e8b06eaad0f
- bzr update-hashes shows some stats on what it did

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
2
 
#
 
1
# (C) 2005 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
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
 
"""Tests for branch implementations - tests a branch format."""
18
 
 
19
 
import os
20
 
import sys
21
 
 
22
 
from bzrlib import (
23
 
    branch,
24
 
    bzrdir,
25
 
    errors,
26
 
    gpg,
27
 
    urlutils,
28
 
    transactions,
29
 
    remote,
30
 
    repository,
31
 
    tests,
32
 
    )
33
 
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
34
 
from bzrlib.delta import TreeDelta
35
 
from bzrlib.errors import (FileExists,
36
 
                           NoSuchRevision,
37
 
                           NoSuchFile,
38
 
                           UninitializableFormat,
39
 
                           NotBranchError,
40
 
                           )
41
 
from bzrlib.osutils import getcwd
42
 
import bzrlib.revision
43
 
from bzrlib.symbol_versioning import deprecated_in
44
 
from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
45
 
from bzrlib.tests.branch_implementations import TestCaseWithBranch
46
 
from bzrlib.tests.http_server import HttpServer
47
 
from bzrlib.trace import mutter
48
 
from bzrlib.transport import get_transport
49
 
from bzrlib.transport.memory import MemoryServer
50
 
from bzrlib.upgrade import upgrade
51
 
from bzrlib.workingtree import WorkingTree
52
 
 
53
 
 
54
 
class TestTestCaseWithBranch(TestCaseWithBranch):
55
 
 
56
 
    def test_branch_format_matches_bzrdir_branch_format(self):
57
 
        bzrdir_branch_format = self.bzrdir_format.get_branch_format()
58
 
        self.assertIs(self.branch_format.__class__,
59
 
                      bzrdir_branch_format.__class__)
60
 
 
61
 
    def test_make_branch_gets_expected_format(self):
62
 
        branch = self.make_branch('.')
63
 
        self.assertIs(self.branch_format.__class__,
64
 
            branch._format.__class__)
65
 
 
66
 
 
67
 
class TestBranch(TestCaseWithBranch):
68
 
 
69
 
    def test_create_tree_with_merge(self):
70
 
        tree = self.create_tree_with_merge()
71
 
        tree.lock_read()
72
 
        self.addCleanup(tree.unlock)
73
 
        graph = tree.branch.repository.get_graph()
74
 
        ancestry_graph = graph.get_parent_map(
75
 
            tree.branch.repository.all_revision_ids())
76
 
        self.assertEqual({'rev-1':('null:',),
77
 
                          'rev-2':('rev-1', ),
78
 
                          'rev-1.1.1':('rev-1', ),
79
 
                          'rev-3':('rev-2', 'rev-1.1.1', ),
80
 
                         }, ancestry_graph)
81
 
 
82
 
    def test_revision_ids_are_utf8(self):
83
 
        wt = self.make_branch_and_tree('tree')
84
 
        wt.commit('f', rev_id='rev1')
85
 
        wt.commit('f', rev_id='rev2')
86
 
        wt.commit('f', rev_id='rev3')
87
 
 
88
 
        br = self.get_branch()
89
 
        br.fetch(wt.branch)
90
 
        br.set_revision_history(['rev1', 'rev2', 'rev3'])
91
 
        rh = br.revision_history()
92
 
        self.assertEqual(['rev1', 'rev2', 'rev3'], rh)
93
 
        for revision_id in rh:
94
 
            self.assertIsInstance(revision_id, str)
95
 
        last = br.last_revision()
96
 
        self.assertEqual('rev3', last)
97
 
        self.assertIsInstance(last, str)
98
 
        revno, last = br.last_revision_info()
99
 
        self.assertEqual(3, revno)
100
 
        self.assertEqual('rev3', last)
101
 
        self.assertIsInstance(last, str)
102
 
 
103
 
    def test_fetch_revisions(self):
104
 
        """Test fetch-revision operation."""
105
 
        wt = self.make_branch_and_tree('b1')
106
 
        b1 = wt.branch
107
 
        self.build_tree_contents([('b1/foo', 'hello')])
108
 
        wt.add(['foo'], ['foo-id'])
109
 
        wt.commit('lala!', rev_id='revision-1', allow_pointless=False)
110
 
 
111
 
        b2 = self.make_branch('b2')
112
 
        self.assertEqual((1, []), b2.fetch(b1))
113
 
 
114
 
        rev = b2.repository.get_revision('revision-1')
115
 
        tree = b2.repository.revision_tree('revision-1')
116
 
        tree.lock_read()
117
 
        self.addCleanup(tree.unlock)
118
 
        self.assertEqual(tree.get_file_text('foo-id'), 'hello')
119
 
 
120
 
    def test_get_revision_delta(self):
121
 
        tree_a = self.make_branch_and_tree('a')
122
 
        self.build_tree(['a/foo'])
123
 
        tree_a.add('foo', 'file1')
124
 
        tree_a.commit('rev1', rev_id='rev1')
125
 
        self.build_tree(['a/vla'])
126
 
        tree_a.add('vla', 'file2')
127
 
        tree_a.commit('rev2', rev_id='rev2')
128
 
 
129
 
        delta = tree_a.branch.get_revision_delta(1)
130
 
        self.assertIsInstance(delta, TreeDelta)
131
 
        self.assertEqual([('foo', 'file1', 'file')], delta.added)
132
 
        delta = tree_a.branch.get_revision_delta(2)
133
 
        self.assertIsInstance(delta, TreeDelta)
134
 
        self.assertEqual([('vla', 'file2', 'file')], delta.added)
135
 
 
136
 
    def get_unbalanced_tree_pair(self):
137
 
        """Return two branches, a and b, with one file in a."""
138
 
        tree_a = self.make_branch_and_tree('a')
139
 
        self.build_tree_contents([('a/b', 'b')])
140
 
        tree_a.add('b')
141
 
        tree_a.commit("silly commit", rev_id='A')
142
 
 
143
 
        tree_b = self.make_branch_and_tree('b')
144
 
        return tree_a, tree_b
145
 
 
146
 
    def get_balanced_branch_pair(self):
147
 
        """Returns br_a, br_b as with one commit in a, and b has a's stores."""
148
 
        tree_a, tree_b = self.get_unbalanced_tree_pair()
149
 
        tree_b.branch.repository.fetch(tree_a.branch.repository)
150
 
        return tree_a, tree_b
151
 
 
152
 
    def test_clone_partial(self):
153
 
        """Copy only part of the history of a branch."""
154
 
        # TODO: RBC 20060208 test with a revision not on revision-history.
155
 
        #       what should that behaviour be ? Emailed the list.
156
 
        # First, make a branch with two commits.
157
 
        wt_a = self.make_branch_and_tree('a')
158
 
        self.build_tree(['a/one'])
159
 
        wt_a.add(['one'])
160
 
        wt_a.commit('commit one', rev_id='1')
161
 
        self.build_tree(['a/two'])
162
 
        wt_a.add(['two'])
163
 
        wt_a.commit('commit two', rev_id='2')
164
 
        # Now make a copy of the repository.
165
 
        repo_b = self.make_repository('b')
166
 
        wt_a.branch.repository.copy_content_into(repo_b)
167
 
        # wt_a might be a lightweight checkout, so get a hold of the actual
168
 
        # branch (because you can't do a partial clone of a lightweight
169
 
        # checkout).
170
 
        branch = wt_a.branch.bzrdir.open_branch()
171
 
        # Then make a branch where the new repository is, but specify a revision
172
 
        # ID.  The new branch's history will stop at the specified revision.
173
 
        br_b = branch.clone(repo_b.bzrdir, revision_id='1')
174
 
        self.assertEqual('1', br_b.last_revision())
175
 
 
176
 
    def get_parented_branch(self):
177
 
        wt_a = self.make_branch_and_tree('a')
178
 
        self.build_tree(['a/one'])
179
 
        wt_a.add(['one'])
180
 
        wt_a.commit('commit one', rev_id='1')
181
 
 
182
 
        branch_b = wt_a.branch.bzrdir.sprout('b', revision_id='1').open_branch()
183
 
        self.assertEqual(wt_a.branch.base, branch_b.get_parent())
184
 
        return branch_b
185
 
 
186
 
    def test_clone_branch_nickname(self):
187
 
        # test the nick name is preserved always
188
 
        raise TestSkipped('XXX branch cloning is not yet tested.')
189
 
 
190
 
    def test_clone_branch_parent(self):
191
 
        # test the parent is preserved always
192
 
        branch_b = self.get_parented_branch()
193
 
        repo_c = self.make_repository('c')
194
 
        branch_b.repository.copy_content_into(repo_c)
195
 
        branch_c = branch_b.clone(repo_c.bzrdir)
196
 
        self.assertNotEqual(None, branch_c.get_parent())
197
 
        self.assertEqual(branch_b.get_parent(), branch_c.get_parent())
198
 
 
199
 
        # We can also set a specific parent, and it should be honored
200
 
        random_parent = 'http://bazaar-vcs.org/path/to/branch'
201
 
        branch_b.set_parent(random_parent)
202
 
        repo_d = self.make_repository('d')
203
 
        branch_b.repository.copy_content_into(repo_d)
204
 
        branch_d = branch_b.clone(repo_d.bzrdir)
205
 
        self.assertEqual(random_parent, branch_d.get_parent())
206
 
 
207
 
    def test_submit_branch(self):
208
 
        """Submit location can be queried and set"""
209
 
        branch = self.make_branch('branch')
210
 
        self.assertEqual(branch.get_submit_branch(), None)
211
 
        branch.set_submit_branch('sftp://example.com')
212
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.com')
213
 
        branch.set_submit_branch('sftp://example.net')
214
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.net')
215
 
        
216
 
    def test_public_branch(self):
217
 
        """public location can be queried and set"""
218
 
        branch = self.make_branch('branch')
219
 
        self.assertEqual(branch.get_public_branch(), None)
220
 
        branch.set_public_branch('sftp://example.com')
221
 
        self.assertEqual(branch.get_public_branch(), 'sftp://example.com')
222
 
        branch.set_public_branch('sftp://example.net')
223
 
        self.assertEqual(branch.get_public_branch(), 'sftp://example.net')
224
 
        branch.set_public_branch(None)
225
 
        self.assertEqual(branch.get_public_branch(), None)
226
 
 
227
 
    def test_record_initial_ghost(self):
228
 
        """Branches should support having ghosts."""
229
 
        wt = self.make_branch_and_tree('.')
230
 
        wt.set_parent_ids(['non:existent@rev--ision--0--2'],
231
 
            allow_leftmost_as_ghost=True)
232
 
        self.assertEqual(['non:existent@rev--ision--0--2'],
233
 
            wt.get_parent_ids())
234
 
        rev_id = wt.commit('commit against a ghost first parent.')
235
 
        rev = wt.branch.repository.get_revision(rev_id)
236
 
        self.assertEqual(rev.parent_ids, ['non:existent@rev--ision--0--2'])
237
 
        # parent_sha1s is not populated now, WTF. rbc 20051003
238
 
        self.assertEqual(len(rev.parent_sha1s), 0)
239
 
 
240
 
    def test_record_two_ghosts(self):
241
 
        """Recording with all ghosts works."""
242
 
        wt = self.make_branch_and_tree('.')
243
 
        wt.set_parent_ids([
244
 
                'foo@azkhazan-123123-abcabc',
245
 
                'wibble@fofof--20050401--1928390812',
246
 
            ],
247
 
            allow_leftmost_as_ghost=True)
248
 
        rev_id = wt.commit("commit from ghost base with one merge")
249
 
        # the revision should have been committed with two parents
250
 
        rev = wt.branch.repository.get_revision(rev_id)
251
 
        self.assertEqual(['foo@azkhazan-123123-abcabc',
252
 
            'wibble@fofof--20050401--1928390812'],
253
 
            rev.parent_ids)
254
 
 
255
 
    def test_bad_revision(self):
256
 
        self.assertRaises(errors.InvalidRevisionId,
257
 
                          self.get_branch().repository.get_revision,
258
 
                          None)
259
 
 
260
 
# TODO 20051003 RBC:
261
 
# compare the gpg-to-sign info for a commit with a ghost and 
262
 
#     an identical tree without a ghost
263
 
# fetch missing should rewrite the TOC of weaves to list newly available parents.
264
 
        
265
 
    def test_sign_existing_revision(self):
266
 
        wt = self.make_branch_and_tree('.')
267
 
        branch = wt.branch
268
 
        wt.commit("base", allow_pointless=True, rev_id='A')
269
 
        from bzrlib.testament import Testament
270
 
        strategy = gpg.LoopbackGPGStrategy(None)
271
 
        branch.repository.lock_write()
272
 
        branch.repository.start_write_group()
273
 
        branch.repository.sign_revision('A', strategy)
274
 
        branch.repository.commit_write_group()
275
 
        branch.repository.unlock()
276
 
        self.assertEqual('-----BEGIN PSEUDO-SIGNED CONTENT-----\n' +
277
 
                         Testament.from_revision(branch.repository,
278
 
                         'A').as_short_text() +
279
 
                         '-----END PSEUDO-SIGNED CONTENT-----\n',
280
 
                         branch.repository.get_signature_text('A'))
281
 
 
282
 
    def test_store_signature(self):
283
 
        wt = self.make_branch_and_tree('.')
284
 
        branch = wt.branch
285
 
        branch.lock_write()
286
 
        try:
287
 
            branch.repository.start_write_group()
288
 
            try:
289
 
                branch.repository.store_revision_signature(
290
 
                    gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
291
 
            except:
292
 
                branch.repository.abort_write_group()
293
 
                raise
294
 
            else:
295
 
                branch.repository.commit_write_group()
296
 
        finally:
297
 
            branch.unlock()
298
 
        # A signature without a revision should not be accessible.
299
 
        self.assertRaises(errors.NoSuchRevision,
300
 
                          branch.repository.has_signature_for_revision_id,
301
 
                          'A')
302
 
        wt.commit("base", allow_pointless=True, rev_id='A')
303
 
        self.assertEqual('-----BEGIN PSEUDO-SIGNED CONTENT-----\n'
304
 
                         'FOO-----END PSEUDO-SIGNED CONTENT-----\n',
305
 
                         branch.repository.get_signature_text('A'))
306
 
 
307
 
    def test_branch_keeps_signatures(self):
308
 
        wt = self.make_branch_and_tree('source')
309
 
        wt.commit('A', allow_pointless=True, rev_id='A')
310
 
        repo = wt.branch.repository
311
 
        repo.lock_write()
312
 
        repo.start_write_group()
313
 
        repo.sign_revision('A', gpg.LoopbackGPGStrategy(None))
314
 
        repo.commit_write_group()
315
 
        repo.unlock()
316
 
        #FIXME: clone should work to urls,
317
 
        # wt.clone should work to disks.
318
 
        self.build_tree(['target/'])
319
 
        d2 = repo.bzrdir.clone(urlutils.local_path_to_url('target'))
320
 
        self.assertEqual(repo.get_signature_text('A'),
321
 
                         d2.open_repository().get_signature_text('A'))
322
 
 
323
 
    def test_missing_revisions(self):
324
 
        t1 = self.make_branch_and_tree('b1')
325
 
        rev1 = t1.commit('one')
326
 
        t2 = t1.bzrdir.sprout('b2').open_workingtree()
327
 
        rev2 = t1.commit('two')
328
 
        rev3 = t1.commit('three')
329
 
 
330
 
        self.assertEqual([rev2, rev3],
331
 
            self.applyDeprecated(deprecated_in((1, 6, 0)),
332
 
            t2.branch.missing_revisions, t1.branch))
333
 
 
334
 
        self.assertEqual([],
335
 
            self.applyDeprecated(deprecated_in((1, 6, 0)),
336
 
            t2.branch.missing_revisions, t1.branch, stop_revision=1))
337
 
        self.assertEqual([rev2],
338
 
            self.applyDeprecated(deprecated_in((1, 6, 0)),
339
 
            t2.branch.missing_revisions, t1.branch, stop_revision=2))
340
 
        self.assertEqual([rev2, rev3],
341
 
            self.applyDeprecated(deprecated_in((1, 6, 0)),
342
 
            t2.branch.missing_revisions, t1.branch, stop_revision=3))
343
 
 
344
 
        self.assertRaises(errors.NoSuchRevision,
345
 
            self.applyDeprecated, deprecated_in((1, 6, 0)),
346
 
            t2.branch.missing_revisions, t1.branch, stop_revision=4)
347
 
 
348
 
        rev4 = t2.commit('four')
349
 
        self.assertRaises(errors.DivergedBranches,
350
 
            self.applyDeprecated, deprecated_in((1, 6, 0)),
351
 
            t2.branch.missing_revisions, t1.branch)
352
 
 
353
 
    def test_nicks(self):
354
 
        """Test explicit and implicit branch nicknames.
355
 
        
356
 
        Nicknames are implicitly the name of the branch's directory, unless an
357
 
        explicit nickname is set.  That is, an explicit nickname always
358
 
        overrides the implicit one.
359
 
        """
360
 
        t = get_transport(self.get_url())
361
 
        branch = self.make_branch('bzr.dev')
362
 
        # The nick will be 'bzr.dev', because there is no explicit nick set.
363
 
        self.assertEqual(branch.nick, 'bzr.dev')
364
 
        # Move the branch to a different directory, 'bzr.ab'.  Now that branch
365
 
        # will report its nick as 'bzr.ab'.
366
 
        t.move('bzr.dev', 'bzr.ab')
367
 
        branch = Branch.open(self.get_url('bzr.ab'))
368
 
        self.assertEqual(branch.nick, 'bzr.ab')
369
 
        # Set the branch nick explicitly.  This will ensure there's a branch
370
 
        # config file in the branch.
371
 
        branch.nick = "Aaron's branch"
372
 
        if not isinstance(branch, remote.RemoteBranch):
373
 
            self.failUnless(branch._transport.has("branch.conf"))
374
 
        # Because the nick has been set explicitly, the nick is now always
375
 
        # "Aaron's branch", regardless of directory name.
376
 
        self.assertEqual(branch.nick, "Aaron's branch")
377
 
        t.move('bzr.ab', 'integration')
378
 
        branch = Branch.open(self.get_url('integration'))
379
 
        self.assertEqual(branch.nick, "Aaron's branch")
380
 
        branch.nick = u"\u1234"
381
 
        self.assertEqual(branch.nick, u"\u1234")
382
 
 
383
 
    def test_commit_nicks(self):
384
 
        """Nicknames are committed to the revision"""
385
 
        wt = self.make_branch_and_tree('bzr.dev')
386
 
        branch = wt.branch
387
 
        branch.nick = "My happy branch"
388
 
        wt.commit('My commit respect da nick.')
389
 
        committed = branch.repository.get_revision(branch.last_revision())
390
 
        self.assertEqual(committed.properties["branch-nick"],
391
 
                         "My happy branch")
392
 
 
393
 
    def test_create_open_branch_uses_repository(self):
394
 
        try:
395
 
            repo = self.make_repository('.', shared=True)
396
 
        except errors.IncompatibleFormat:
397
 
            return
398
 
        child_transport = repo.bzrdir.root_transport.clone('child')
399
 
        child_transport.mkdir('.')
400
 
        child_dir = self.bzrdir_format.initialize_on_transport(child_transport)
401
 
        try:
402
 
            child_branch = self.branch_format.initialize(child_dir)
403
 
        except errors.UninitializableFormat:
404
 
            # branch references are not default init'able.
405
 
            return
406
 
        self.assertEqual(repo.bzrdir.root_transport.base,
407
 
                         child_branch.repository.bzrdir.root_transport.base)
408
 
        child_branch = branch.Branch.open(self.get_url('child'))
409
 
        self.assertEqual(repo.bzrdir.root_transport.base,
410
 
                         child_branch.repository.bzrdir.root_transport.base)
411
 
 
412
 
    def test_format_description(self):
413
 
        tree = self.make_branch_and_tree('tree')
414
 
        text = tree.branch._format.get_format_description()
415
 
        self.failUnless(len(text))
416
 
 
417
 
    def test_get_commit_builder(self):
418
 
        branch = self.make_branch(".")
419
 
        branch.lock_write()
420
 
        builder = branch.get_commit_builder([])
421
 
        self.assertIsInstance(builder, repository.CommitBuilder)
422
 
        branch.repository.commit_write_group()
423
 
        branch.unlock()
424
 
 
425
 
    def test_generate_revision_history(self):
426
 
        """Create a fake revision history easily."""
427
 
        tree = self.make_branch_and_tree('.')
428
 
        rev1 = tree.commit('foo')
429
 
        orig_history = tree.branch.revision_history()
430
 
        rev2 = tree.commit('bar', allow_pointless=True)
431
 
        tree.branch.generate_revision_history(rev1)
432
 
        self.assertEqual(orig_history, tree.branch.revision_history())
433
 
 
434
 
    def test_generate_revision_history_NULL_REVISION(self):
435
 
        tree = self.make_branch_and_tree('.')
436
 
        rev1 = tree.commit('foo')
437
 
        tree.branch.generate_revision_history(bzrlib.revision.NULL_REVISION)
438
 
        self.assertEqual([], tree.branch.revision_history())
439
 
 
440
 
    def test_create_checkout(self):
441
 
        tree_a = self.make_branch_and_tree('a')
442
 
        branch_a = tree_a.branch
443
 
        checkout_b = branch_a.create_checkout('b')
444
 
        self.assertEqual('null:', checkout_b.last_revision())
445
 
        checkout_b.commit('rev1', rev_id='rev1')
446
 
        self.assertEqual('rev1', branch_a.last_revision())
447
 
        self.assertNotEqual(checkout_b.branch.base, branch_a.base)
448
 
 
449
 
        checkout_c = branch_a.create_checkout('c', lightweight=True)
450
 
        self.assertEqual('rev1', checkout_c.last_revision())
451
 
        checkout_c.commit('rev2', rev_id='rev2')
452
 
        self.assertEqual('rev2', branch_a.last_revision())
453
 
        self.assertEqual(checkout_c.branch.base, branch_a.base)
454
 
 
455
 
        os.mkdir('d')
456
 
        checkout_d = branch_a.create_checkout('d', lightweight=True)
457
 
        self.assertEqual('rev2', checkout_d.last_revision())
458
 
        os.mkdir('e')
459
 
        checkout_e = branch_a.create_checkout('e')
460
 
        self.assertEqual('rev2', checkout_e.last_revision())
461
 
 
462
 
    def test_create_anonymous_lightweight_checkout(self):
463
 
        """A lightweight checkout from a readonly branch should succeed."""
464
 
        tree_a = self.make_branch_and_tree('a')
465
 
        rev_id = tree_a.commit('put some content in the branch')
466
 
        # open the branch via a readonly transport
467
 
        source_branch = bzrlib.branch.Branch.open(self.get_readonly_url('a'))
468
 
        # sanity check that the test will be valid
469
 
        self.assertRaises((errors.LockError, errors.TransportNotPossible),
470
 
            source_branch.lock_write)
471
 
        checkout = source_branch.create_checkout('c', lightweight=True)
472
 
        self.assertEqual(rev_id, checkout.last_revision())
473
 
 
474
 
    def test_create_anonymous_heavyweight_checkout(self):
475
 
        """A regular checkout from a readonly branch should succeed."""
476
 
        tree_a = self.make_branch_and_tree('a')
477
 
        rev_id = tree_a.commit('put some content in the branch')
478
 
        # open the branch via a readonly transport
479
 
        source_branch = bzrlib.branch.Branch.open(self.get_readonly_url('a'))
480
 
        # sanity check that the test will be valid
481
 
        self.assertRaises((errors.LockError, errors.TransportNotPossible),
482
 
            source_branch.lock_write)
483
 
        checkout = source_branch.create_checkout('c')
484
 
        self.assertEqual(rev_id, checkout.last_revision())
485
 
 
486
 
    def test_set_revision_history(self):
487
 
        tree = self.make_branch_and_tree('a')
488
 
        tree.commit('a commit', rev_id='rev1')
489
 
        br = tree.branch
490
 
        br.set_revision_history(["rev1"])
491
 
        self.assertEquals(br.revision_history(), ["rev1"])
492
 
        br.set_revision_history([])
493
 
        self.assertEquals(br.revision_history(), [])
494
 
 
495
 
 
496
 
class ChrootedTests(TestCaseWithBranch):
497
 
    """A support class that provides readonly urls outside the local namespace.
498
 
 
499
 
    This is done by checking if self.transport_server is a MemoryServer. if it
500
 
    is then we are chrooted already, if it is not then an HttpServer is used
501
 
    for readonly urls.
502
 
    """
503
 
 
504
 
    def setUp(self):
505
 
        super(ChrootedTests, self).setUp()
506
 
        if not self.vfs_transport_factory == MemoryServer:
507
 
            self.transport_readonly_server = HttpServer
508
 
 
509
 
    def test_open_containing(self):
510
 
        self.assertRaises(NotBranchError, Branch.open_containing,
511
 
                          self.get_readonly_url(''))
512
 
        self.assertRaises(NotBranchError, Branch.open_containing,
513
 
                          self.get_readonly_url('g/p/q'))
514
 
        branch = self.make_branch('.')
515
 
        branch, relpath = Branch.open_containing(self.get_readonly_url(''))
516
 
        self.assertEqual('', relpath)
517
 
        branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
518
 
        self.assertEqual('g/p/q', relpath)
519
 
 
520
 
 
521
 
class InstrumentedTransaction(object):
522
 
 
523
 
    def finish(self):
524
 
        self.calls.append('finish')
525
 
 
526
 
    def __init__(self):
527
 
        self.calls = []
528
 
 
529
 
 
530
 
class TestDecorator(object):
531
 
 
532
 
    def __init__(self):
533
 
        self._calls = []
534
 
 
535
 
    def lock_read(self):
536
 
        self._calls.append('lr')
537
 
 
538
 
    def lock_write(self):
539
 
        self._calls.append('lw')
540
 
 
541
 
    def unlock(self):
542
 
        self._calls.append('ul')
543
 
 
544
 
    @needs_read_lock
545
 
    def do_with_read(self):
546
 
        return 1
547
 
 
548
 
    @needs_read_lock
549
 
    def except_with_read(self):
550
 
        raise RuntimeError
551
 
 
552
 
    @needs_write_lock
553
 
    def do_with_write(self):
554
 
        return 2
555
 
 
556
 
    @needs_write_lock
557
 
    def except_with_write(self):
558
 
        raise RuntimeError
559
 
 
560
 
 
561
 
class TestDecorators(TestCase):
562
 
 
563
 
    def test_needs_read_lock(self):
564
 
        branch = TestDecorator()
565
 
        self.assertEqual(1, branch.do_with_read())
566
 
        self.assertEqual(['lr', 'ul'], branch._calls)
567
 
 
568
 
    def test_excepts_in_read_lock(self):
569
 
        branch = TestDecorator()
570
 
        self.assertRaises(RuntimeError, branch.except_with_read)
571
 
        self.assertEqual(['lr', 'ul'], branch._calls)
572
 
 
573
 
    def test_needs_write_lock(self):
574
 
        branch = TestDecorator()
575
 
        self.assertEqual(2, branch.do_with_write())
576
 
        self.assertEqual(['lw', 'ul'], branch._calls)
577
 
 
578
 
    def test_excepts_in_write_lock(self):
579
 
        branch = TestDecorator()
580
 
        self.assertRaises(RuntimeError, branch.except_with_write)
581
 
        self.assertEqual(['lw', 'ul'], branch._calls)
582
 
 
583
 
 
584
 
class TestBranchPushLocations(TestCaseWithBranch):
585
 
 
586
 
    def test_get_push_location_unset(self):
587
 
        self.assertEqual(None, self.get_branch().get_push_location())
588
 
 
589
 
    def test_get_push_location_exact(self):
590
 
        from bzrlib.config import (locations_config_filename,
591
 
                                   ensure_config_dir_exists)
592
 
        ensure_config_dir_exists()
593
 
        fn = locations_config_filename()
594
 
        open(fn, 'wt').write(("[%s]\n"
595
 
                                  "push_location=foo\n" %
596
 
                                  self.get_branch().base[:-1]))
597
 
        self.assertEqual("foo", self.get_branch().get_push_location())
598
 
 
599
 
    def test_set_push_location(self):
600
 
        branch = self.get_branch()
601
 
        branch.set_push_location('foo')
602
 
        self.assertEqual('foo', branch.get_push_location())
603
 
 
604
 
 
605
 
class TestFormat(TestCaseWithBranch):
606
 
    """Tests for the format itself."""
607
 
 
608
 
    def test_get_reference(self):
609
 
        """get_reference on all regular branches should return None."""
610
 
        if not self.branch_format.is_supported():
611
 
            # unsupported formats are not loopback testable
612
 
            # because the default open will not open them and
613
 
            # they may not be initializable.
614
 
            return
615
 
        made_branch = self.make_branch('.')
616
 
        self.assertEqual(None,
617
 
            made_branch._format.get_reference(made_branch.bzrdir))
618
 
 
619
 
    def test_set_reference(self):
620
 
        """set_reference on all regular branches should be callable."""
621
 
        if not self.branch_format.is_supported():
622
 
            # unsupported formats are not loopback testable
623
 
            # because the default open will not open them and
624
 
            # they may not be initializable.
625
 
            return
626
 
        this_branch = self.make_branch('this')
627
 
        other_branch = self.make_branch('other')
628
 
        try:
629
 
            this_branch._format.set_reference(this_branch.bzrdir, other_branch)
630
 
        except NotImplementedError:
631
 
            # that's ok
632
 
            pass
633
 
        else:
634
 
            ref = this_branch._format.get_reference(this_branch.bzrdir)
635
 
            self.assertEqual(ref, other_branch.base)
636
 
 
637
 
    def test_format_initialize_find_open(self):
638
 
        # loopback test to check the current format initializes to itself.
639
 
        if not self.branch_format.is_supported():
640
 
            # unsupported formats are not loopback testable
641
 
            # because the default open will not open them and
642
 
            # they may not be initializable.
643
 
            return
644
 
        # supported formats must be able to init and open
645
 
        t = get_transport(self.get_url())
646
 
        readonly_t = get_transport(self.get_readonly_url())
647
 
        made_branch = self.make_branch('.')
648
 
        self.failUnless(isinstance(made_branch, branch.Branch))
649
 
 
650
 
        # find it via bzrdir opening:
651
 
        opened_control = bzrdir.BzrDir.open(readonly_t.base)
652
 
        direct_opened_branch = opened_control.open_branch()
653
 
        self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
654
 
        self.assertEqual(opened_control, direct_opened_branch.bzrdir)
655
 
        self.failUnless(isinstance(direct_opened_branch._format,
656
 
                        self.branch_format.__class__))
657
 
 
658
 
        # find it via Branch.open
659
 
        opened_branch = branch.Branch.open(readonly_t.base)
660
 
        self.failUnless(isinstance(opened_branch, made_branch.__class__))
661
 
        self.assertEqual(made_branch._format.__class__,
662
 
                         opened_branch._format.__class__)
663
 
        # if it has a unique id string, can we probe for it ?
664
 
        try:
665
 
            self.branch_format.get_format_string()
666
 
        except NotImplementedError:
667
 
            return
668
 
        self.assertEqual(self.branch_format,
669
 
                         opened_control.find_branch_format())
670
 
 
671
 
 
672
 
class TestBound(TestCaseWithBranch):
673
 
 
674
 
    def test_bind_unbind(self):
675
 
        branch = self.make_branch('1')
676
 
        branch2 = self.make_branch('2')
677
 
        try:
678
 
            branch.bind(branch2)
679
 
        except errors.UpgradeRequired:
680
 
            raise tests.TestNotApplicable('Format does not support binding')
681
 
        self.assertTrue(branch.unbind())
682
 
        self.assertFalse(branch.unbind())
683
 
        self.assertIs(None, branch.get_bound_location())
684
 
 
685
 
    def test_old_bound_location(self):
686
 
        branch = self.make_branch('branch1')
687
 
        try:
688
 
            self.assertIs(None, branch.get_old_bound_location())
689
 
        except errors.UpgradeRequired:
690
 
            raise tests.TestNotApplicable(
691
 
                    'Format does not store old bound locations')
692
 
        branch2 = self.make_branch('branch2')
693
 
        branch.bind(branch2)
694
 
        self.assertIs(None, branch.get_old_bound_location())
695
 
        branch.unbind()
696
 
        self.assertContainsRe(branch.get_old_bound_location(), '\/branch2\/$')
697
 
 
698
 
    def test_bind_diverged(self):
699
 
        tree_a = self.make_branch_and_tree('tree_a')
700
 
        tree_a.commit('rev1a')
701
 
        tree_b = tree_a.bzrdir.sprout('tree_b').open_workingtree()
702
 
        tree_a.commit('rev2a')
703
 
        tree_b.commit('rev2b')
704
 
        try:
705
 
            tree_b.branch.bind(tree_a.branch)
706
 
        except errors.UpgradeRequired:
707
 
            raise tests.TestNotApplicable('Format does not support binding')
708
 
 
709
 
 
710
 
class TestStrict(TestCaseWithBranch):
711
 
 
712
 
    def test_strict_history(self):
713
 
        tree1 = self.make_branch_and_tree('tree1')
714
 
        try:
715
 
            tree1.branch.set_append_revisions_only(True)
716
 
        except errors.UpgradeRequired:
717
 
            raise TestSkipped('Format does not support strict history')
718
 
        tree1.commit('empty commit')
719
 
        tree2 = tree1.bzrdir.sprout('tree2').open_workingtree()
720
 
        tree2.commit('empty commit 2')
721
 
        tree1.pull(tree2.branch)
722
 
        tree1.commit('empty commit 3')
723
 
        tree2.commit('empty commit 4')
724
 
        self.assertRaises(errors.DivergedBranches, tree1.pull, tree2.branch)
725
 
        tree2.merge_from_branch(tree1.branch)
726
 
        tree2.commit('empty commit 5')
727
 
        self.assertRaises(errors.AppendRevisionsOnlyViolation, tree1.pull,
728
 
                          tree2.branch)
729
 
        tree3 = tree1.bzrdir.sprout('tree3').open_workingtree()
730
 
        tree3.merge_from_branch(tree2.branch)
731
 
        tree3.commit('empty commit 6')
732
 
        tree2.pull(tree3.branch)
 
17
from bzrlib.selftest import InTempDir
 
18
 
 
19
 
 
20
 
 
21
class TestAppendRevisions(InTempDir):
 
22
    """Test appending more than one revision"""
 
23
    def runTest(self):
 
24
        from bzrlib.branch import Branch
 
25
        br = Branch(".", init=True)
 
26
        br.append_revision("rev1")
 
27
        self.assertEquals(br.revision_history(), ["rev1",])
 
28
        br.append_revision("rev2", "rev3")
 
29
        self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
 
30
        
 
31
 
 
32
 
 
33
TEST_CLASSES = [
 
34
    TestAppendRevisions,
 
35
    ]