~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/branch_implementations/test_branch.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-02-18 02:33:47 UTC
  • mfrom: (1534.1.24 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20060218023347-0952c65f668bfd68
Merge Robert Collins integration.

Show diffs side-by-side

added added

removed removed

Lines of Context:
19
19
import os
20
20
import sys
21
21
 
22
 
import bzrlib.branch as branch
 
22
import bzrlib.branch
 
23
import bzrlib.bzrdir as bzrdir
23
24
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
24
25
from bzrlib.commit import commit
25
26
import bzrlib.errors as errors
31
32
                           )
32
33
import bzrlib.gpg
33
34
from bzrlib.osutils import getcwd
34
 
from bzrlib.revision import NULL_REVISION
35
35
from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
36
36
from bzrlib.trace import mutter
37
37
import bzrlib.transactions as transactions
41
41
from bzrlib.upgrade import upgrade
42
42
from bzrlib.workingtree import WorkingTree
43
43
 
 
44
 
44
45
# TODO: Make a branch using basis branch, and check that it 
45
46
# doesn't request any files that could have been avoided, by 
46
47
# hooking into the Transport.
58
59
        return self.branch
59
60
 
60
61
    def make_branch(self, relpath):
 
62
        repo = self.make_repository(relpath)
 
63
        # fixme RBC 20060210 this isnt necessarily a fixable thing,
 
64
        # Skipped is the wrong exception to raise.
 
65
        try:
 
66
            return self.branch_format.initialize(repo.bzrdir)
 
67
        except errors.UninitializableFormat:
 
68
            raise TestSkipped('Uninitializable branch format')
 
69
 
 
70
    def make_repository(self, relpath, shared=False):
61
71
        try:
62
72
            url = self.get_url(relpath)
63
73
            segments = url.split('/')
68
78
                    t.mkdir(segments[-1])
69
79
                except FileExists:
70
80
                    pass
71
 
            return self.branch_format.initialize(url)
 
81
            made_control = self.bzrdir_format.initialize(url)
 
82
            return made_control.create_repository(shared=shared)
72
83
        except UninitializableFormat:
73
84
            raise TestSkipped("Format %s is not initializable.")
74
85
 
88
99
        from bzrlib.fetch import Fetcher
89
100
        get_transport(self.get_url()).mkdir('b1')
90
101
        get_transport(self.get_url()).mkdir('b2')
91
 
        b1 = self.make_branch('b1')
 
102
        wt = self.make_branch_and_tree('b1')
 
103
        b1 = wt.branch
92
104
        b2 = self.make_branch('b2')
93
 
        wt = WorkingTree.create(b1, 'b1')
94
105
        file('b1/foo', 'w').write('hello')
95
106
        wt.add(['foo'], ['foo-id'])
96
107
        wt.commit('lala!', rev_id='revision-1', allow_pointless=False)
105
116
        tree = b2.repository.revision_tree('revision-1')
106
117
        eq(tree.get_file_text('foo-id'), 'hello')
107
118
 
108
 
    def test_revision_tree(self):
109
 
        b1 = self.get_branch()
110
 
        wt = WorkingTree.create(b1, '.')
111
 
        wt.commit('lala!', rev_id='revision-1', allow_pointless=True)
112
 
        tree = b1.repository.revision_tree('revision-1')
113
 
        tree = b1.repository.revision_tree(None)
114
 
        self.assertEqual(len(tree.list_files()), 0)
115
 
        tree = b1.repository.revision_tree(NULL_REVISION)
116
 
        self.assertEqual(len(tree.list_files()), 0)
117
 
 
118
119
    def get_unbalanced_tree_pair(self):
119
120
        """Return two branches, a and b, with one file in a."""
120
121
        get_transport(self.get_url()).mkdir('a')
121
 
        br_a = self.make_branch('a')
122
 
        tree_a = WorkingTree.create(br_a, 'a')
 
122
        tree_a = self.make_branch_and_tree('a')
123
123
        file('a/b', 'wb').write('b')
124
124
        tree_a.add('b')
125
125
        tree_a.commit("silly commit", rev_id='A')
126
126
 
127
127
        get_transport(self.get_url()).mkdir('b')
128
 
        br_b = self.make_branch('b')
129
 
        tree_b = WorkingTree.create(br_b, 'b')
 
128
        tree_b = self.make_branch_and_tree('b')
130
129
        return tree_a, tree_b
131
130
 
