~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

  • Committer: Robert Collins
  • Date: 2005-10-11 07:00:25 UTC
  • mto: This revision was merged to the branch mainline in revision 1443.
  • Revision ID: robertc@robertcollins.net-20051011070025-bac6b53cb6186dfd
create a config module - there is enough config logic to make this worthwhile, and start testing config processing.

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
 
18
from bzrlib.branch import Branch
 
19
from bzrlib.clone import copy_branch
 
20
from bzrlib.commit import commit
 
21
import bzrlib.errors as errors
 
22
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
 
23
from bzrlib.selftest import TestCaseInTempDir
43
24
from bzrlib.trace import mutter
44
 
from bzrlib.transport import get_transport
45
 
from bzrlib.transport.http import HttpServer
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):
 
25
import bzrlib.transactions as transactions
 
26
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
 
27
 
 
28
class TestBranch(TestCaseInTempDir):
82
29
 
83
30
    def test_append_revisions(self):
84
31
        """Test appending more than one revision"""
85
 
        br = self.get_branch()
 
32
        br = Branch.initialize(".")
86
33
        br.append_revision("rev1")
87
34
        self.assertEquals(br.revision_history(), ["rev1",])
88
35
        br.append_revision("rev2", "rev3")
90
37
 
91
38
    def test_fetch_revisions(self):
92
39
        """Test fetch-revision operation."""
93
 
        get_transport(self.get_url()).mkdir('b1')
94
 
        get_transport(self.get_url()).mkdir('b2')
95
 
        wt = self.make_branch_and_tree('b1')
96
 
        b1 = wt.branch
97
 
        b2 = self.make_branch('b2')
98
 
        file('b1/foo', 'w').write('hello')
99
 
        wt.add(['foo'], ['foo-id'])
100
 
        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)
101
48
 
102
49
        mutter('start fetch')
103
 
        self.assertEqual((1, []), b2.fetch(b1))
104
 
 
105
 
        rev = b2.repository.get_revision('revision-1')
106
 
        tree = b2.repository.revision_tree('revision-1')
107
 
        self.assertEqual(tree.get_file_text('foo-id'), 'hello')
108
 
 
109
 
    def test_get_revision_delta(self):
110
 
        tree_a = self.make_branch_and_tree('a')
111
 
        self.build_tree(['a/foo'])
112
 
        tree_a.add('foo', 'file1')
113
 
        tree_a.commit('rev1', rev_id='rev1')
114
 
        self.build_tree(['a/vla'])
115
 
        tree_a.add('vla', 'file2')
116
 
        tree_a.commit('rev2', rev_id='rev2')
117
 
 
118
 
        delta = tree_a.branch.get_revision_delta(1)
119
 
        self.assertIsInstance(delta, TreeDelta)
120
 
        self.assertEqual([('foo', 'file1', 'file')], delta.added)
121
 
        delta = tree_a.branch.get_revision_delta(2)
122
 
        self.assertIsInstance(delta, TreeDelta)
123
 
        self.assertEqual([('vla', 'file2', 'file')], delta.added)
124
 
 
125
 
    def get_unbalanced_tree_pair(self):
126
 
        """Return two branches, a and b, with one file in a."""
127
 
        get_transport(self.get_url()).mkdir('a')
128
 
        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")
129
63
        file('a/b', 'wb').write('b')
130
 
        tree_a.add('b')
131
 
        tree_a.commit("silly commit", rev_id='A')
132
 
 
133
 
        get_transport(self.get_url()).mkdir('b')
134
 
        tree_b = self.make_branch_and_tree('b')
135
 
        return tree_a, tree_b
136
 
 
137
 
    def get_balanced_branch_pair(self):
138
 
        """Returns br_a, br_b as with one commit in a, and b has a's stores."""
139
 
        tree_a, tree_b = self.get_unbalanced_tree_pair()
140
 
        tree_b.branch.repository.fetch(tree_a.branch.repository)
141
 
        return tree_a, tree_b
142
 
 
143
 
    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):
144
80
        """Copy the stores from one branch to another"""
145
 
        tree_a, tree_b = self.get_balanced_branch_pair()
