~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

  • Committer: Martin Pool
  • Date: 2005-09-30 05:56:05 UTC
  • mto: (1185.14.2)
  • mto: This revision was merged to the branch mainline in revision 1396.
  • Revision ID: mbp@sourcefrog.net-20050930055605-a2c534529b392a7d
- fix upgrade for transport changes

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 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
17
import os
20
 
import sys
21
 
 
22
 
from bzrlib import (
23
 
    branch,
24
 
    bzrdir,
25
 
    errors,
26
 
    gpg,
27
 
    urlutils,
28
 
    transactions,
29
 
    repository,
30
 
    )
31
 
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
32
 
from bzrlib.delta import TreeDelta
33
 
from bzrlib.errors import (FileExists,
34
 
                           NoSuchRevision,
35
 
                           NoSuchFile,
36
 
                           UninitializableFormat,
37
 
                           NotBranchError,
38
 
                           )
39
 
from bzrlib.osutils import getcwd
40
 
import bzrlib.revision
41
 
from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
42
 
from bzrlib.tests.bzrdir_implementations.test_bzrdir import TestCaseWithBzrDir
43
 
from bzrlib.tests.HttpServer import HttpServer
 
18
from bzrlib.branch import Branch
 
19
from bzrlib.clone import copy_branch
 
20
from bzrlib.commit import commit
 
21
from bzrlib.errors import NoSuchRevision, UnlistableBranch
 
22
from bzrlib.selftest import TestCaseInTempDir
44
23
from bzrlib.trace import mutter
45
 
from bzrlib.transport import get_transport
46
 
from bzrlib.transport.memory import MemoryServer
47
 
from bzrlib.upgrade import upgrade
48
 
from bzrlib.workingtree import WorkingTree
49
 
 
50
 
 
51
 
# TODO: Make a branch using basis branch, and check that it 
52
 
# doesn't request any files that could have been avoided, by 
53
 
# hooking into the Transport.
54
 
 
55
 
 
56
 
class TestCaseWithBranch(TestCaseWithBzrDir):
57
 
 
58
 
    def setUp(self):
59
 
        super(TestCaseWithBranch, self).setUp()
60
 
        self.branch = None
61
 
 
62
 
    def get_branch(self):
63
 
        if self.branch is None:
64
 
            self.branch = self.make_branch('')
65
 
        return self.branch
66
 
 
67
 
    def make_branch(self, relpath, format=None):
68
 
        repo = self.make_repository(relpath, format=format)
69
 
        # fixme RBC 20060210 this isnt necessarily a fixable thing,
70
 
        # Skipped is the wrong exception to raise.
71
 
        try:
72
 
            return self.branch_format.initialize(repo.bzrdir)
73
 
        except errors.UninitializableFormat:
74
 
            raise TestSkipped('Uninitializable branch format')
75
 
 
76
 
    def make_repository(self, relpath, shared=False, format=None):
77
 
        made_control = self.make_bzrdir(relpath, format=format)
78
 
        return made_control.create_repository(shared=shared)
79
 
 
80
 
 
81
 
class TestBranch(TestCaseWithBranch):
 
24
 
 
25
class TestBranch(TestCaseInTempDir):
82
26
 
83
27
    def test_append_revisions(self):
84
28
        """Test appending more than one revision"""
85
 
        wt = self.make_branch_and_tree('tree')
86
 
        wt.commit('f', rev_id='rev1')
87
 
        wt.commit('f', rev_id='rev2')
88
 
        wt.commit('f', rev_id='rev3')
89
 
 
90
 
        br = self.get_branch()
91
 
        br.fetch(wt.branch)
 
29
        br = Branch.initialize(".")
92
30
        br.append_revision("rev1")
93
31
        self.assertEquals(br.revision_history(), ["rev1",])
94
32
        br.append_revision("rev2", "rev3")
95
33
        self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
96
 
        self.assertRaises(errors.ReservedId, br.append_revision, 'current:')
97
 
 
98
 
    def test_revision_ids_are_utf8(self):
99
 
        wt = self.make_branch_and_tree('tree')
100
 
        wt.commit('f', rev_id='rev1')
101
 
        wt.commit('f', rev_id='rev2')
102
 
        wt.commit('f', rev_id='rev3')
103
 
 
104
 
        br = self.get_branch()
105
 
        br.fetch(wt.branch)
106
 
        br.set_revision_history(['rev1', 'rev2', 'rev3'])
107
 
        rh = br.revision_history()
108
 
        self.assertEqual(['rev1', 'rev2', 'rev3'], rh)
109
 
        for revision_id in rh:
110
 
            self.assertIsInstance(revision_id, str)
111
 
        last = br.last_revision()
