← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

[merge] from robert and newformat

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testbranch.py
18
18
from bzrlib.branch import Branch
19
19
from bzrlib.clone import copy_branch
20
20
from bzrlib.commit import commit
21
 
from bzrlib.errors import NoSuchRevision, UnlistableBranch
 
21
import bzrlib.errors as errors
 
22
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
22
23
from bzrlib.selftest import TestCaseInTempDir
23
24
from bzrlib.trace import mutter
24
 
 
 
25
import bzrlib.transactions as transactions
 
26
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
25
27
 
26
28
class TestBranch(TestCaseInTempDir):
27
29
 
138
140
                           'wibble@fofof--20050401--1928390812')
139
141
        # list should be cleared when we do a commit
140
142
        self.assertEquals(b.pending_merges(), [])
141
 
 
142
 
 
 
143
 
 
144
 
 
145
class TestRemote(TestCaseWithWebserver):
 
146
 
 
147
    def test_open_containing(self):
 
148
        self.assertRaises(NotBranchError, Branch.open_containing,
 
149
                          self.get_remote_url(''))
 
150
        self.assertRaises(NotBranchError, Branch.open_containing,
 
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'))
 
155
        
143
156
# TODO: rewrite this as a regular unittest, without relying on the displayed output        
144
157
#         >>> from bzrlib.commit import commit
145
158
#         >>> bzrlib.trace.silent = True
158
171
#         Added 0 revisions.
159
172
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
160
173
#         True
 
174
 
 
175
class InstrumentedTransaction(object):
 
176
 
 
177
    def finish(self):
 
178
        self.calls.append('finish')
 
179
 
 
180
    def __init__(self):
 
181
        self.calls = []
 
182
 
 
183
 
 
184
class TestBranchTransaction(TestCaseInTempDir):
 
185
 
 
186
    def setUp(self):
 
187
        super(TestBranchTransaction, self).setUp()
 
188
        self.branch = Branch.initialize('.')
 
189
        
 
190
    def test_default_get_transaction(self):
 
191
        """branch.get_transaction on a new branch should give a PassThrough."""
 
192
        self.failUnless(isinstance(self.branch.get_transaction(),
 
193
                                   transactions.PassThroughTransaction))
 
194
 
 
195
    def test__set_new_transaction(self):
 
196
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
197
 
 
198
    def test__set_over_existing_transaction_raises(self):
 
199
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
200
        self.assertRaises(errors.LockError,
 
201
                          self.branch._set_transaction,
 
202
                          transactions.ReadOnlyTransaction())
 
203
 
 
204
    def test_finish_no_transaction_raises(self):
 
205
        self.assertRaises(errors.LockError, self.branch._finish_transaction)
 
206
 
 
207
    def test_finish_readonly_transaction_works(self):
 
208
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
209
        self.branch._finish_transaction()
 
210
        self.assertEqual(None, self.branch._transaction)
 
211
 
 
212
    def test_unlock_calls_finish(self):
 
213
        self.branch.lock_read()
 
214
        transaction = InstrumentedTransaction()
 
215
        self.branch._transaction = transaction
 
216
        self.branch.unlock()
 
217
        self.assertEqual(['finish'], transaction.calls)
 
218
 
 
219
    def test_lock_read_acquires_ro_transaction(self):
 
220
        self.branch.lock_read()
 
221
        self.failUnless(isinstance(self.branch.get_transaction(),
 
222
                                   transactions.ReadOnlyTransaction))
 
223
        self.branch.unlock()
 
224
        
 
225
    def test_lock_write_acquires_passthrough_transaction(self):
 
226
        self.branch.lock_write()
 
227
        # cannot use get_transaction as its magic
 
228
        self.failUnless(isinstance(self.branch._transaction,
 
229
                                   transactions.PassThroughTransaction))
 
230
        self.branch.unlock()