132
131
    def get_balanced_branch_pair(self):
133
132
        """Returns br_a, br_b as with one commit in a, and b has a's stores."""
134
133
        tree_a, tree_b = self.get_unbalanced_tree_pair()
135
 
        tree_a.branch.push_stores(tree_b.branch)
 
134
        tree_b.branch.repository.fetch(tree_a.branch.repository)
136
135
        return tree_a, tree_b
137
136
 
138
 
    def test_push_stores(self):
139
 
        """Copy the stores from one branch to another"""
140
 
        tree_a, tree_b = self.get_unbalanced_tree_pair()
141
 
        br_a = tree_a.branch
142
 
        br_b = tree_b.branch
143
 
        # ensure the revision is missing.
144
 
        self.assertRaises(NoSuchRevision, br_b.repository.get_revision, 
145
 
                          br_a.revision_history()[0])
146
 
        br_a.push_stores(br_b)
147
 
        # check that b now has all the data from a's first commit.
148
 
        rev = br_b.repository.get_revision(br_a.revision_history()[0])
149
 
        tree = br_b.repository.revision_tree(br_a.revision_history()[0])
150
 
        for file_id in tree:
151
 
            if tree.inventory[file_id].kind == "file":
152
 
                tree.get_file(file_id).read()
153
 
 
154
137
    def test_clone_branch(self):
155
138
        """Copy the stores from one branch to another"""
156
139
        tree_a, tree_b = self.get_balanced_branch_pair()
157
140
        tree_b.commit("silly commit")
158
141
        os.mkdir('c')
159
 
        br_c = tree_a.branch.clone('c', basis_branch=tree_b.branch)
 
142
        # this fails to test that the history from a was not used.
 
143
        dir_c = tree_a.bzrdir.clone('c', basis=tree_b.bzrdir)
160
144
        self.assertEqual(tree_a.branch.revision_history(),
161
 
                         br_c.revision_history())
 
145
                         dir_c.open_branch().revision_history())
162
146
 
163
147
    def test_clone_partial(self):
164
148
        """Copy only part of the history of a branch."""
165
 
        get_transport(self.get_url()).mkdir('a')
166
 
        br_a = self.make_branch('a')
167
 
        wt = WorkingTree.create(br_a, "a")
168
 
        self.build_tree(['a/one'])
169
 
        wt.add(['one'])
170
 
        wt.commit('commit one', rev_id='u@d-1')
171
 
        self.build_tree(['a/two'])
172
 
        wt.add(['two'])
173
 
        wt.commit('commit two', rev_id='u@d-2')
174
 
        br_b = br_a.clone('b', revision='u@d-1')
175
 
        self.assertEqual(br_b.last_revision(), 'u@d-1')
176
 
        self.assertTrue(os.path.exists('b/one'))
177
 
        self.assertFalse(os.path.exists('b/two'))
 
149
        # TODO: RBC 20060208 test with a revision not on revision-history.
 
150
        #       what should that behaviour be ? Emailed the list.
 
151
        wt_a = self.make_branch_and_tree('a')
 
152
        self.build_tree(['a/one'])
 
153
        wt_a.add(['one'])
 
154
        wt_a.commit('commit one', rev_id='1')
 
155
        self.build_tree(['a/two'])
 
156
        wt_a.add(['two'])
 
157
        wt_a.commit('commit two', rev_id='2')
 
158
        repo_b = self.make_repository('b')
 
159
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
 
160
        br_b = wt_a.bzrdir.open_branch().clone(repo_b.bzrdir, revision_id='1')
 
161
        self.assertEqual(br_b.last_revision(), '1')
 
162
 
 
163
    def test_sprout_partial(self):
 
164
        # test sprouting with a prefix of the revision-history.
 
165
        # also needs not-on-revision-history behaviour defined.
 
166
        wt_a = self.make_branch_and_tree('a')
 
167
        self.build_tree(['a/one'])
 
168
        wt_a.add(['one'])
 
169
        wt_a.commit('commit one', rev_id='1')
 
170
        self.build_tree(['a/two'])
 
171
        wt_a.add(['two'])
 
172
        wt_a.commit('commit two', rev_id='2')
 
173
        repo_b = self.make_repository('b')
 
174
        wt_a.bzrdir.open_repository().copy_content_into(repo_b)
 
175
        br_b = wt_a.bzrdir.open_branch().sprout(repo_b.bzrdir, revision_id='1')
 