112
 
        self.assertEqual('rev3', last)
113
 
        self.assertIsInstance(last, str)
114
 
        revno, last = br.last_revision_info()
115
 
        self.assertEqual(3, revno)
116
 
        self.assertEqual('rev3', last)
117
 
        self.assertIsInstance(last, str)
 
34
 
 
35
 
 
36
class TestFetch(TestCaseInTempDir):
118
37
 
119
38
    def test_fetch_revisions(self):
120
39
        """Test fetch-revision operation."""
121
 
        get_transport(self.get_url()).mkdir('b1')
122
 
        get_transport(self.get_url()).mkdir('b2')
123
 
        wt = self.make_branch_and_tree('b1')
124
 
        b1 = wt.branch
125
 
        b2 = self.make_branch('b2')
126
 
        file('b1/foo', 'w').write('hello')
127
 
        wt.add(['foo'], ['foo-id'])
128
 
        wt.commit('lala!', rev_id='revision-1', allow_pointless=False)
 
40
        from bzrlib.fetch import Fetcher
 
41
        os.mkdir('b1')
 
42
        os.mkdir('b2')
 
43
        b1 = Branch.initialize('b1')
 
44
        b2 = Branch.initialize('b2')
 
45
        file(os.sep.join(['b1', 'foo']), 'w').write('hello')
 
46
        b1.add(['foo'], ['foo-id'])
 
47
        b1.commit('lala!', rev_id='revision-1', allow_pointless=False)
129
48
 
130
49
        mutter('start fetch')
131
 
        self.assertEqual((1, []), b2.fetch(b1))
132
 
 
133
 
        rev = b2.repository.get_revision('revision-1')
134
 
        tree = b2.repository.revision_tree('revision-1')
135
 
        self.assertEqual(tree.get_file_text('foo-id'), 'hello')
136
 
 
137
 
    def test_get_revision_delta(self):
138
 
        tree_a = self.make_branch_and_tree('a')
139
 
        self.build_tree(['a/foo'])
140
 
        tree_a.add('foo', 'file1')
141
 
        tree_a.commit('rev1', rev_id='rev1')
142
 
        self.build_tree(['a/vla'])
143
 
        tree_a.add('vla', 'file2')
144
 
        tree_a.commit('rev2', rev_id='rev2')
145
 
 
146
 
        delta = tree_a.branch.get_revision_delta(1)
147
 
        self.assertIsInstance(delta, TreeDelta)
148
 
        self.assertEqual([('foo', 'file1', 'file')], delta.added)
149
 
        delta = tree_a.branch.get_revision_delta(2)
150
 
        self.assertIsInstance(delta, TreeDelta)
151
 
        self.assertEqual([('vla', 'file2', 'file')], delta.added)
152
 
 
153
 
    def get_unbalanced_tree_pair(self):
154
 
        """Return two branches, a and b, with one file in a."""
155
 
        get_transport(self.get_url()).mkdir('a')
156
 
        tree_a = self.make_branch_and_tree('a')
 
50
        f = Fetcher(from_branch=b1, to_branch=b2)
 
51
        eq = self.assertEquals
 
52
        eq(f.count_copied, 1)
 
53
        eq(f.last_revision, 'revision-1')
 
54
 
 
55
        rev = b2.get_revision('revision-1')
 
56
        tree = b2.revision_tree('revision-1')
 
57
        eq(tree.get_file_text('foo-id'), 'hello')
 
58
 
 
59
    def test_push_stores(self):
 
60
        """Copy the stores from one branch to another"""
 
61
        os.mkdir('a')
 
62
        br_a = Branch.initialize("a")
157
63
        file('a/b', 'wb').write('b')
158
 
        tree_a.add('b')
159
 
        tree_a.commit("silly commit", rev_id='A')
160
 
 
161
 
        get_transport(self.get_url()).mkdir('b')
162
 
        tree_b = self.make_branch_and_tree('b')
163
 
        return tree_a, tree_b
164
 
 
165
 
    def get_balanced_branch_pair(self):
166
 
        """Returns br_a, br_b as with one commit in a, and b has a's stores."""
167
 
        tree_a, tree_b = self.get_unbalanced_tree_pair()
168
 
        tree_b.branch.repository.fetch(tree_a.branch.repository)
169
 
        return tree_a, tree_b
170
 
 
171
 
    def test_clone_branch(self):
 
64
        br_a.add('b')
 
65
        commit(br_a, "silly commit")
 
66
 
 
67
        os.mkdir('b')
 
68
        br_b = Branch.initialize("b")
 
69
        self.assertRaises(NoSuchRevision, br_b.get_revision, 
 
70
                          br_a.revision_history()[0])
 
71
        br_a.push_stores(br_b)
 