146
 
        tree_b.commit("silly commit")
 
81
        br_a, br_b = self.test_push_stores()
 
82
        commit(br_b, "silly commit")
147
83
        os.mkdir('c')
148
 
        # this fails to test that the history from a was not used.
149
 
        dir_c = tree_a.bzrdir.clone('c', basis=tree_b.bzrdir)
150
 
        self.assertEqual(tree_a.branch.revision_history(),
151
 
                         dir_c.open_branch().revision_history())
 
84
        br_c = copy_branch(br_a, 'c', basis_branch=br_b)
 
85
        self.assertEqual(br_a.revision_history(), br_c.revision_history())
 
86
        ## # basis branches currently disabled for weave format
 
87
        ## self.assertFalse(br_b.last_revision() in br_c.revision_history())
 
88
        ## br_c.get_revision(br_b.last_revision())
152
89
 
153
 
    def test_clone_partial(self):
 
90
    def test_copy_partial(self):
154
91
        """Copy only part of the history of a branch."""
155
 
        # TODO: RBC 20060208 test with a revision not on revision-history.
156
 
        #       what should that behaviour be ? Emailed the list.
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
 
        repo_b = self.make_repository('b')
165
 
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
166
 
        br_b = wt_a.bzrdir.open_branch().clone(repo_b.bzrdir, revision_id='1')
167
 
        self.assertEqual(br_b.last_revision(), '1')
168
 
 
169
 
    def test_sprout_partial(self):
170
 
        # test sprouting with a prefix of the revision-history.
171
 
        # also needs not-on-revision-history behaviour defined.
172
 
        wt_a = self.make_branch_and_tree('a')
173
 
        self.build_tree(['a/one'])
174
 
        wt_a.add(['one'])
175
 
        wt_a.commit('commit one', rev_id='1')
176
 
        self.build_tree(['a/two'])
177
 
        wt_a.add(['two'])
178
 
        wt_a.commit('commit two', rev_id='2')
179
 
        repo_b = self.make_repository('b')
180
 
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
181
 
        br_b = wt_a.bzrdir.open_branch().sprout(repo_b.bzrdir, revision_id='1')
182
 
        self.assertEqual(br_b.last_revision(), '1')
183
 
 
184
 
    def test_clone_branch_nickname(self):
185
 
        # test the nick name is preserved always
186
 
        raise TestSkipped('XXX branch cloning is not yet tested..')
187
 
 
188
 
    def test_clone_branch_parent(self):
189
 
        # test the parent is preserved always
190
 
        raise TestSkipped('XXX branch cloning is not yet tested..')
191
 
        
192
 
    def test_sprout_branch_nickname(self):
193
 
        # test the nick name is reset always
194
 
        raise TestSkipped('XXX branch sprouting is not yet tested..')
195
 
 
196
 
    def test_sprout_branch_parent(self):
197
 
        source = self.make_branch('source')
198
 
        target = source.bzrdir.sprout(self.get_url('target')).open_branch()
199
 
        self.assertEqual(source.bzrdir.root_transport.base, target.get_parent())
200
 
 
201
 
    def test_submit_branch(self):
202
 
        """Submit location can be queried and set"""
203
 
        branch = self.make_branch('branch')
204
 
        self.assertEqual(branch.get_submit_branch(), None)
205
 
        branch.set_submit_branch('sftp://example.com')
206
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.com')
207
 
        branch.set_submit_branch('sftp://example.net')
208
 
        self.assertEqual(branch.get_submit_branch(), 'sftp://example.net')
209
 
        
 
92
        self.build_tree(['a/', 'a/one'])
 
93
        br_a = Branch.initialize('a')
 
94
        br_a.add(['one'])
 
95
        br_a.commit('commit one', rev_id='u@d-1')
 
96
        self.build_tree(['a/two'])
 
97
        br_a.add(['two'])
 
98
        br_a.commit('commit two', rev_id='u@d-2')
 
99
        br_b = copy_branch(br_a, 'b', revision='u@d-1')
 
100
        self.assertEqual(br_b.last_revision(), 'u@d-1')
 
