← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransactions.py

  • Committer: Robert Collins
  • Date: 2005-10-18 13:11:57 UTC
  • mfrom: (1185.16.72) (0.2.1)
  • Revision ID: robertc@robertcollins.net-20051018131157-76a9970aa78e927e
Merged Martin.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransactions.py
1
 
# Copyright (C) 2005 Canonical Ltd
 
1
# Copyright (C) 2005 by Canonical Ltd
2
2
#   Authors: Robert Collins <robert.collins@canonical.com>
3
3
#
4
4
# This program is free software; you can redistribute it and/or modify
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')
139
134
        self.assertEqual(DummyWeave('a weave'),
140
135
                         self.transaction.map.find_weave("id"))
141
136
 
142
 
    def test_writable(self):
143
 
        self.assertFalse(self.transaction.writeable())
 
137
    def test_precious_revision_history(self):
 
138
        """Disabled test until revision-history is a real object."""
 
139
        print "Disabled: test_precious_revision_history"
 
140
        return
 
141
        self.transaction.set_cache_size(0)
 
142
        history = []
 
143
        self.transaction.map.add_revision_history(history)
 
144
        self.assertEqual(history, self.transaction.map.find_revision_history())
 
145
        history = None
 
146
        # add an object, should not fall out even with no references.
 
147
        self.transaction.register_clean(
 
148
            self.transaction.map.find_revision_history(), precious=True)
 
149
        self.assertEqual([], self.transaction.map.find_revision_history())
144
150
 
145
151
 
146
152
class TestPassThroughTransaction(TestCase):
156
162
        transaction = transactions.PassThroughTransaction()
157
163
        transaction.register_dirty("anobject")
158
164
    
 
165
    def test_commit_nothing_returns(self):
 
166
        transaction = transactions.PassThroughTransaction()
 
167
        transaction.commit()
 
168
 
159
169
    def test_map(self):
160
170
        transaction = transactions.PassThroughTransaction()
161
171
        self.assertNotEqual(None, getattr(transaction, "map", None))
166
176
        transaction.map.add_weave("id", weave)
167
177
        self.assertEqual(None, transaction.map.find_weave("id"))
168
178
        
 
179
    def test_rollback_asserts(self):
 
180
        transaction = transactions.PassThroughTransaction()
 
181
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
 
182
 
169
183
    def test_finish_returns(self):
170
184
        transaction = transactions.PassThroughTransaction()
171
185
        transaction.finish()
172
186
 
173
 
    def test_finish_tells_versioned_file_finished(self):
174
 
        # pass through transactions allow writes so they
175
 
        # need to inform versioned files about finishing
176
 
        weave = DummyWeave('a weave')
177
 
        transaction = transactions.PassThroughTransaction()
178
 
        transaction.register_dirty(weave)
179
 
        transaction.finish()
180
 
        self.assertTrue(weave.finished)
181
 
 
182
187
    def test_cache_is_ignored(self):
183
188
        transaction = transactions.PassThroughTransaction()
184
189
        transaction.set_cache_size(100)
185
190
        weave = "a weave"
186
191
        transaction.map.add_weave("id", weave)
187
192
        self.assertEqual(None, transaction.map.find_weave("id"))
188
 
 
189
 
    def test_writable(self):
190
 
        transaction = transactions.PassThroughTransaction()
191
 
        self.assertTrue(transaction.writeable())
192
 
        
193
 
 
194
 
class TestWriteTransaction(TestCase):
195
 
 
196
 
    def setUp(self):
197
 
        self.transaction = transactions.WriteTransaction()
198
 
        super(TestWriteTransaction, self).setUp()
199
 
 
200
 
    def test_register_clean(self):
201
 
        self.transaction.register_clean("anobject")
202
 
    
203
 
    def test_register_dirty(self):
204
 
        self.transaction.register_dirty("anobject")
205
 
    
206
 
    def test_map(self):
207
 
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
208
 
    
209
 
    def test_add_and_get(self):
210
 
        weave = "a weave"
211
 
        self.transaction.map.add_weave("id", weave)
