15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
from bzrlib.branch import Branch
19
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
19
20
from bzrlib.clone import copy_branch
20
21
from bzrlib.commit import commit
21
from bzrlib.errors import NoSuchRevision, UnlistableBranch
22
from bzrlib.selftest import TestCaseInTempDir
22
import bzrlib.errors as errors
23
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
25
from bzrlib.selftest import TestCase, TestCaseInTempDir
26
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
23
27
from bzrlib.trace import mutter
28
import bzrlib.transactions as transactions
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.
26
34
class TestBranch(TestCaseInTempDir):
54
62
tree = b2.revision_tree('revision-1')
55
63
eq(tree.get_file_text('foo-id'), 'hello')
57
def test_push_stores(self):
58
"""Copy the stores from one branch to another"""
65
def get_unbalanced_branch_pair(self):
66
"""Return two branches, a and b, with one file in a."""
60
68
br_a = Branch.initialize("a")
61
69
file('a/b', 'wb').write('b')
63
commit(br_a, "silly commit")
71
commit(br_a, "silly commit", rev_id='A')
66
73
br_b = Branch.initialize("b")
76
def get_balanced_branch_pair(self):
77
"""Returns br_a, br_b as with one commit in a, and b has a's stores."""
78
br_a, br_b = self.get_unbalanced_branch_pair()
79
br_a.push_stores(br_b)
82
def test_push_stores(self):
83
"""Copy the stores from one branch to another"""
84
br_a, br_b = self.get_unbalanced_branch_pair()
85
# ensure the revision is missing.
67
86
self.assertRaises(NoSuchRevision, br_b.get_revision,
68
87
br_a.revision_history()[0])
69
88
br_a.push_stores(br_b)
89
# check that b now has all the data from a's first commit.
70
90
rev = br_b.get_revision(br_a.revision_history()[0])
71
91
tree = br_b.revision_tree(br_a.revision_history()[0])
72
92
for file_id in tree:
77
97
def test_copy_branch(self):
78
98
"""Copy the stores from one branch to another"""
79
br_a, br_b = self.test_push_stores()
99
br_a, br_b = self.get_balanced_branch_pair()
80
100
commit(br_b, "silly commit")
82
102
br_c = copy_branch(br_a, 'c', basis_branch=br_b)
83
103
self.assertEqual(br_a.revision_history(), br_c.revision_history())
84
## # basis branches currently disabled for weave format
85
## self.assertFalse(br_b.last_revision() in br_c.revision_history())
86
## br_c.get_revision(br_b.last_revision())
88
105
def test_copy_partial(self):
89
106
"""Copy only part of the history of a branch."""
138
155
'wibble@fofof--20050401--1928390812')
139
156
# list should be cleared when we do a commit
140
157
self.assertEquals(b.pending_merges(), [])
159
def test_sign_existing_revision(self):
160
branch = Branch.initialize('.')
161
branch.commit("base", allow_pointless=True, rev_id='A')
162
from bzrlib.testament import Testament
163
branch.sign_revision('A', bzrlib.gpg.LoopbackGPGStrategy(None))
164
self.assertEqual(Testament.from_revision(branch, 'A').as_short_text(),
165
branch.revision_store.get('A', 'sig').read())
167
def test_store_signature(self):
168
branch = Branch.initialize('.')
169
branch.store_revision_signature(bzrlib.gpg.LoopbackGPGStrategy(None),
171
self.assertEqual('FOO', branch.revision_store.get('A', 'sig').read())
174
class TestRemote(TestCaseWithWebserver):
176
def test_open_containing(self):
177
self.assertRaises(NotBranchError, Branch.open_containing,
178
self.get_remote_url(''))
179
self.assertRaises(NotBranchError, Branch.open_containing,
180
self.get_remote_url('g/p/q'))
181
b = Branch.initialize('.')
182
branch, relpath = Branch.open_containing(self.get_remote_url(''))
183
self.assertEqual('', relpath)
184
branch, relpath = Branch.open_containing(self.get_remote_url('g/p/q'))
185
self.assertEqual('g/p/q', relpath)
143
187
# TODO: rewrite this as a regular unittest, without relying on the displayed output
144
188
# >>> from bzrlib.commit import commit
145
189
# >>> bzrlib.trace.silent = True
158
202
# Added 0 revisions.
159
203
# >>> br1.text_store.total_size() == br2.text_store.total_size()
206
class InstrumentedTransaction(object):
209
self.calls.append('finish')
215
class TestDecorator(object):
221
self._calls.append('lr')
223
def lock_write(self):
224
self._calls.append('lw')
227
self._calls.append('ul')
230
def do_with_read(self):
234
def except_with_read(self):
238
def do_with_write(self):
242
def except_with_write(self):
246
class TestDecorators(TestCase):
248
def test_needs_read_lock(self):
249
branch = TestDecorator()
250
self.assertEqual(1, branch.do_with_read())
251
self.assertEqual(['lr', 'ul'], branch._calls)
253
def test_excepts_in_read_lock(self):
254
branch = TestDecorator()
255
self.assertRaises(RuntimeError, branch.except_with_read)
256
self.assertEqual(['lr', 'ul'], branch._calls)
258
def test_needs_write_lock(self):
259
branch = TestDecorator()
260
self.assertEqual(2, branch.do_with_write())
261
self.assertEqual(['lw', 'ul'], branch._calls)
263
def test_excepts_in_write_lock(self):
264
branch = TestDecorator()
265
self.assertRaises(RuntimeError, branch.except_with_write)
266
self.assertEqual(['lw', 'ul'], branch._calls)
269
class TestBranchTransaction(TestCaseInTempDir):
272
super(TestBranchTransaction, self).setUp()
273
self.branch = Branch.initialize('.')
275
def test_default_get_transaction(self):
276
"""branch.get_transaction on a new branch should give a PassThrough."""
277
self.failUnless(isinstance(self.branch.get_transaction(),
278
transactions.PassThroughTransaction))
280
def test__set_new_transaction(self):
281
self.branch._set_transaction(transactions.ReadOnlyTransaction())
283
def test__set_over_existing_transaction_raises(self):
284
self.branch._set_transaction(transactions.ReadOnlyTransaction())
285
self.assertRaises(errors.LockError,
286
self.branch._set_transaction,
287
transactions.ReadOnlyTransaction())
289
def test_finish_no_transaction_raises(self):
290
self.assertRaises(errors.LockError, self.branch._finish_transaction)
292
def test_finish_readonly_transaction_works(self):
293
self.branch._set_transaction(transactions.ReadOnlyTransaction())
294
self.branch._finish_transaction()
295
self.assertEqual(None, self.branch._transaction)
297
def test_unlock_calls_finish(self):
298
self.branch.lock_read()
299
transaction = InstrumentedTransaction()
300
self.branch._transaction = transaction
302
self.assertEqual(['finish'], transaction.calls)
304
def test_lock_read_acquires_ro_transaction(self):
305
self.branch.lock_read()
306
self.failUnless(isinstance(self.branch.get_transaction(),
307
transactions.ReadOnlyTransaction))
310
def test_lock_write_acquires_passthrough_transaction(self):
311
self.branch.lock_write()
312
# cannot use get_transaction as its magic
313
self.failUnless(isinstance(self.branch._transaction,
314
transactions.PassThroughTransaction))
added
removed
bzrlib/selftest/testbranch.py
