60
56
self.transaction.register_clean("anobject")
62
58
def test_register_dirty_raises(self):
63
self.assertRaises(errors.ReadOnlyError,
59
self.assertRaises(errors.ReadOnlyError,
64
60
self.transaction.register_dirty,"anobject")
62
def test_commit_raises(self):
63
self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
66
65
def test_map(self):
67
66
self.assertNotEqual(None, getattr(self.transaction, "map", None))
69
68
def test_add_and_get(self):
71
70
self.transaction.map.add_weave("id", weave)
72
71
self.assertEqual(weave, self.transaction.map.find_weave("id"))
73
def test_rollback_returns(self):
74
self.transaction.rollback()
74
76
def test_finish_returns(self):
75
77
self.transaction.finish()
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)
84
79
def test_zero_size_cache(self):
85
80
self.transaction.set_cache_size(0)
86
81
weave = DummyWeave('a weave')
151
157
def test_register_clean(self):
152
158
transaction = transactions.PassThroughTransaction()
153
159
transaction.register_clean("anobject")
155
161
def test_register_dirty(self):
156
162
transaction = transactions.PassThroughTransaction()
157
163
transaction.register_dirty("anobject")
165
def test_commit_nothing_returns(self):
166
transaction = transactions.PassThroughTransaction()
159
169
def test_map(self):
160
170
transaction = transactions.PassThroughTransaction()
161
171
self.assertNotEqual(None, getattr(transaction, "map", None))
163
173
def test_add_and_get(self):
164
174
transaction = transactions.PassThroughTransaction()
165
175
weave = "a weave"
166
176
transaction.map.add_weave("id", weave)
167
177
self.assertEqual(None, transaction.map.find_weave("id"))
179
def test_rollback_asserts(self):
180
transaction = transactions.PassThroughTransaction()
181
self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
169
183
def test_finish_returns(self):
170
184
transaction = transactions.PassThroughTransaction()
171
185
transaction.finish()
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)
180
self.assertTrue(weave.finished)
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"))
189
def test_writable(self):
190
transaction = transactions.PassThroughTransaction()
191
self.assertTrue(transaction.writeable())
194
class TestWriteTransaction(TestCase):
197
self.transaction = transactions.WriteTransaction()
198
super(TestWriteTransaction, self).setUp()
200
def test_register_clean(self):
201
self.transaction.register_clean("anobject")
203
def test_register_dirty(self):
204
self.transaction.register_dirty("anobject")
207
self.assertNotEqual(None, getattr(self.transaction, "map", None))
209
def test_add_and_get(self):
211
self.transaction.map.add_weave("id", weave)
212
self.assertEqual(weave, self.transaction.map.find_weave("id"))
214
def test_finish_returns(self):
215
self.transaction.finish()
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)
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"))
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"))
240
# its not a weakref system
241
self.assertEqual(DummyWeave("another weave"),
242
self.transaction.map.find_weave("id"))
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"))
251
self.transaction.register_dirty(self.transaction.map.find_weave("id"))
252
self.assertNotEqual(None, self.transaction.map.find_weave("id"))
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))
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"))
274
self.assertEqual(None, self.transaction.map.find_weave("id"))
275
self.assertEqual(DummyWeave("a weave also"),
276
self.transaction.map.find_weave("id2"))
278
def test_small_cache_with_references(self):
279
# if we have a reference it should stick around
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"))
287
# its not a weakref system
288
self.assertEqual("a weave", self.transaction.map.find_weave("id"))
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"))
296
# add an object, should not fall out even with no references.
297
self.transaction.register_clean(self.transaction.map.find_weave("id"),
299
self.assertEqual(DummyWeave('a weave'),
300
self.transaction.map.find_weave("id"))
302
def test_writable(self):
303
self.assertTrue(self.transaction.writeable())