← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransactions.py

- refactor handling of short option names

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransactions.py
23
23
 
24
24
#import bzrlib specific imports here
25
25
import bzrlib.errors as errors
26
 
from bzrlib.tests import TestCase, TestCaseInTempDir
 
26
from bzrlib.selftest import TestCase, TestCaseInTempDir
27
27
import bzrlib.transactions as transactions
28
28
 
29
29
 
32
32
 
33
33
    def __init__(self, message):
34
34
        self._message = message
35
 
        self.finished = False
36
35
 
37
36
    def __eq__(self, other):
38
37
        if other is None:
39
38
            return False
40
39
        return self._message == other._message
41
40
 
42
 
    def transaction_finished(self):
43
 
        self.finished = True
44
 
 
45
41
 
46
42
class TestSymbols(TestCase):
47
43
 
63
59
        self.assertRaises(errors.ReadOnlyError, 
64
60
                          self.transaction.register_dirty,"anobject")
65
61
    
 
62
    def test_commit_raises(self):
 
63
        self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
 
64
 
66
65
    def test_map(self):
67
66
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
68
67
    
71
70
        self.transaction.map.add_weave("id", weave)
72
71
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
73
72
 
 
73
    def test_rollback_returns(self):
 
74
        self.transaction.rollback()
 
75
 
74
76
    def test_finish_returns(self):
75
77
        self.transaction.finish()
76
78
 
77
 
    def test_finish_does_not_tell_versioned_file_finished(self):
78
 
        # read only transactions never write, so theres no
79
 
        # need to inform versioned files about finishing
80
 
        weave = DummyWeave('a weave')
81
 
        self.transaction.finish()
82
 
        self.assertFalse(weave.finished)
83
 
 
84
79
    def test_zero_size_cache(self):
85
80
        self.transaction.set_cache_size(0)
86
81
        weave = DummyWeave('a weave')
153
148
            self.transaction.map.find_revision_history(), precious=True)
154
149
        self.assertEqual([], self.transaction.map.find_revision_history())
155
150
 
156
 
    def test_writable(self):
157
 
        self.assertFalse(self.transaction.writeable())
158
 
 
159
151
 
160
152
class TestPassThroughTransaction(TestCase):
161
153
 
170
162
        transaction = transactions.PassThroughTransaction()
171
163
        transaction.register_dirty("anobject")
172
164
    
 
165
    def test_commit_nothing_returns(self):
 
166
        transaction = transactions.PassThroughTransaction()
 
167
        transaction.commit()
 
168
 
173
169
    def test_map(self):
174
170
        transaction = transactions.PassThroughTransaction()
175
171
        self.assertNotEqual(None, getattr(transaction, "map", None))
180
176
        transaction.map.add_weave("id", weave)
181
177
        self.assertEqual(None, transaction.map.find_weave("id"))
182
178
        
 
179
    def test_rollback_asserts(self):
 
180
        transaction = transactions.PassThroughTransaction()
 
181
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
 
182
 
183
183
    def test_finish_returns(self):
184
184
        transaction = transactions.PassThroughTransaction()
185
185
        transaction.finish()
186
186
 
187
 
    def test_finish_tells_versioned_file_finished(self):
188
 
        # pass through transactions allow writes so they
189
 
        # need to inform versioned files about finishing
190
 
        weave = DummyWeave('a weave')
191
 
        transaction = transactions.PassThroughTransaction()
192
 
        transaction.register_dirty(weave)
193
 
        transaction.finish()
194
 
        self.assertTrue(weave.finished)
195
 
 
196
187
    def test_cache_is_ignored(self):
197
188
        transaction = transactions.PassThroughTransaction()
198
189
        transaction.set_cache_size(100)
199
190
        weave = "a weave"
200
191
        transaction.map.add_weave("id", weave)
201
192
        self.assertEqual(None, transaction.map.find_weave("id"))
202
 
 
203
 
    def test_writable(self):
204
 
        transaction = transactions.PassThroughTransaction()
205
 
        self.assertTrue(transaction.writeable())
206
 
        
207
 
 
208
 
class TestWriteTransaction(TestCase):
209
 
 
210
 
    def setUp(self):
211
 
        self.transaction = transactions.WriteTransaction()
212
 
        super(TestWriteTransaction, self).setUp()
213
 
 
214
 
    def test_register_clean(self):
215
 
        self.transaction.register_clean("anobject")