101
        self.assertTrue(os.path.exists('b/one'))
 
102
        self.assertFalse(os.path.exists('b/two'))
 
103
        
 
104
 
210
105
    def test_record_initial_ghost_merge(self):
211
106
        """A pending merge with no revision present is still a merge."""
212
 
        wt = self.make_branch_and_tree('.')
213
 
        branch = wt.branch
214
 
        wt.add_pending_merge('non:existent@rev--ision--0--2')
215
 
        wt.commit('pretend to merge nonexistent-revision', rev_id='first')
216
 
        rev = branch.repository.get_revision(branch.last_revision())
 
107
        branch = Branch.initialize('.')
 
108
        branch.add_pending_merge('non:existent@rev--ision--0--2')
 
109
        branch.commit('pretend to merge nonexistent-revision', rev_id='first')
 
110
        rev = branch.get_revision(branch.last_revision())
217
111
        self.assertEqual(len(rev.parent_ids), 1)
218
112
        # parent_sha1s is not populated now, WTF. rbc 20051003
219
113
        self.assertEqual(len(rev.parent_sha1s), 0)
220
114
        self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
221
115
 
222
 
    def test_bad_revision(self):
223
 
        self.assertRaises(errors.InvalidRevisionId,
224
 
                          self.get_branch().repository.get_revision,
225
 
                          None)
226
 
 
227
116
# TODO 20051003 RBC:
228
117
# compare the gpg-to-sign info for a commit with a ghost and 
229
118
#     an identical tree without a ghost
231
120
        
232
121
    def test_pending_merges(self):
233
122
        """Tracking pending-merged revisions."""
234
 
        wt = self.make_branch_and_tree('.')
235
 
        b = wt.branch
236
 
        self.assertEquals(wt.pending_merges(), [])
237
 
        wt.add_pending_merge('foo@azkhazan-123123-abcabc')
238
 
        self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
239
 
        wt.add_pending_merge('foo@azkhazan-123123-abcabc')
240
 
        self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
241
 
        wt.add_pending_merge('wibble@fofof--20050401--1928390812')