176
        self.assertEqual(br_b.last_revision(), '1')
 
177
 
 
178
    def test_clone_branch_nickname(self):
 
179
        # test the nick name is preserved always
 
180
        raise TestSkipped('XXX branch cloning is not yet tested..')
 
181
 
 
182
    def test_clone_branch_parent(self):
 
183
        # test the parent is preserved always
 
184
        raise TestSkipped('XXX branch cloning is not yet tested..')
 
185
        
 
186
    def test_sprout_branch_nickname(self):
 
187
        # test the nick name is reset always
 
188
        raise TestSkipped('XXX branch sprouting is not yet tested..')
 
189
 
 
190
    def test_sprout_branch_parent(self):
 
191
        source = self.make_branch('source')
 
192
        target = source.bzrdir.sprout(self.get_url('target')).open_branch()
 
193
        self.assertEqual(source.bzrdir.root_transport.base, target.get_parent())
178
194
        
179
195
    def test_record_initial_ghost_merge(self):
180
196
        """A pending merge with no revision present is still a merge."""
181
 
        branch = self.get_branch()
182
 
        wt = WorkingTree.create(branch, ".")
 
197
        wt = self.make_branch_and_tree('.')
 
198
        branch = wt.branch
183
199
        wt.add_pending_merge('non:existent@rev--ision--0--2')
184
200
        wt.commit('pretend to merge nonexistent-revision', rev_id='first')
185
201
        rev = branch.repository.get_revision(branch.last_revision())
200
216
        
201
217
    def test_pending_merges(self):
202
218
        """Tracking pending-merged revisions."""
203
 
        b = self.get_branch()
204
 
        wt = WorkingTree.create(b, '.')
 
219
        wt = self.make_branch_and_tree('.')
 
220
        b = wt.branch
205
221
        self.assertEquals(wt.pending_merges(), [])
206
222
        wt.add_pending_merge('foo@azkhazan-123123-abcabc')
207
223
        self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
222
238
        self.assertEquals(wt.pending_merges(), [])
223
239
 
224
240
    def test_sign_existing_revision(self):
225
 
        branch = self.get_branch()
226
 
        wt = WorkingTree.create(branch, ".")
 
241
        wt = self.make_branch_and_tree('.')
 
242
        branch = wt.branch
227
243
        wt.commit("base", allow_pointless=True, rev_id='A')
228
244
        from bzrlib.testament import Testament
229
245
        strategy = bzrlib.gpg.LoopbackGPGStrategy(None)
249
265
        #FIXME: clone should work to urls,
250
266
        # wt.clone should work to disks.
251
267
        self.build_tree(['target/'])
252
 
        b2 = wt.branch.clone('target')
 
268
        d2 = wt.bzrdir.clone('target')
253
269
        self.assertEqual(wt.branch.repository.revision_store.get('A', 
254
270
                            'sig').read(),
255
 
                         b2.repository.revision_store.get('A', 
 
271
                         d2.open_repository().revision_store.get('A', 
256
272
                            'sig').read())
257
273
 
258
274
    def test_upgrade_preserves_signatures(self):
259
 
        # this is in the current test format
260
275
        wt = self.make_branch_and_tree('source')
261
276
        wt.commit('A', allow_pointless=True, rev_id='A')
262
277
        wt.branch.repository.sign_revision('A',
263
278
            bzrlib.gpg.LoopbackGPGStrategy(None))
264
279
        old_signature = wt.branch.repository.revision_store.get('A',
265
280
            'sig').read()
266
 
        upgrade(wt.basedir)
267
 
        wt = WorkingTree(wt.basedir)
 
281
        try:
 
282
            upgrade(wt.basedir)
 
283
        except errors.UpToDateFormat:
 
284
            # this is in the most current format already.
 
285
            return
 
286
        wt = WorkingTree.open(wt.basedir)
268
287
        new_signature = wt.branch.repository.revision_store.get('A',
269
288
            'sig').read()
270
289
        self.assertEqual(old_signature, new_signature)
297
316
    def test_commit_nicks(self):
298
317
        """Nicknames are committed to the revision"""
299
318
        get_transport(self.get_url()).mkdir('bzr.dev')
300
 
        branch = self.make_branch('bzr.dev')
 
319
        wt = self.make_branch_and_tree('bzr.dev')
 
320
        branch = wt.branch
301
321
        branch.nick = "My happy branch"
302
 
        WorkingTree.create(branch, 'bzr.dev').commit('My commit respect da nick.')
 
322
        wt.commit('My commit respect da nick.')
303
323
        committed = branch.repository.get_revision(branch.last_revision())
304
324
        self.assertEqual(committed.properties["branch-nick"], 
305
325
                         "My happy branch")
306
326
 
307
 
    def test_no_ancestry_weave(self):
308
 
        # We no longer need to create the ancestry.weave file
309
 
        # since it is *never* used.
310
 
        branch = Branch.create('.')
311
 
        self.failIfExists('.bzr/ancestry.weave')
 
327
    def test_create_open_branch_uses_repository(self):
 
328
        try:
 
329
            repo = self.make_repository('.', shared=True)
 
330
        except errors.IncompatibleFormat:
 
331
            return
 
332
        repo.bzrdir.root_transport.mkdir('child')
 
333
        child_dir = self.bzrdir_format.initialize('child')
 
334
        try:
 
335
            child_branch = self.branch_format.initialize(child_dir)
 
336
        except errors.UninitializableFormat:
 
337
            # branch references are not default init'able.
 
338
            return
 
339
        self.assertEqual(repo.bzrdir.root_transport.base,
 
340
                         child_branch.repository.bzrdir.root_transport.base)
 
341
        child_branch = bzrlib.branch.Branch.open(self.get_url('child'))
 
342
        self.assertEqual(repo.bzrdir.root_transport.base,
 
343
                         child_branch.repository.bzrdir.root_transport.base)
312
344
 
313
345
 
314
346
class ChrootedTests(TestCaseWithBranch):
329
361
                          self.get_readonly_url(''))
