18
18
from bzrlib.branch import Branch
19
19
from bzrlib.clone import copy_branch
20
20
from bzrlib.commit import commit
21
import bzrlib.errors as errors
21
22
from bzrlib.errors import NoSuchRevision, UnlistableBranch
22
23
from bzrlib.selftest import TestCaseInTempDir
23
24
from bzrlib.trace import mutter
25
import bzrlib.transactions as transactions
26
28
class TestBranch(TestCaseInTempDir):
158
160
# Added 0 revisions.
159
161
# >>> br1.text_store.total_size() == br2.text_store.total_size()
164
class InstrumentedTransaction(object):
167
self.calls.append('finish')
173
class TestBranchTransaction(TestCaseInTempDir):
176
super(TestBranchTransaction, self).setUp()
177
self.branch = Branch.initialize('.')
179
def test_default_get_transaction(self):
180
"""branch.get_transaction on a new branch should give a PassThrough."""
181
self.failUnless(isinstance(self.branch.get_transaction(),
182
transactions.PassThroughTransaction))
184
def test__set_new_transaction(self):
185
self.branch._set_transaction(transactions.ReadOnlyTransaction())
187
def test__set_over_existing_transaction_raises(self):
188
self.branch._set_transaction(transactions.ReadOnlyTransaction())
189
self.assertRaises(errors.LockError,
190
self.branch._set_transaction,
191
transactions.ReadOnlyTransaction())
193
def test_finish_no_transaction_raises(self):
194
self.assertRaises(errors.LockError, self.branch._finish_transaction)
196
def test_finish_readonly_transaction_works(self):
197
self.branch._set_transaction(transactions.ReadOnlyTransaction())
198
self.branch._finish_transaction()
199
self.assertEqual(None, self.branch._transaction)
201
def test_unlock_calls_finish(self):
202
self.branch.lock_read()
203
transaction = InstrumentedTransaction()
204
self.branch._transaction = transaction
206
self.assertEqual(['finish'], transaction.calls)
208
def test_lock_read_acquires_ro_transaction(self):
209
self.branch.lock_read()
210
self.failUnless(isinstance(self.branch.get_transaction(),
211
transactions.ReadOnlyTransaction))
214
def test_lock_write_acquires_passthrough_transaction(self):
215
self.branch.lock_write()
216
# cannot use get_transaction as its magic
217
self.failUnless(isinstance(self.branch._transaction,
218
transactions.PassThroughTransaction))