242
 
        self.assertEquals(wt.pending_merges(),
 
123
        b = Branch.initialize('.')
 
124
 
 
125
        self.assertEquals(b.pending_merges(), [])
 
126
        b.add_pending_merge('foo@azkhazan-123123-abcabc')
 
127
        self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
 
128
        b.add_pending_merge('foo@azkhazan-123123-abcabc')
 
129
        self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
 
130
        b.add_pending_merge('wibble@fofof--20050401--1928390812')
 
131
        self.assertEquals(b.pending_merges(),
243
132
                          ['foo@azkhazan-123123-abcabc',
244
133
                           'wibble@fofof--20050401--1928390812'])
245
 
        wt.commit("commit from base with two merges")
246
 
        rev = b.repository.get_revision(b.revision_history()[0])
 
134
        b.commit("commit from base with two merges")
 
135
        rev = b.get_revision(b.revision_history()[0])
247
136
        self.assertEquals(len(rev.parent_ids), 2)
248
137
        self.assertEquals(rev.parent_ids[0],
249
138
                          'foo@azkhazan-123123-abcabc')
250
139
        self.assertEquals(rev.parent_ids[1],
251
140
                           'wibble@fofof--20050401--1928390812')
252
141
        # list should be cleared when we do a commit
253
 
        self.assertEquals(wt.pending_merges(), [])
254
 
 
255
 
    def test_sign_existing_revision(self):
256
 
        wt = self.make_branch_and_tree('.')
257
 
        branch = wt.branch
258
 
        wt.commit("base", allow_pointless=True, rev_id='A')
259
 
        from bzrlib.testament import Testament
260
 
        strategy = gpg.LoopbackGPGStrategy(None)
261
 
        branch.repository.sign_revision('A', strategy)
262
 
        self.assertEqual(Testament.from_revision(branch.repository, 
263
 
                         'A').as_short_text(),
264
 
                         branch.repository.get_signature_text('A'))
265
 
 
266
 
    def test_store_signature(self):
267
 
        wt = self.make_branch_and_tree('.')
268
 
        branch = wt.branch
269
 
        branch.repository.store_revision_signature(
270
 
            gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
271
 
        self.assertRaises(errors.NoSuchRevision,
272
 
                          branch.repository.has_signature_for_revision_id,
273
 
                          'A')
274
 
        wt.commit("base", allow_pointless=True, rev_id='A')
275
 
        self.assertEqual('FOO', 
276
 
                         branch.repository.get_signature_text('A'))
277
 
 
278
 
    def test_branch_keeps_signatures(self):
279
 
        wt = self.make_branch_and_tree('source')
280
 
        wt.commit('A', allow_pointless=True, rev_id='A')
281
 
        wt.branch.repository.sign_revision('A',
282
 
            gpg.LoopbackGPGStrategy(None))
283
 
        #FIXME: clone should work to urls,
284
 
        # wt.clone should work to disks.
285
 
        self.build_tree(['target/'])
286
 
        d2 = wt.bzrdir.clone('target')
287
 
        self.assertEqual(wt.branch.repository.get_signature_text('A'),
288
 
                         d2.open_repository().get_signature_text('A'))
289
 
 
290
 
    def test_nicks(self):
291
 
        """Branch nicknames"""
292
 
        t = get_transport(self.get_url())
293
 
        t.mkdir('bzr.dev')
294
 
        branch = self.make_branch('bzr.dev')
295
 
        self.assertEqual(branch.nick, 'bzr.dev')
296
 
        t.move('bzr.dev', 'bzr.ab')
297
 
        branch = Branch.open(self.get_url('bzr.ab'))
298
 
        self.assertEqual(branch.nick, 'bzr.ab')
299
 
        branch.nick = "Aaron's branch"
300
 
        branch.nick = "Aaron's branch"
301
 
        self.failUnless(
302
 
            t.has(
303
 
                t.relpath(
304
 
                    branch.control_files.controlfilename("branch.conf")
305
 
                    )
306
 
                )
307
 
            )
308
 
        self.assertEqual(branch.nick, "Aaron's branch")
309
 
        t.move('bzr.ab', 'integration')
310
 
        branch = Branch.open(self.get_url('integration'))
311
 
        self.assertEqual(branch.nick, "Aaron's branch")
312
 
        branch.nick = u"\u1234"
313
 
        self.assertEqual(branch.nick, u"\u1234")
314
 
 
315
 
    def test_commit_nicks(self):
316
 
        """Nicknames are committed to the revision"""
317
 
        get_transport(self.get_url()).mkdir('bzr.dev')
318
 
        wt = self.make_branch_and_tree('bzr.dev')
319
 
        branch = wt.branch
320
 
        branch.nick = "My happy branch"
321
 
        wt.commit('My commit respect da nick.')
322
 
        committed = branch.repository.get_revision(branch.last_revision())
323
 
        self.assertEqual(committed.properties["branch-nick"], 
324
 
                         "My happy branch")
325
 
 
326
 
    def test_create_open_branch_uses_repository(self):
327
 
        try:
328
 
            repo = self.make_repository('.', shared=True)
329
 
        except errors.IncompatibleFormat:
330
 
            return
331
 
        repo.bzrdir.root_transport.mkdir('child')
332
 
        child_dir = self.bzrdir_format.initialize('child')
333
 
        try:
334
 
            child_branch = self.branch_format.initialize(child_dir)
335
 
        except errors.UninitializableFormat:
336
 
            # branch references are not default init'able.
337
 
            return
338
 
        self.assertEqual(repo.bzrdir.root_transport.base,
339
 
                         child_branch.repository.bzrdir.root_transport.base)
340
 
        child_branch = branch.Branch.open(self.get_url('child'))
341
 
        self.assertEqual(repo.bzrdir.root_transport.base,
342
 
                         child_branch.repository.bzrdir.root_transport.base)
343
 
 
344
 
    def test_format_description(self):
345
 
        tree = self.make_branch_and_tree('tree')
346
 
        text = tree.branch._format.get_format_description()
347
 
        self.failUnless(len(text))
348
 
 
349
 
    def test_check_branch_report_results(self):
350
 
        """Checking a branch produces results which can be printed"""
351
 
        branch = self.make_branch('.')
352
 
        result = branch.check()
353
 
        # reports results through logging
354
 
        result.report_results(verbose=True)
355
 
        result.report_results(verbose=False)
356
 
 
357
 
    def test_get_commit_builder(self):
358
 
        self.assertIsInstance(self.make_branch(".").get_commit_builder([]), 
359
 
            repository.CommitBuilder)
360
 
 
361
 
    def test_generate_revision_history(self):
362
 
        """Create a fake revision history easily."""
363
 
        tree = self.make_branch_and_tree('.')
364
 
        rev1 = tree.commit('foo')
365
 
        orig_history = tree.branch.revision_history()
366
 
        rev2 = tree.commit('bar', allow_pointless=True)
367
 
        tree.branch.generate_revision_history(rev1)
368
 
        self.assertEqual(orig_history, tree.branch.revision_history())
369
 
 
370
 
    def test_generate_revision_history_NULL_REVISION(self):
371
 
        tree = self.make_branch_and_tree('.')
372
 
        rev1 = tree.commit('foo')
373
 
        tree.branch.generate_revision_history(bzrlib.revision.NULL_REVISION)
374
 
        self.assertEqual([], tree.branch.revision_history())
375
 
 
376
 
 
377
 
class ChrootedTests(TestCaseWithBranch):
378
 
    """A support class that provides readonly urls outside the local namespace.
379
 
 
380
 
    This is done by checking if self.transport_server is a MemoryServer. if it
381
 
    is then we are chrooted already, if it is not then an HttpServer is used
382
 
    for readonly urls.
383
 
    """
384
 
 
385
 
    def setUp(self):
386
 
        super(ChrootedTests, self).setUp()
387
 
        if not self.transport_server == MemoryServer:
388
 
            self.transport_readonly_server = HttpServer
 
142
        self.assertEquals(b.pending_merges(), [])
 
143
 
 
144
 
 
145
class TestRemote(TestCaseWithWebserver):
389
146
 
390
147
    def test_open_containing(self):
391
148
        self.assertRaises(NotBranchError, Branch.open_containing,
392
 
                          self.get_readonly_url(''))
 
149
                          self.get_remote_url(''))
393
150
        self.assertRaises(NotBranchError, Branch.open_containing,
394
 
                          self.get_readonly_url('g/p/q'))
395
 
        branch = self.make_branch('.')
396
 
        branch, relpath = Branch.open_containing(self.get_readonly_url(''))
397
 
        self.assertEqual('', relpath)
398
 
        branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
399
 
        self.assertEqual('g/p/q', relpath)
 
151
                          self.get_remote_url('g/p/q'))
 