72
        rev = br_b.get_revision(br_a.revision_history()[0])
 
73
        tree = br_b.revision_tree(br_a.revision_history()[0])
 
74
        for file_id in tree:
 
75
            if tree.inventory[file_id].kind == "file":
 
76
                tree.get_file(file_id).read()
 
77
        return br_a, br_b
 
78
 
 
79
    def test_copy_branch(self):
172
80
        """Copy the stores from one branch to another"""
173
 
        tree_a, tree_b = self.get_balanced_branch_pair()
174
 
        tree_b.commit("silly commit")
 
81
        br_a, br_b = self.test_push_stores()
 
82
        commit(br_b, "silly commit")
175
83
        os.mkdir('c')
176
 
        # this fails to test that the history from a was not used.
177
 
        dir_c = tree_a.bzrdir.clone('c', basis=tree_b.bzrdir)
178
 
        self.assertEqual(tree_a.branch.revision_history(),
179
 
                         dir_c.open_branch().revision_history())
180
 
 
181
 
    def test_clone_partial(self):
182
 
        """Copy only part of the history of a branch."""
183
 
        # TODO: RBC 20060208 test with a revision not on revision-history.
184
 
        #       what should that behaviour be ? Emailed the list.
185
 
        wt_a = self.make_branch_and_tree('a')
186
 
        self.build_tree(['a/one'])
187
 
        wt_a.add(['one'])
188
 
        wt_a.commit('commit one', rev_id='1')
189
 
        self.build_tree(['a/two'])
190
 
        wt_a.add(['two'])
191
 
        wt_a.commit('commit two', rev_id='2')
192
 
        repo_b = self.make_repository('b')
193
 
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
194
 
        br_b = wt_a.bzrdir.open_branch().clone(repo_b.bzrdir, revision_id='1')
195
 
        self.assertEqual('1', br_b.last_revision())
196
 
 
197
 
    def test_sprout_partial(self):
198
 
        # test sprouting with a prefix of the revision-history.
199
 
        # also needs not-on-revision-history behaviour defined.
200
 
        wt_a = self.make_branch_and_tree('a')
201
 
        self.build_tree(['a/one'])
202
 
        wt_a.add(['one'])
203
 
        wt_a.commit('commit one', rev_id='1')
204
 
        self.build_tree(['a/two'])
205
 
        wt_a.add(['two'])
206
 
        wt_a.commit('commit two', rev_id='2')
207
 
        repo_b = self.make_repository('b')
208
 
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
209
 
        br_b = wt_a.bzrdir.open_branch().sprout(repo_b.bzrdir, revision_id='1')
210
 
        self.assertEqual('1', br_b.last_revision())
211
 
 
212
 
    def get_parented_branch(self):
213
 
        wt_a = self.make_branch_and_tree('a')
214
 
        self.build_tree(['a/one'])
215
 
        wt_a.add(['one'])
216
 
        wt_a.commit('commit one', rev_id='1')
217
 
 
218
 
        branch_b = wt_a.bzrdir.sprout('b', revision_id='1').open_branch()
219
 
        self.assertEqual(wt_a.branch.base, branch_b.get_parent())
220
 
        return branch_b
221
 
 
222
 
    def test_clone_branch_nickname(self):
223
 
        # test the nick name is preserved always
224
 
        raise TestSkipped('XXX branch cloning is not yet tested..')
225
 
 
226
 
    def test_clone_branch_parent(self):
227
 
        # test the parent is preserved always
228
 
        branch_b = self.get_parented_branch()
229
 
        repo_c = self.make_repository('c')
230
 
        branch_b.repository.copy_content_into(repo_c)
231
 
        branch_c = branch_b.clone(repo_c.bzrdir)
232
 
        self.assertNotEqual(None, branch_c.get_parent())
233
 
        self.assertEqual(branch_b.get_parent(), branch_c.get_parent())
234
 
 
235
 
        # We can also set a specific parent, and it should be honored
236
 
        random_parent = 'http://bazaar-vcs.org/path/to/branch'
237
 
        branch_b.set_parent(random_parent)
238
 
        repo_d = self.make_repository('d')
239
 
        branch_b.repository.copy_content_into(repo_d)
240
 
        branch_d = branch_b.clone(repo_d.bzrdir)
241
 
        self.assertEqual(random_parent, branch_d.get_parent())
242
 
 
243
 
    def test_copy_content_incomplete(self):
244
 
        tree = self.make_branch_and_tree('commit_tree')
245
 
        self.build_tree(['foo'], transport=tree.bzrdir.root_transport)
246
 
        tree.add('foo')
247
 
        tree.commit('revision 1', rev_id='1')
248
 
        source = self.make_branch_and_tree('source')