216
 
    
217
 
    def test_register_dirty(self):
218
 
        self.transaction.register_dirty("anobject")
219
 
    
220
 
    def test_map(self):
221
 
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
222
 
    
223
 
    def test_add_and_get(self):
224
 
        weave = "a weave"
225
 
        self.transaction.map.add_weave("id", weave)
226
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
227
 
        
228
 
    def test_finish_returns(self):
229
 
        self.transaction.finish()
230
 
 
231
 
    def test_finish_tells_versioned_file_finished(self):
232
 
        # write transactions allow writes so they
233
 
        # need to inform versioned files about finishing
234
 
        weave = DummyWeave('a weave')
235
 
        self.transaction.register_dirty(weave)
236
 
        self.transaction.finish()
237
 
        self.assertTrue(weave.finished)
238
 
 
239
 
    def test_zero_size_cache(self):
240
 
        self.transaction.set_cache_size(0)
241
 
        # add an object, should fall right out if there are no references
242
 
        weave = DummyWeave('a weave')
243
 
        self.transaction.map.add_weave("id", weave)
244
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
245
 
        weave = None
246
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
247
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
248
 
        # but if we have a reference to a clean object it should stick around
249
 
        weave = DummyWeave("another weave")
250
 
        self.transaction.map.add_weave("id", weave)
251
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
252
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
253
 
        del weave
254
 
        # its not a weakref system
255
 
        self.assertEqual(DummyWeave("another weave"),
256
 
                         self.transaction.map.find_weave("id"))
257
 
 
258
 
    def test_zero_size_cache_dirty_objects(self):
259
 
        self.transaction.set_cache_size(0)
260
 
        # add a dirty object, which should not fall right out.
261
 
        weave = DummyWeave('a weave')
262
 
        self.transaction.map.add_weave("id", weave)
263
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
264
 
        weave = None
265
 
        self.transaction.register_dirty(self.transaction.map.find_weave("id"))
266
 
        self.assertNotEqual(None, self.transaction.map.find_weave("id"))
267
 
    
268
 
    def test_clean_to_dirty(self):
269
 
        # a clean object may become dirty.
270
 
        weave = DummyWeave('A weave')
271
 
        self.transaction.map.add_weave("id", weave)
272
 
        self.transaction.register_clean(weave)
273
 
        self.transaction.register_dirty(weave)
274
 
        self.assertTrue(self.transaction.is_dirty(weave))
275
 
        self.assertFalse(self.transaction.is_clean(weave))
276
 
 
277
 
    def test_small_cache(self):
278
 
        self.transaction.set_cache_size(1)
279
 
        # add an object, should not fall right out if there are no references
280
 
        #sys.getrefcounts(foo)
281
 
        self.transaction.map.add_weave("id", DummyWeave("a weave"))
282
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
283
 
        self.assertEqual(DummyWeave("a weave"),
284
 
                         self.transaction.map.find_weave("id"))
285
 
        self.transaction.map.add_weave("id2", DummyWeave("a weave also"))
286
 
        self.transaction.register_clean(self.transaction.map.find_weave("id2"))
287
 
        # currently a fifo
288
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
289
 
        self.assertEqual(DummyWeave("a weave also"),
290
 
                         self.transaction.map.find_weave("id2"))
291
 
 
292
 
    def test_small_cache_with_references(self):
293
 
        # if we have a reference it should stick around
294
 
        weave = "a weave"
295
 
        weave2 = "another weave"
296
 
        self.transaction.map.add_weave("id", weave)
297
 
        self.transaction.map.add_weave("id2", weave2)
298
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
299
 
        self.assertEqual(weave2, self.transaction.map.find_weave("id2"))
300
 
        weave = None
301
 
        # its not a weakref system
302
 
        self.assertEqual("a weave", self.transaction.map.find_weave("id"))
303
 
 
304
 
    def test_precious_with_zero_size_cache(self):
305
 
        self.transaction.set_cache_size(0)
306
 
        weave = DummyWeave('a weave')
307
 
        self.transaction.map.add_weave("id", weave)
308
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
309
 
        weave = None
310
 
        # add an object, should not fall out even with no references.
311
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"),
312
 
                                        precious=True)
313
 
        self.assertEqual(DummyWeave('a weave'),
314
 
                         self.transaction.map.find_weave("id"))
315
 
 
316
 
    def test_writable(self):
317
 
        self.assertTrue(self.transaction.writeable())