152
        b = Branch.initialize('.')
 
153
        Branch.open_containing(self.get_remote_url(''))
 
154
        Branch.open_containing(self.get_remote_url('g/p/q'))
400
155
        
 
156
# TODO: rewrite this as a regular unittest, without relying on the displayed output        
 
157
#         >>> from bzrlib.commit import commit
 
158
#         >>> bzrlib.trace.silent = True
 
159
#         >>> br1 = ScratchBranch(files=['foo', 'bar'])
 
160
#         >>> br1.add('foo')
 
161
#         >>> br1.add('bar')
 
162
#         >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
 
163
#         >>> br2 = ScratchBranch()
 
164
#         >>> br2.update_revisions(br1)
 
165
#         Added 2 texts.
 
166
#         Added 1 inventories.
 
167
#         Added 1 revisions.
 
168
#         >>> br2.revision_history()
 
169
#         [u'REVISION-ID-1']
 
170
#         >>> br2.update_revisions(br1)
 
171
#         Added 0 revisions.
 
172
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
 
173
#         True
401
174
 
402
175
class InstrumentedTransaction(object):
403
176
 
408
181
        self.calls = []
409
182
 
410
183
 
411
 
class TestDecorator(object):
412
 
 
413
 
    def __init__(self):
414
 
        self._calls = []
415
 
 
416
 
    def lock_read(self):
417
 
        self._calls.append('lr')
418
 
 
419
 
    def lock_write(self):