249
 
        # this gives us an incomplete repository
250
 
        tree.bzrdir.open_repository().copy_content_into(
251
 
            source.branch.repository)
252
 
        tree.commit('revision 2', rev_id='2', allow_pointless=True)
253
 
        tree.bzrdir.open_branch().copy_content_into(source.branch)
254
 
 
255
 
 
256
 
    def test_sprout_branch_nickname(self):
257
 
        # test the nick name is reset always
258
 
        raise TestSkipped('XXX branch sprouting is not yet tested..')
259
 
 
260
 
    def test_sprout_branch_parent(self):
261
 
        source = self.make_branch('source')
262
 
        target = source.bzrdir.sprout(self.get_url('target')).open_branch()
263
 
        self.assertEqual(source.bzrdir.root_transport.base, target.get_parent())
264
 
 
265
 
    def test_submit_branch(self):
266
 
        """Submit location can be queried and set"""
267
 
        branch = self.make_branch('branch')
268
 
        self.assertEqual(branch.get_submit_branch(), None)
269
 
        branch.set_submit_branch('sftp://example.com')
270
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.com')
271
 
        branch.set_submit_branch('sftp://example.net')
272
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.net')
273
 
        
274
 
    def test_record_initial_ghost(self):
275
 
        """Branches should support having ghosts."""
276
 
        wt = self.make_branch_and_tree('.')
277
 
        wt.set_parent_ids(['non:existent@rev--ision--0--2'],
278
 
            allow_leftmost_as_ghost=True)
279
 
        rev_id = wt.commit('commit against a ghost first parent.')
280
 
        rev = wt.branch.repository.get_revision(rev_id)
281
 
        self.assertEqual(rev.parent_ids, ['non:existent@rev--ision--0--2'])
282
 
        # parent_sha1s is not populated now, WTF. rbc 20051003
283
 
        self.assertEqual(len(rev.parent_sha1s), 0)
284
 
 
285
 
    def test_record_two_ghosts(self):
286
 
        """Recording with all ghosts works."""
287
 
        wt = self.make_branch_and_tree('.')
288
 
        wt.set_parent_ids([
289
 
                'foo@azkhazan-123123-abcabc',
290
 
                'wibble@fofof--20050401--1928390812',
291
 
            ],
292
 
            allow_leftmost_as_ghost=True)
293
 
        rev_id = wt.commit("commit from ghost base with one merge")
294
 
        # the revision should have been committed with two parents
295
 
        rev = wt.branch.repository.get_revision(rev_id)
296
 
        self.assertEqual(['foo@azkhazan-123123-abcabc',
297
 
            'wibble@fofof--20050401--1928390812'],
298
 
            rev.parent_ids)
299
 
 
300
 
    def test_bad_revision(self):
301
 
        self.assertRaises(errors.InvalidRevisionId,
302
 
                          self.get_branch().repository.get_revision,
303
 
                          None)
304
 
 
305
 
# TODO 20051003 RBC:
306
 
# compare the gpg-to-sign info for a commit with a ghost and 
307
 
#     an identical tree without a ghost
308
 
# fetch missing should rewrite the TOC of weaves to list newly available parents.
309
 
        
310
 
    def test_sign_existing_revision(self):
311
 
        wt = self.make_branch_and_tree('.')
312
 
        branch = wt.branch
313
 
        wt.commit("base", allow_pointless=True, rev_id='A')
314
 
        from bzrlib.testament import Testament
315
 
        strategy = gpg.LoopbackGPGStrategy(None)
316
 
        branch.repository.sign_revision('A', strategy)
317
 
        self.assertEqual(Testament.from_revision(branch.repository, 
318
 
                         'A').as_short_text(),
319
 
                         branch.repository.get_signature_text('A'))
320
 
 
321
 
    def test_store_signature(self):
322
 
        wt = self.make_branch_and_tree('.')
323
 
        branch = wt.branch
