/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

Update news and readme

- better explanation of dependencies

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testbranch.py
 
1
# (C) 2005 Canonical Ltd
 
2
 
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
 
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
 
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
import os
 
18
 
 
19
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
 
20
from bzrlib.clone import copy_branch
 
21
from bzrlib.commit import commit
 
22
import bzrlib.errors as errors
 
23
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
 
24
import bzrlib.gpg
 
25
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
26
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
 
27
from bzrlib.trace import mutter
 
28
import bzrlib.transactions as transactions
 
29
 
 
30
# TODO: Make a branch using basis branch, and check that it 
 
31
# doesn't request any files that could have been avoided, by 
 
32
# hooking into the Transport.
 
33
 
 
34
class TestBranch(TestCaseInTempDir):
 
35
 
 
36
    def test_append_revisions(self):
 
37
        """Test appending more than one revision"""
 
38
        br = Branch.initialize(".")
 
39
        br.append_revision("rev1")
 
40
        self.assertEquals(br.revision_history(), ["rev1",])
 
41
        br.append_revision("rev2", "rev3")
 
42
        self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
 
43
 
 
44
    def test_fetch_revisions(self):
 
45
        """Test fetch-revision operation."""
 
46
        from bzrlib.fetch import Fetcher
 
47
        os.mkdir('b1')
 
48
        os.mkdir('b2')
 
49
        b1 = Branch.initialize('b1')
 
50
        b2 = Branch.initialize('b2')
 
51
        file(os.sep.join(['b1', 'foo']), 'w').write('hello')
 
52
        b1.add(['foo'], ['foo-id'])
 
53
        b1.commit('lala!', rev_id='revision-1', allow_pointless=False)
 
54
 
 
55
        mutter('start fetch')
 
56
        f = Fetcher(from_branch=b1, to_branch=b2)
 
57
        eq = self.assertEquals
 
58
        eq(f.count_copied, 1)
 
59
        eq(f.last_revision, 'revision-1')
 
60
 
 
61
        rev = b2.get_revision('revision-1')
 
62
        tree = b2.revision_tree('revision-1')
 
63
        eq(tree.get_file_text('foo-id'), 'hello')
 
64
 
 
65
    def test_push_stores(self):
 
66
        """Copy the stores from one branch to another"""
 
67
        os.mkdir('a')
 
68
        br_a = Branch.initialize("a")
 
69
        file('a/b', 'wb').write('b')
 
70
        br_a.add('b')
 
71
        commit(br_a, "silly commit")
 
72
 
 
73
        os.mkdir('b')
 
74
        br_b = Branch.initialize("b")
 
75
        self.assertRaises(NoSuchRevision, br_b.get_revision, 
 
76
                          br_a.revision_history()[0])
 
77
        br_a.push_stores(br_b)
 
78
        rev = br_b.get_revision(br_a.revision_history()[0])
 
79
        tree = br_b.revision_tree(br_a.revision_history()[0])
 
80
        for file_id in tree:
 
81
            if tree.inventory[file_id].kind == "file":
 
82
                tree.get_file(file_id).read()
 
83
        return br_a, br_b
 
84
 
 
85
    def test_copy_branch(self):
 
86
        """Copy the stores from one branch to another"""
 
87
        br_a, br_b = self.test_push_stores()
 
88
        commit(br_b, "silly commit")
 
89
        os.mkdir('c')
 
90
        br_c = copy_branch(br_a, 'c', basis_branch=br_b)
 
91
        self.assertEqual(br_a.revision_history(), br_c.revision_history())
 
92
 
 
93
    def test_copy_partial(self):
 
94
        """Copy only part of the history of a branch."""
 
95
        self.build_tree(['a/', 'a/one'])
 
96
        br_a = Branch.initialize('a')
 
97
        br_a.add(['one'])
 
98
        br_a.commit('commit one', rev_id='u@d-1')
 
99
        self.build_tree(['a/two'])
 
100
        br_a.add(['two'])
 
101
        br_a.commit('commit two', rev_id='u@d-2')
 
102
        br_b = copy_branch(br_a, 'b', revision='u@d-1')
 
103
        self.assertEqual(br_b.last_revision(), 'u@d-1')
 
104
        self.assertTrue(os.path.exists('b/one'))
 
105
        self.assertFalse(os.path.exists('b/two'))
 
106
        
 
107
 
 
108
    def test_record_initial_ghost_merge(self):
 