420
 
        self._calls.append('lw')
421
 
 
422
 
    def unlock(self):
423
 
        self._calls.append('ul')
424
 
 
425
 
    @needs_read_lock
426
 
    def do_with_read(self):
427
 
        return 1
428
 
 
429
 
    @needs_read_lock
430
 
    def except_with_read(self):
431
 
        raise RuntimeError
432
 
 
433
 
    @needs_write_lock
434
 
    def do_with_write(self):
435
 
        return 2
436
 
 
437
 
    @needs_write_lock
438
 
    def except_with_write(self):
439
 
        raise RuntimeError
440
 
 
441
 
 
442
 
class TestDecorators(TestCase):
443
 
 
444
 
    def test_needs_read_lock(self):
445
 
        branch = TestDecorator()
446
 
        self.assertEqual(1, branch.do_with_read())
447
 
        self.assertEqual(['lr', 'ul'], branch._calls)
448
 
 
449
 
    def test_excepts_in_read_lock(self):
450
 
        branch = TestDecorator()
451
 
        self.assertRaises(RuntimeError, branch.except_with_read)
452
 
        self.assertEqual(['lr', 'ul'], branch._calls)
453
 
 
454
 
    def test_needs_write_lock(self):
455
 
        branch = TestDecorator()
456
 
        self.assertEqual(2, branch.do_with_write())
457
 
        self.assertEqual(['lw', 'ul'], branch._calls)
458
 
 
459
 
    def test_excepts_in_write_lock(self):
460
 
        branch = TestDecorator()
461
 
        self.assertRaises(RuntimeError, branch.except_with_write)
462
 
        self.assertEqual(['lw', 'ul'], branch._calls)
463
 
 
464
 
 
465
 
class TestBranchTransaction(TestCaseWithBranch):
 
184
class TestBranchTransaction(TestCaseInTempDir):
466
185
 
467
186
    def setUp(self):
468
187
        super(TestBranchTransaction, self).setUp()
469
 
        self.branch = None
 
188
        self.branch = Branch.initialize('.')
470
189
        
471
190
    def test_default_get_transaction(self):
472
191
        """branch.get_transaction on a new branch should give a PassThrough."""