212
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
213
 
        
214
 
    def test_finish_returns(self):
215
 
        self.transaction.finish()
216
 
 
217
 
    def test_finish_tells_versioned_file_finished(self):
218
 
        # write transactions allow writes so they
219
 
        # need to inform versioned files about finishing
220
 
        weave = DummyWeave('a weave')
221
 
        self.transaction.register_dirty(weave)
222
 
        self.transaction.finish()
223
 
        self.assertTrue(weave.finished)
224
 
 
225
 
    def test_zero_size_cache(self):
226
 
        self.transaction.set_cache_size(0)
227
 
        # add an object, should fall right out if there are no references
228
 
        weave = DummyWeave('a weave')
229
 
        self.transaction.map.add_weave("id", weave)
230
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
231
 
        weave = None
232
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
233
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
234
 
        # but if we have a reference to a clean object it should stick around
235
 
        weave = DummyWeave("another weave")
236
 
        self.transaction.map.add_weave("id", weave)
237
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
238
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
239
 
        del weave
240
 
        # its not a weakref system
241
 
        self.assertEqual(DummyWeave("another weave"),
242
 
                         self.transaction.map.find_weave("id"))
243
 
 
244
 
    def test_zero_size_cache_dirty_objects(self):
245
 
        self.transaction.set_cache_size(0)
246
 
        # add a dirty object, which should not fall right out.
247
 
        weave = DummyWeave('a weave')
248
 
        self.transaction.map.add_weave("id", weave)
249
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
250
 
        weave = None
251
 
        self.transaction.register_dirty(self.transaction.map.find_weave("id"))
252
 
        self.assertNotEqual(None, self.transaction.map.find_weave("id"))
253
 
    
254
 
    def test_clean_to_dirty(self):
255
 
        # a clean object may become dirty.
256
 
        weave = DummyWeave('A weave')
257
 
        self.transaction.map.add_weave("id", weave)
258
 
        self.transaction.register_clean(weave)
259
 
        self.transaction.register_dirty(weave)
260
 
        self.assertTrue(self.transaction.is_dirty(weave))
261
 
        self.assertFalse(self.transaction.is_clean(weave))
262
 
 
263
 
    def test_small_cache(self):
264
 
        self.transaction.set_cache_size(1)
265
 
        # add an object, should not fall right out if there are no references
266
 
        #sys.getrefcounts(foo)
267
 
        self.transaction.map.add_weave("id", DummyWeave("a weave"))
268
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"))
269
 
        self.assertEqual(DummyWeave("a weave"),
270
 
                         self.transaction.map.find_weave("id"))
271
 
        self.transaction.map.add_weave("id2", DummyWeave("a weave also"))
272
 
        self.transaction.register_clean(self.transaction.map.find_weave("id2"))
273
 
        # currently a fifo
274
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
275
 
        self.assertEqual(DummyWeave("a weave also"),
276
 
                         self.transaction.map.find_weave("id2"))
277
 
 
278
 
    def test_small_cache_with_references(self):
279
 
        # if we have a reference it should stick around
280
 
        weave = "a weave"
281
 
        weave2 = "another weave"
282
 
        self.transaction.map.add_weave("id", weave)
283
 
        self.transaction.map.add_weave("id2", weave2)
284
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
285
 
        self.assertEqual(weave2, self.transaction.map.find_weave("id2"))
286
 
        weave = None
287
 
        # its not a weakref system
288
 
        self.assertEqual("a weave", self.transaction.map.find_weave("id"))
289
 
 
290
 
    def test_precious_with_zero_size_cache(self):
291
 
        self.transaction.set_cache_size(0)
292
 
        weave = DummyWeave('a weave')
293
 
        self.transaction.map.add_weave("id", weave)
294
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
295
 
        weave = None
296
 
        # add an object, should not fall out even with no references.
297
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"),
298
 
                                        precious=True)
299
 
        self.assertEqual(DummyWeave('a weave'),
300
 
                         self.transaction.map.find_weave("id"))
301
 
 
302
 
    def test_writable(self):
303
 
        self.assertTrue(self.transaction.writeable())