← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transactions.py

  • Committer: Aaron Bentley
  • Date: 2006-11-10 01:55:55 UTC
  • mto: This revision was merged to the branch mainline in revision 2127.
  • Revision ID: aaron.bentley@utoronto.ca-20061110015555-f48202744b630209
Ignore html docs (both kinds)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transactions.py
1
 
# Copyright (C) 2005 by Canonical Ltd
 
1
# Copyright (C) 2005 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.selftest import TestCase, TestCaseInTempDir
 
26
from bzrlib.tests 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
35
36
 
36
37
    def __eq__(self, other):
37
38
        if other is None:
38
39
            return False
39
40
        return self._message == other._message
40
41
 
 
42
    def transaction_finished(self):
 
43
        self.finished = True
 
44
 
41
45
 
42
46
class TestSymbols(TestCase):
43
47
 
59
63
        self.assertRaises(errors.ReadOnlyError, 
60
64
                          self.transaction.register_dirty,"anobject")
61
65
    
62
 
    def test_commit_raises(self):
63
 
        self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
64
 
 
65
66
    def test_map(self):
66
67
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
67
68
    
70
71
        self.transaction.map.add_weave("id", weave)
71
72
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
72
73
 
73
 
    def test_rollback_returns(self):
74
 
        self.transaction.rollback()
75
 
 
76
74
    def test_finish_returns(self):
77
75
        self.transaction.finish()
78
76
 
 
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
 
79
84
    def test_zero_size_cache(self):
80
85
        self.transaction.set_cache_size(0)
81
86
        weave = DummyWeave('a weave')
134
139
        self.assertEqual(DummyWeave('a weave'),
135
140
                         self.transaction.map.find_weave("id"))
136
141
 
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())
 
142
    def test_writable(self):
 
143
        self.assertFalse(self.transaction.writeable())
150
144
 
151
145
 
152
146
class TestPassThroughTransaction(TestCase):
162
156
        transaction = transactions.PassThroughTransaction()
163
157
        transaction.register_dirty("anobject")
164
158
    
165
 
    def test_commit_nothing_returns(self):
166
 
        transaction = transactions.PassThroughTransaction()
167
 
        transaction.commit()
168
 
 
169
159
    def test_map(self):
170
160
        transaction = transactions.PassThroughTransaction()
171
161
        self.assertNotEqual(None, getattr(transaction, "map", None))
176
166
        transaction.map.add_weave("id", weave)
177
167
        self.assertEqual(None, transaction.map.find_weave("id"))
178
168
        
179
 
    def test_rollback_asserts(self):
180
 
        transaction = transactions.PassThroughTransaction()
181
 
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
182
 
 
183
169
    def test_finish_returns(self):
184
170
        transaction = transactions.PassThroughTransaction()
185
171
        transaction.finish()
186
172
 
 
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
 
187
182
    def test_cache_is_ignored(self):
188
183
        transaction = transactions.PassThroughTransaction()
189
184
        transaction.set_cache_size(100)
190
185
        weave = "a weave"
191
186
        transaction.map.add_weave("id", weave)
192
187
        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())