109
        """A pending merge with no revision present is still a merge."""
 
110
        branch = Branch.initialize('.')
 
111
        branch.add_pending_merge('non:existent@rev--ision--0--2')
 
112
        branch.commit('pretend to merge nonexistent-revision', rev_id='first')
 
113
        rev = branch.get_revision(branch.last_revision())
 
114
        self.assertEqual(len(rev.parent_ids), 1)
 
115
        # parent_sha1s is not populated now, WTF. rbc 20051003
 
116
        self.assertEqual(len(rev.parent_sha1s), 0)
 
117
        self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
 
118
 
 
119
# TODO 20051003 RBC:
 
120
# compare the gpg-to-sign info for a commit with a ghost and 
 
121
#     an identical tree without a ghost
 
122
# fetch missing should rewrite the TOC of weaves to list newly available parents.
 
123
        
 
124
    def test_pending_merges(self):
 
125
        """Tracking pending-merged revisions."""
 
126
        b = Branch.initialize('.')
 
127
 
 
128
        self.assertEquals(b.pending_merges(), [])
 
129
        b.add_pending_merge('foo@azkhazan-123123-abcabc')
 
130
        self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
 
131
        b.add_pending_merge('foo@azkhazan-123123-abcabc')
 
132
        self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
 
133
        b.add_pending_merge('wibble@fofof--20050401--1928390812')
 
134
        self.assertEquals(b.pending_merges(),
 
135
                          ['foo@azkhazan-123123-abcabc',
 
136
                           'wibble@fofof--20050401--1928390812'])
 
137
        b.commit("commit from base with two merges")
 
138
        rev = b.get_revision(b.revision_history()[0])
 
139
        self.assertEquals(len(rev.parent_ids), 2)
 
140
        self.assertEquals(rev.parent_ids[0],
 
141
                          'foo@azkhazan-123123-abcabc')
 
142
        self.assertEquals(rev.parent_ids[1],
 
143
                           'wibble@fofof--20050401--1928390812')
 
144
        # list should be cleared when we do a commit
 
145
        self.assertEquals(b.pending_merges(), [])
 
146
 
 
147
    def test_sign_existing_revision(self):
 
148
        branch = Branch.initialize('.')
 
149
        branch.commit("base", allow_pointless=True, rev_id='A')
 
150
        from bzrlib.testament import Testament
 
151
        branch.sign_revision('A', bzrlib.gpg.LoopbackGPGStrategy(None))
 
152
        self.assertEqual(Testament.from_revision(branch, 'A').as_short_text(),
 
153
                         branch.revision_store.get('A', 'sig').read())
 
154
 
 
155
    def test_store_signature(self):
 
156
        branch = Branch.initialize('.')
 
157
        branch.store_revision_signature(bzrlib.gpg.LoopbackGPGStrategy(None),
 
158
                                        'FOO', 'A')
 
159
        self.assertEqual('FOO', branch.revision_store.get('A', 'sig').read())
 
160
 
 
161
 
 
162
class TestRemote(TestCaseWithWebserver):
 
163
 
 
164
    def test_open_containing(self):
 
165
        self.assertRaises(NotBranchError, Branch.open_containing,
 
166
                          self.get_remote_url(''))
 
167
        self.assertRaises(NotBranchError, Branch.open_containing,
 
168
                          self.get_remote_url('g/p/q'))
 
169
        b = Branch.initialize('.')
 
170
        branch, relpath = Branch.open_containing(self.get_remote_url(''))
 
171
        self.assertEqual('', relpath)
 
172
        branch, relpath = Branch.open_containing(self.get_remote_url('g/p/q'))
 
173
        self.assertEqual('g/p/q', relpath)
 
174
        
 
175
# TODO: rewrite this as a regular unittest, without relying on the displayed output        
 
176
#         >>> from bzrlib.commit import commit
 
177
#         >>> bzrlib.trace.silent = True
 
178
#         >>> br1 = ScratchBranch(files=['foo', 'bar'])
 
179
#         >>> br1.add('foo')
 
180
#         >>> br1.add('bar')
 
181
#         >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
 
182
#         >>> br2 = ScratchBranch()
 
183
#         >>> br2.update_revisions(br1)
 
184
#         Added 2 texts.
 
185
#         Added 1 inventories.
 
186
#         Added 1 revisions.
 
187
#         >>> br2.revision_history()
 
188
#         [u'REVISION-ID-1']
 
189
#         >>> br2.update_revisions(br1)
 