324
 
        branch.repository.store_revision_signature(
325
 
            gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
326
 
        self.assertRaises(errors.NoSuchRevision,
327
 
                          branch.repository.has_signature_for_revision_id,
328
 
                          'A')
329
 
        wt.commit("base", allow_pointless=True, rev_id='A')
330
 
        self.assertEqual('FOO', 
331
 
                         branch.repository.get_signature_text('A'))
332
 
 
333
 
    def test_branch_keeps_signatures(self):
334
 
        wt = self.make_branch_and_tree('source')
335
 
        wt.commit('A', allow_pointless=True, rev_id='A')
336
 
        wt.branch.repository.sign_revision('A',
337
 
            gpg.LoopbackGPGStrategy(None))
338
 
        #FIXME: clone should work to urls,
339
 
        # wt.clone should work to disks.
340
 
        self.build_tree(['target/'])
341
 
        d2 = wt.bzrdir.clone('target')
342
 
        self.assertEqual(wt.branch.repository.get_signature_text('A'),
343
 
                         d2.open_repository().get_signature_text('A'))
344
 
 
345
 
    def test_nicks(self):
346
 
        """Branch nicknames"""
347
 
        t = get_transport(self.get_url())
348
 
        t.mkdir('bzr.dev')
349
 
        branch = self.make_branch('bzr.dev')
350
 
        self.assertEqual(branch.nick, 'bzr.dev')
351
 
        t.move('bzr.dev', 'bzr.ab')
352
 
        branch = Branch.open(self.get_url('bzr.ab'))
353
 
        self.assertEqual(branch.nick, 'bzr.ab')
354
 
        branch.nick = "Aaron's branch"
355
 
        branch.nick = "Aaron's branch"
356
 
        self.failUnless(
357
 
            t.has(
358
 
                t.relpath(
359
 
                    branch.control_files.controlfilename("branch.conf")
360
 
                    )
361
 
                )
362
 
            )
363
 
        self.assertEqual(branch.nick, "Aaron's branch")
364
 
        t.move('bzr.ab', 'integration')
365
 
        branch = Branch.open(self.get_url('integration'))
366
 
        self.assertEqual(branch.nick, "Aaron's branch")
367
 
        branch.nick = u"\u1234"
368
 
        self.assertEqual(branch.nick, u"\u1234")
369
 
 
370
 
    def test_create_open_branch_uses_repository(self):
371
 
        try:
372
 
            repo = self.make_repository('.', shared=True)
373
 
        except errors.IncompatibleFormat:
374
 
            return
375
 
        repo.bzrdir.root_transport.mkdir('child')
376
 
        child_dir = self.bzrdir_format.initialize('child')
377
 
        try:
378
 
            child_branch = self.branch_format.initialize(child_dir)
379
 
        except errors.UninitializableFormat:
380
 
            # branch references are not default init'able.
381
 
            return
382
 
        self.assertEqual(repo.bzrdir.root_transport.base,
383
 
                         child_branch.repository.bzrdir.root_transport.base)
384
 
        child_branch = branch.Branch.open(self.get_url('child'))
385
 
        self.assertEqual(repo.bzrdir.root_transport.base,
386
 
                         child_branch.repository.bzrdir.root_transport.base)
387
 
 
388
 
    def test_format_description(self):
389
 
        tree = self.make_branch_and_tree('tree')
390
 
        text = tree.branch._format.get_format_description()
391
 
        self.failUnless(len(text))
392
 
 
393
 
    def test_check_branch_report_results(self):
394
 
        """Checking a branch produces results which can be printed"""
395
 
        branch = self.make_branch('.')
396
 
        result = branch.check()
397
 
        # reports results through logging
398
 
        result.report_results(verbose=True)
399
 
        result.report_results(verbose=False)
400
 
 
401
 
    def test_get_commit_builder(self):
402
 
        self.assertIsInstance(self.make_branch(".").get_commit_builder([]), 
403
 
            repository.CommitBuilder)
404
 
 
405
 
    def test_generate_revision_history(self):
406
 
        """Create a fake revision history easily."""
407
 
        tree = self.make_branch_and_tree('.')
408
 
        rev1 = tree.commit('foo')
409
 
        orig_history = tree.branch.revision_history()
410
 
        rev2 = tree.commit('bar', allow_pointless=True)
411
 
        tree.branch.generate_revision_history(rev1)
412
 
        self.assertEqual(orig_history, tree.branch.revision_history())
413
 
 
414
 
    def test_generate_revision_history_NULL_REVISION(self):
415
 
        tree = self.make_branch_and_tree('.')
416
 
        rev1 = tree.commit('foo')
417
 
        tree.branch.generate_revision_history(bzrlib.revision.NULL_REVISION)
418
 
        self.assertEqual([], tree.branch.revision_history())
419
 
 
420
 
    def test_create_checkout(self):
421
 
        tree_a = self.make_branch_and_tree('a')
422
 
        branch_a = tree_a.branch
423
 
        checkout_b = branch_a.create_checkout('b')
424
 
        self.assertEqual(None, checkout_b.last_revision())
425
 
        checkout_b.commit('rev1', rev_id='rev1')
426
 
        self.assertEqual('rev1', branch_a.last_revision())
427
 
        self.assertNotEqual(checkout_b.branch.base, branch_a.base)
428
 
 
429
 
        checkout_c = branch_a.create_checkout('c', lightweight=True)
430
 
        self.assertEqual('rev1', checkout_c.last_revision())
431
 
        checkout_c.commit('rev2', rev_id='rev2')
432
 
        self.assertEqual('rev2', branch_a.last_revision())
433
 
        self.assertEqual(checkout_c.branch.base, branch_a.base)
434
 
 
435
 
        os.mkdir('d')
436
 
        checkout_d = branch_a.create_checkout('d', lightweight=True)
437
 
        self.assertEqual('rev2', checkout_d.last_revision())
438
 
        os.mkdir('e')
439
 
        checkout_e = branch_a.create_checkout('e')
440
 
        self.assertEqual('rev2', checkout_e.last_revision())
441
 
 
442
 
    def test_create_anonymous_lightweight_checkout(self):
443
 
        """A lightweight checkout from a readonly branch should succeed."""
444
 
        tree_a = self.make_branch_and_tree('a')
445
 
        rev_id = tree_a.commit('put some content in the branch')
446
 
        source_branch = bzrlib.branch.Branch.open(
447
 
            'readonly+' + tree_a.bzrdir.root_transport.base)
448
 
        # sanity check that the test will be valid
449
 
        self.assertRaises((errors.LockError, errors.TransportNotPossible),
450
 
            source_branch.lock_write)
451
 
        checkout = source_branch.create_checkout('c', lightweight=True)
452
 
        self.assertEqual(rev_id, checkout.last_revision())
453
 
 
454
 
    def test_create_anonymous_heavyweight_checkout(self):
455
 
        """A regular checkout from a readonly branch should succeed."""
456
 
        tree_a = self.make_branch_and_tree('a')
457
 
        rev_id = tree_a.commit('put some content in the branch')
458
 
        source_branch = bzrlib.branch.Branch.open(
459
 
            'readonly+' + tree_a.bzrdir.root_transport.base)
460
 
        # sanity check that the test will be valid
461
 
        self.assertRaises((errors.LockError, errors.TransportNotPossible),
462
 
            source_branch.lock_write)
463
 
        checkout = source_branch.create_checkout('c')
464
 
        self.assertEqual(rev_id, checkout.last_revision())
465
 
 
466
 
 
467
 
class ChrootedTests(TestCaseWithBranch):
468
 
    """A support class that provides readonly urls outside the local namespace.
469
 
 
470
 
    This is done by checking if self.transport_server is a MemoryServer. if it
471
 
    is then we are chrooted already, if it is not then an HttpServer is used
472
 
    for readonly urls.
473
 
    """
474
 
 
475
 
    def setUp(self):
476
 
        super(ChrootedTests, self).setUp()
477
 
        if not self.transport_server == MemoryServer:
478
 
            self.transport_readonly_server = HttpServer
479
 
 
480
 
    def test_open_containing(self):
481
 
        self.assertRaises(NotBranchError, Branch.open_containing,
482
 
                          self.get_readonly_url(''))
483
 
        self.assertRaises(NotBranchError, Branch.open_containing,
484
 
                          self.get_readonly_url('g/p/q'))
485
 
        branch = self.make_branch('.')
486
 
        branch, relpath = Branch.open_containing(self.get_readonly_url(''))
487
 
        self.assertEqual('', relpath)
488
 
        branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
489
 
        self.assertEqual('g/p/q', relpath)
490
 
        
491
 
 
492
 
class InstrumentedTransaction(object):
493
 
 
494
 
    def finish(self):
495
 
        self.calls.append('finish')
496
 
 
497
 
    def __init__(self):
498
 
        self.calls = []
499
 
 
500
 
 
501
 
class TestDecorator(object):
502
 
 
503
 
    def __init__(self):
504
 
        self._calls = []
505
 
 
506
 
    def lock_read(self):
507
 
        self._calls.append('lr')
508
 
 
509
 
    def lock_write(self):
510
 
        self._calls.append('lw')
511
 
 
512
 
    def unlock(self):
513
 
        self._calls.append('ul')
514
 
 
515
 
    @needs_read_lock
516
 
    def do_with_read(self):
517
 
        return 1
518
 
 
519
 
    @needs_read_lock
520
 
    def except_with_read(self):
521
 
        raise RuntimeError
522
 
 
523
 
    @needs_write_lock
524
 
    def do_with_write(self):
525
 
        return 2
526
 
 
527
 
    @needs_write_lock
528
 
    def except_with_write(self):
529
 
        raise RuntimeError
530
 
 
531
 
 
532
 
class TestDecorators(TestCase):
533
 
 
534
 
    def test_needs_read_lock(self):
535
 
        branch = TestDecorator()
536
 
        self.assertEqual(1, branch.do_with_read())
537
 
        self.assertEqual(['lr', 'ul'], branch._calls)
538
 
 
539
 
    def test_excepts_in_read_lock(self):
540
 
        branch = TestDecorator()
541
 
        self.assertRaises(RuntimeError, branch.except_with_read)
542
 
        self.assertEqual(['lr', 'ul'], branch._calls)
543
 
 
544
 
    def test_needs_write_lock(self):
545
 
        branch = TestDecorator()
546
 
        self.assertEqual(2, branch.do_with_write())
547
 
        self.assertEqual(['lw', 'ul'], branch._calls)
548
 
 
549
 
    def test_excepts_in_write_lock(self):
550
 
        branch = TestDecorator()
551
 
        self.assertRaises(RuntimeError, branch.except_with_write)
552
 
        self.assertEqual(['lw', 'ul'], branch._calls)
553
 
 
554
 
 
555
 
class TestBranchTransaction(TestCaseWithBranch):
556
 
 
557
 
    def setUp(self):
558
 
        super(TestBranchTransaction, self).setUp()
559
 
        self.branch = None
560
 
        
561
 
    def test_default_get_transaction(self):
562
 
        """branch.get_transaction on a new branch should give a PassThrough."""
563
 
        self.failUnless(isinstance(self.get_branch().get_transaction(),
564
 
                                   transactions.PassThroughTransaction))
565
 
 
566
 
    def test__set_new_transaction(self):
567
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
568
 
 
569
 
    def test__set_over_existing_transaction_raises(self):
570
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
571
 
        self.assertRaises(errors.LockError,
572
 
                          self.get_branch()._set_transaction,
573
 
                          transactions.ReadOnlyTransaction())
574
 
 
575
 
    def test_finish_no_transaction_raises(self):
576
 
        self.assertRaises(errors.LockError, self.get_branch()._finish_transaction)
577
 
 
578
 
    def test_finish_readonly_transaction_works(self):
579
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
580
 
        self.get_branch()._finish_transaction()
581
 
        self.assertEqual(None, self.get_branch().control_files._transaction)
582
 
 
583
 
    def test_unlock_calls_finish(self):
584
 
        self.get_branch().lock_read()
585
 
        transaction = InstrumentedTransaction()
586
 
        self.get_branch().control_files._transaction = transaction
587
 
        self.get_branch().unlock()
588
 
        self.assertEqual(['finish'], transaction.calls)
589
 
 
590
 
    def test_lock_read_acquires_ro_transaction(self):
591
 
        self.get_branch().lock_read()
592
 
        self.failUnless(isinstance(self.get_branch().get_transaction(),
593
 
                                   transactions.ReadOnlyTransaction))
594
 
        self.get_branch().unlock()
595
 
        
596
 
    def test_lock_write_acquires_write_transaction(self):
597
 
        self.get_branch().lock_write()
598
 
        # cannot use get_transaction as its magic
599
 
        self.failUnless(isinstance(self.get_branch().control_files._transaction,
600
 
                                   transactions.WriteTransaction))
601
 
        self.get_branch().unlock()
602
 
 
603
 
 
604
 
class TestBranchPushLocations(TestCaseWithBranch):
605
 
 
606
 
    def test_get_push_location_unset(self):
607
 
        self.assertEqual(None, self.get_branch().get_push_location())
608
 
 
609
 
    def test_get_push_location_exact(self):
610
 
        from bzrlib.config import (locations_config_filename,
611
 
                                   ensure_config_dir_exists)
612
 
        ensure_config_dir_exists()
613
 
        fn = locations_config_filename()
614
 
        print >> open(fn, 'wt'), ("[%s]\n"
615
 
                                  "push_location=foo" %
616
 
                                  self.get_branch().base[:-1])
617
 
        self.assertEqual("foo", self.get_branch().get_push_location())
618
 
 
619
 
    def test_set_push_location(self):
620
 
        branch = self.get_branch()
621
 
        branch.set_push_location('foo')
622
 
        self.assertEqual('foo', branch.get_push_location())
623
 
 
624
 
 
625
 
class TestFormat(TestCaseWithBranch):
626
 
    """Tests for the format itself."""
627
 
 
628
 
    def test_format_initialize_find_open(self):
629
 
        # loopback test to check the current format initializes to itself.
630
 
        if not self.branch_format.is_supported():
631
 
            # unsupported formats are not loopback testable
632
 
            # because the default open will not open them and
633
 
            # they may not be initializable.
634
 
            return
635
 
        # supported formats must be able to init and open
636
 
        t = get_transport(self.get_url())
637
 
        readonly_t = get_transport(self.get_readonly_url())
638
 
        made_branch = self.make_branch('.')
639
 
        self.failUnless(isinstance(made_branch, branch.Branch))
640
 
 
641
 
        # find it via bzrdir opening:
642
 
        opened_control = bzrdir.BzrDir.open(readonly_t.base)
643
 
        direct_opened_branch = opened_control.open_branch()
644
 
        self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
645
 
        self.assertEqual(opened_control, direct_opened_branch.bzrdir)
646
 
        self.failUnless(isinstance(direct_opened_branch._format,
647
 
                        self.branch_format.__class__))
648
 
 
649
 
        # find it via Branch.open
650
 
        opened_branch = branch.Branch.open(readonly_t.base)
651
 
        self.failUnless(isinstance(opened_branch, made_branch.__class__))
652
 
        self.assertEqual(made_branch._format.__class__,
653
 
                         opened_branch._format.__class__)
654
 
        # if it has a unique id string, can we probe for it ?
655
 
        try:
656
 
            self.branch_format.get_format_string()
657
 
        except NotImplementedError:
658
 
            return
659
 
        self.assertEqual(self.branch_format,
660
 
                         branch.BranchFormat.find_format(opened_control))
661
 
 
662
 
 
663
 
class TestBound(TestCaseWithBranch):
664
 
 
665
 
    def test_bind_unbind(self):
666
 
        branch = self.make_branch('1')
667
 
        branch2 = self.make_branch('2')
668
 
        try:
669
 
            branch.bind(branch2)
670
 
        except errors.UpgradeRequired:
671
 
            raise TestSkipped('Format does not support binding')
672
 
        self.assertTrue(branch.unbind())
673
 
        self.assertFalse(branch.unbind())
674
 
        self.assertIs(None, branch.get_bound_location())
675
 
 
676
 
    def test_old_bound_location(self):
677
 
        branch = self.make_branch('branch1')
678
 
        try:
679
 
            self.assertIs(None, branch.get_old_bound_location())
680
 
        except errors.UpgradeRequired:
681
 
            raise TestSkipped('Format does not store old bound locations')
682
 
        branch2 = self.make_branch('branch2')
683
 
        branch.bind(branch2)
684
 
        self.assertIs(None, branch.get_old_bound_location())
685
 
        branch.unbind()
686
 
        self.assertContainsRe(branch.get_old_bound_location(), '\/branch2\/$')
687
 
 
688
 
 
689
 
class TestStrict(TestCaseWithBranch):
690
 
 
691
 
    def test_strict_history(self):
692
 
        tree1 = self.make_branch_and_tree('tree1')
693
 
        try:
694
 
            tree1.branch.set_append_revisions_only(True)
695
 
        except errors.UpgradeRequired:
696
 
            raise TestSkipped('Format does not support strict history')
697
 
        tree1.commit('empty commit')
698
 
        tree2 = tree1.bzrdir.sprout('tree2').open_workingtree()
699
 
        tree2.commit('empty commit 2')
700
 
        tree1.pull(tree2.branch)
701
 
        tree1.commit('empty commit 3')
702
 
        tree2.commit('empty commit 4')
703
 
        self.assertRaises(errors.DivergedBranches, tree1.pull, tree2.branch)
704
 
        tree2.merge_from_branch(tree1.branch)
705
 
        tree2.commit('empty commit 5')
706
 
        self.assertRaises(errors.AppendRevisionsOnlyViolation, tree1.pull,
707
 
                          tree2.branch)
708
 
        tree3 = tree1.bzrdir.sprout('tree3').open_workingtree()
709
 
        tree3.merge_from_branch(tree2.branch)
710
 
        tree3.commit('empty commit 6')
711
 
        tree2.pull(tree3.branch)
 
84
        br_c = copy_branch(br_a, 'c', basis_branch=br_b)
 
85
        self.assertEqual(br_a.revision_history(), br_c.revision_history())
 
86
        self.assertFalse(br_b.last_revision() in br_c.revision_history())
 
87
        br_c.get_revision(br_b.last_revision())
 
88
        
 
89
# TODO: rewrite this as a regular unittest, without relying on the displayed output        
 
90
#         >>> from bzrlib.commit import commit
 
91
#         >>> bzrlib.trace.silent = True
 
92
#         >>> br1 = ScratchBranch(files=['foo', 'bar'])
 
93
#         >>> br1.add('foo')
 
94
#         >>> br1.add('bar')
 
95
#         >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
 
96
#         >>> br2 = ScratchBranch()
 
97
#         >>> br2.update_revisions(br1)
 
98
#         Added 2 texts.
 
99
#         Added 1 inventories.
 
100
#         Added 1 revisions.
 
101
#         >>> br2.revision_history()
 
102
#         [u'REVISION-ID-1']
 
103
#         >>> br2.update_revisions(br1)
 
104
#         Added 0 revisions.
 
105
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
 
106
#         True