330
362
        self.assertRaises(NotBranchError, Branch.open_containing,
331
363
                          self.get_readonly_url('g/p/q'))
332
 
        try:
333
 
            branch = self.branch_format.initialize(self.get_url())
334
 
        except UninitializableFormat:
335
 
            raise TestSkipped("Format %s is not initializable.")
 
364
        branch = self.make_branch('.')
336
365
        branch, relpath = Branch.open_containing(self.get_readonly_url(''))
337
366
        self.assertEqual('', relpath)
338
367
        branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
511
540
        # supported formats must be able to init and open
512
541
        t = get_transport(self.get_url())
513
542
        readonly_t = get_transport(self.get_readonly_url())
514
 
        made_branch = self.branch_format.initialize(t.base)
515
 
        self.failUnless(isinstance(made_branch, branch.Branch))
 
543
        made_branch = self.make_branch('.')
 
544
        self.failUnless(isinstance(made_branch, bzrlib.branch.Branch))
 
545
 
 
546
        # find it via bzrdir opening:
 
547
        opened_control = bzrdir.BzrDir.open(readonly_t.base)
 
548
        direct_opened_branch = opened_control.open_branch()
 
549
        self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
 
550
        self.assertEqual(opened_control, direct_opened_branch.bzrdir)
 
551
        self.failUnless(isinstance(direct_opened_branch._format,
 
552
                        self.branch_format.__class__))
 
553
 
 
554
        # find it via Branch.open
 
555
        opened_branch = bzrlib.branch.Branch.open(readonly_t.base)
 
556
        self.failUnless(isinstance(opened_branch, made_branch.__class__))
 
557
        self.assertEqual(made_branch._format.__class__,
 
558
                         opened_branch._format.__class__)
 
559
        # if it has a unique id string, can we probe for it ?
 
560
        try:
 
561
            self.branch_format.get_format_string()
 
562
        except NotImplementedError:
 
563
            return
516
564
        self.assertEqual(self.branch_format,
517
 
                         branch.BzrBranchFormat.find_format(readonly_t))
518
 
        direct_opened_branch = self.branch_format.open(readonly_t)
519
 
        opened_branch = branch.Branch.open(t.base)
520
 
        self.assertEqual(made_branch._branch_format,
521
 
                         opened_branch._branch_format)
522
 
        self.assertEqual(direct_opened_branch._branch_format,
523
 
                         opened_branch._branch_format)
524
 
        self.failUnless(isinstance(opened_branch, branch.Branch))
525
 
 
526
 
    def test_open_not_branch(self):
527
 
        self.assertRaises(NoSuchFile,
528
 
                          self.branch_format.open,
529
 
                          get_transport(self.get_readonly_url()))
 
565
                         bzrlib.branch.BranchFormat.find_format(opened_control))