190
#         Added 0 revisions.
 
191
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
 
192
#         True
 
193
 
 
194
class InstrumentedTransaction(object):
 
195
 
 
196
    def finish(self):
 
197
        self.calls.append('finish')
 
198
 
 
199
    def __init__(self):
 
200
        self.calls = []
 
201
 
 
202
 
 
203
class TestDecorator(object):
 
204
 
 
205
    def __init__(self):
 
206
        self._calls = []
 
207
 
 
208
    def lock_read(self):
 
209
        self._calls.append('lr')
 
210
 
 
211
    def lock_write(self):
 
212
        self._calls.append('lw')
 
213
 
 
214
    def unlock(self):
 
215
        self._calls.append('ul')
 
216
 
 
217
    @needs_read_lock
 
218
    def do_with_read(self):
 
219
        return 1
 
220
 
 
221
    @needs_read_lock
 
222
    def except_with_read(self):
 
223
        raise RuntimeError
 
224
 
 
225
    @needs_write_lock
 
226
    def do_with_write(self):
 
227
        return 2
 
228
 
 
229
    @needs_write_lock
 
230
    def except_with_write(self):
 
231
        raise RuntimeError
 
232
 
 
233
 
 
234
class TestDecorators(TestCase):
 
235
 
 
236
    def test_needs_read_lock(self):
 
237
        branch = TestDecorator()
 
238
        self.assertEqual(1, branch.do_with_read())
 
239
        self.assertEqual(['lr', 'ul'], branch._calls)
 
240
 
 
241
    def test_excepts_in_read_lock(self):
 
242
        branch = TestDecorator()
 
243
        self.assertRaises(RuntimeError, branch.except_with_read)
 
244
        self.assertEqual(['lr', 'ul'], branch._calls)
 
245
 
 
246
    def test_needs_write_lock(self):
 
247
        branch = TestDecorator()
 
248
        self.assertEqual(2, branch.do_with_write())
 
249
        self.assertEqual(['lw', 'ul'], branch._calls)
 
250
 
 
251
    def test_excepts_in_write_lock(self):
 
252
        branch = TestDecorator()
 
253
        self.assertRaises(RuntimeError, branch.except_with_write)
 
254
        self.assertEqual(['lw', 'ul'], branch._calls)
 
255
 
 
256
 
 
257
class TestBranchTransaction(TestCaseInTempDir):
 
258
 
 
259
    def setUp(self):
 
260
        super(TestBranchTransaction, self).setUp()
 
261
        self.branch = Branch.initialize('.')
 
262
        
 
263
    def test_default_get_transaction(self):
 
264
        """branch.get_transaction on a new branch should give a PassThrough."""
 
265
        self.failUnless(isinstance(self.branch.get_transaction(),
 
266
                                   transactions.PassThroughTransaction))
 
267
 
 
268
    def test__set_new_transaction(self):
 
269
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
270
 
 
271
    def test__set_over_existing_transaction_raises(self):
 
272
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
273
        self.assertRaises(errors.LockError,
 
274
                          self.branch._set_transaction,
 
275
                          transactions.ReadOnlyTransaction())
 
276
 
 
277
    def test_finish_no_transaction_raises(self):
 
278
        self.assertRaises(errors.LockError, self.branch._finish_transaction)
 
279
 
 
280
    def test_finish_readonly_transaction_works(self):
 
281
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
282
        self.branch._finish_transaction()
 
283
        self.assertEqual(None, self.branch._transaction)
 
284
 
 
285
    def test_unlock_calls_finish(self):
 
286
        self.branch.lock_read()
 
287
        transaction = InstrumentedTransaction()
 
288
        self.branch._transaction = transaction
 
289
        self.branch.unlock()
 
290
        self.assertEqual(['finish'], transaction.calls)
 
291
 
 
292
    def test_lock_read_acquires_ro_transaction(self):
 
293
        self.branch.lock_read()
 
294
        self.failUnless(isinstance(self.branch.get_transaction(),
 
295
                                   transactions.ReadOnlyTransaction))
 
296
        self.branch.unlock()
 
297
        
 
298
    def test_lock_write_acquires_passthrough_transaction(self):
 
299
        self.branch.lock_write()
 
300
        # cannot use get_transaction as its magic
 
301
        self.failUnless(isinstance(self.branch._transaction,
 
302
                                   transactions.PassThroughTransaction))
 
303
        self.branch.unlock()