473
 
        self.failUnless(isinstance(self.get_branch().get_transaction(),
 
192
        self.failUnless(isinstance(self.branch.get_transaction(),
474
193
                                   transactions.PassThroughTransaction))
475
194
 
476
195
    def test__set_new_transaction(self):
477
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
 
196
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
478
197
 
479
198
    def test__set_over_existing_transaction_raises(self):
480
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
 
199
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
481
200
        self.assertRaises(errors.LockError,
482
 
                          self.get_branch()._set_transaction,
 
201
                          self.branch._set_transaction,
483
202
                          transactions.ReadOnlyTransaction())
484
203
 
485
204
    def test_finish_no_transaction_raises(self):
486
 
        self.assertRaises(errors.LockError, self.get_branch()._finish_transaction)
 
205
        self.assertRaises(errors.LockError, self.branch._finish_transaction)
487
206
 
488
207
    def test_finish_readonly_transaction_works(self):
489
 
        self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
490
 
        self.get_branch()._finish_transaction()
491
 
        self.assertEqual(None, self.get_branch().control_files._transaction)
 
208
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
209
        self.branch._finish_transaction()
 
210
        self.assertEqual(None, self.branch._transaction)
492
211
 
493
212
    def test_unlock_calls_finish(self):
494
 
        self.get_branch().lock_read()
 
213
        self.branch.lock_read()
495
214
        transaction = InstrumentedTransaction()
496
 
        self.get_branch().control_files._transaction = transaction
497
 
        self.get_branch().unlock()
 
215
        self.branch._transaction = transaction
 
216
        self.branch.unlock()
498
217
        self.assertEqual(['finish'], transaction.calls)
499
218
 
500
219
    def test_lock_read_acquires_ro_transaction(self):
501
 
        self.get_branch().lock_read()
502
 
        self.failUnless(isinstance(self.get_branch().get_transaction(),
 
220
        self.branch.lock_read()
 
221
        self.failUnless(isinstance(self.branch.get_transaction(),
503
222
                                   transactions.ReadOnlyTransaction))
504
 
        self.get_branch().unlock()
 
223
        self.branch.unlock()
505
224
        
506
 
    def test_lock_write_acquires_write_transaction(self):
507
 
        self.get_branch().lock_write()
 
225
    def test_lock_write_acquires_passthrough_transaction(self):
 
226
        self.branch.lock_write()
508
227
        # cannot use get_transaction as its magic
509
 
        self.failUnless(isinstance(self.get_branch().control_files._transaction,
510
 
                                   transactions.WriteTransaction))
511
 
        self.get_branch().unlock()
512
 
 
513
 
 
514
 
class TestBranchPushLocations(TestCaseWithBranch):
515
 
 
516
 
    def test_get_push_location_unset(self):
517
 
        self.assertEqual(None, self.get_branch().get_push_location())
518
 
 
519
 
    def test_get_push_location_exact(self):
520
 
        from bzrlib.config import (locations_config_filename,
521
 
                                   ensure_config_dir_exists)
522
 
        ensure_config_dir_exists()
523
 
        fn = locations_config_filename()
524
 
        print >> open(fn, 'wt'), ("[%s]\n"
525
 
                                  "push_location=foo" %
526
 
                                  self.get_branch().base[:-1])
527
 
        self.assertEqual("foo", self.get_branch().get_push_location())
528
 
 
529
 
    def test_set_push_location(self):
530
 
        from bzrlib.config import (locations_config_filename,
531
 
                                   ensure_config_dir_exists)
532
 
        ensure_config_dir_exists()
533
 
        fn = locations_config_filename()
534
 
        branch = self.get_branch()
535
 
        branch.set_push_location('foo')
536
 
        local_path = urlutils.local_path_from_url(branch.base[:-1])
537
 
        self.assertFileEqual("[%s]\n"
538
 
                             "push_location = foo" % local_path,
539
 
                             fn)
540
 
 
541
 
    # TODO RBC 20051029 test getting a push location from a branch in a 
542
 
    # recursive section - that is, it appends the branch name.
543
 
 
544
 
 
545
 
class TestFormat(TestCaseWithBranch):
546
 
    """Tests for the format itself."""
547
 
 
548
 
    def test_format_initialize_find_open(self):
549
 
        # loopback test to check the current format initializes to itself.
550
 
        if not self.branch_format.is_supported():
551
 
            # unsupported formats are not loopback testable
552
 
            # because the default open will not open them and
553
 
            # they may not be initializable.
554
 
            return
555
 
        # supported formats must be able to init and open
556
 
        t = get_transport(self.get_url())
557
 
        readonly_t = get_transport(self.get_readonly_url())
558
 
        made_branch = self.make_branch('.')
559
 
        self.failUnless(isinstance(made_branch, branch.Branch))
560
 
 
561
 
        # find it via bzrdir opening:
562
 
        opened_control = bzrdir.BzrDir.open(readonly_t.base)
563
 
        direct_opened_branch = opened_control.open_branch()
564
 
        self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
565
 
        self.assertEqual(opened_control, direct_opened_branch.bzrdir)
566
 
        self.failUnless(isinstance(direct_opened_branch._format,
567
 
                        self.branch_format.__class__))
568
 
 
569
 
        # find it via Branch.open
570
 
        opened_branch = branch.Branch.open(readonly_t.base)
571
 
        self.failUnless(isinstance(opened_branch, made_branch.__class__))
572
 
        self.assertEqual(made_branch._format.__class__,
573
 
                         opened_branch._format.__class__)
574
 
        # if it has a unique id string, can we probe for it ?
575
 
        try:
576
 
            self.branch_format.get_format_string()
577
 
        except NotImplementedError:
578
 
            return
579
 
        self.assertEqual(self.branch_format,
580
 
                         branch.BranchFormat.find_format(opened_control))
581
 
 
582
 
 
 
228
        self.failUnless(isinstance(self.branch._transaction,
 
229
                                   transactions.PassThroughTransaction))
 
230
        self.branch.unlock()