56
56
self.transaction.register_clean("anobject")
58
58
def test_register_dirty_raises(self):
59
self.assertRaises(errors.ReadOnlyError,
59
self.assertRaises(errors.ReadOnlyError,
60
60
self.transaction.register_dirty,"anobject")
62
def test_commit_raises(self):
63
self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
62
65
def test_map(self):
63
66
self.assertNotEqual(None, getattr(self.transaction, "map", None))
65
68
def test_add_and_get(self):
67
70
self.transaction.map.add_weave("id", weave)
68
71
self.assertEqual(weave, self.transaction.map.find_weave("id"))
73
def test_rollback_returns(self):
74
self.transaction.rollback()
70
76
def test_finish_returns(self):
71
77
self.transaction.finish()
73
def test_finish_does_not_tell_versioned_file_finished(self):
74
# read only transactions never write, so theres no
75
# need to inform versioned files about finishing
76
weave = DummyWeave('a weave')
77
self.transaction.finish()
78
self.assertFalse(weave.finished)
80
79
def test_zero_size_cache(self):
81
80
self.transaction.set_cache_size(0)
82
81
weave = DummyWeave('a weave')
147
157
def test_register_clean(self):
148
158
transaction = transactions.PassThroughTransaction()
149
159
transaction.register_clean("anobject")
151
161
def test_register_dirty(self):
152
162
transaction = transactions.PassThroughTransaction()
153
163
transaction.register_dirty("anobject")
165
def test_commit_nothing_returns(self):
166
transaction = transactions.PassThroughTransaction()
155
169
def test_map(self):
156
170
transaction = transactions.PassThroughTransaction()
157
171
self.assertNotEqual(None, getattr(transaction, "map", None))
159
173
def test_add_and_get(self):
160
174
transaction = transactions.PassThroughTransaction()
161
175
weave = "a weave"
162
176
transaction.map.add_weave("id", weave)
163
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)
165
183
def test_finish_returns(self):
166
184
transaction = transactions.PassThroughTransaction()
167
185
transaction.finish()
169
def test_finish_tells_versioned_file_finished(self):
170
# pass through transactions allow writes so they
171
# need to inform versioned files about finishing
172
weave = DummyWeave('a weave')
173
transaction = transactions.PassThroughTransaction()
174
transaction.register_dirty(weave)
176
self.assertTrue(weave.finished)
178
187
def test_cache_is_ignored(self):
179
188
transaction = transactions.PassThroughTransaction()
180
189
transaction.set_cache_size(100)
181
190
weave = "a weave"
182
191
transaction.map.add_weave("id", weave)
183
192
self.assertEqual(None, transaction.map.find_weave("id"))
185
def test_writable(self):
186
transaction = transactions.PassThroughTransaction()
187
self.assertTrue(transaction.writeable())
190
class TestWriteTransaction(TestCase):
193
self.transaction = transactions.WriteTransaction()
194
super(TestWriteTransaction, self).setUp()
196
def test_register_clean(self):
197
self.transaction.register_clean("anobject")
199
def test_register_dirty(self):
200
self.transaction.register_dirty("anobject")
203
self.assertNotEqual(None, getattr(self.transaction, "map", None))
205
def test_add_and_get(self):
207
self.transaction.map.add_weave("id", weave)
208
self.assertEqual(weave, self.transaction.map.find_weave("id"))
210
def test_finish_returns(self):
211
self.transaction.finish()
213
def test_finish_tells_versioned_file_finished(self):
214
# write transactions allow writes so they
215
# need to inform versioned files about finishing
216
weave = DummyWeave('a weave')
217
self.transaction.register_dirty(weave)
218
self.transaction.finish()
219
self.assertTrue(weave.finished)
221
def test_zero_size_cache(self):
222
self.transaction.set_cache_size(0)
223
# add an object, should fall right out if there are no references
224
weave = DummyWeave('a weave')
225
self.transaction.map.add_weave("id", weave)
226
self.assertEqual(weave, self.transaction.map.find_weave("id"))
228
self.transaction.register_clean(self.transaction.map.find_weave("id"))
229
self.assertEqual(None, self.transaction.map.find_weave("id"))
230
# but if we have a reference to a clean object it should stick around
231
weave = DummyWeave("another weave")
232
self.transaction.map.add_weave("id", weave)
233
self.transaction.register_clean(self.transaction.map.find_weave("id"))
234
self.assertEqual(weave, self.transaction.map.find_weave("id"))
236
# its not a weakref system
237
self.assertEqual(DummyWeave("another weave"),
238
self.transaction.map.find_weave("id"))
240
def test_zero_size_cache_dirty_objects(self):
241
self.transaction.set_cache_size(0)
242
# add a dirty object, which should not fall right out.
243
weave = DummyWeave('a weave')
244
self.transaction.map.add_weave("id", weave)
245
self.assertEqual(weave, self.transaction.map.find_weave("id"))
247
self.transaction.register_dirty(self.transaction.map.find_weave("id"))
248
self.assertNotEqual(None, self.transaction.map.find_weave("id"))
250
def test_clean_to_dirty(self):
251
# a clean object may become dirty.
252
weave = DummyWeave('A weave')
253
self.transaction.map.add_weave("id", weave)
254
self.transaction.register_clean(weave)
255
self.transaction.register_dirty(weave)
256
self.assertTrue(self.transaction.is_dirty(weave))
257
self.assertFalse(self.transaction.is_clean(weave))
259
def test_small_cache(self):
260
self.transaction.set_cache_size(1)
261
# add an object, should not fall right out if there are no references
262
#sys.getrefcounts(foo)
263
self.transaction.map.add_weave("id", DummyWeave("a weave"))
264
self.transaction.register_clean(self.transaction.map.find_weave("id"))
265
self.assertEqual(DummyWeave("a weave"),
266
self.transaction.map.find_weave("id"))
267
self.transaction.map.add_weave("id2", DummyWeave("a weave also"))
268
self.transaction.register_clean(self.transaction.map.find_weave("id2"))
270
self.assertEqual(None, self.transaction.map.find_weave("id"))
271
self.assertEqual(DummyWeave("a weave also"),
272
self.transaction.map.find_weave("id2"))
274
def test_small_cache_with_references(self):
275
# if we have a reference it should stick around
277
weave2 = "another weave"
278
self.transaction.map.add_weave("id", weave)
279
self.transaction.map.add_weave("id2", weave2)
280
self.assertEqual(weave, self.transaction.map.find_weave("id"))
281
self.assertEqual(weave2, self.transaction.map.find_weave("id2"))
283
# its not a weakref system
284
self.assertEqual("a weave", self.transaction.map.find_weave("id"))
286
def test_precious_with_zero_size_cache(self):
287
self.transaction.set_cache_size(0)
288
weave = DummyWeave('a weave')
289
self.transaction.map.add_weave("id", weave)
290
self.assertEqual(weave, self.transaction.map.find_weave("id"))
292
# add an object, should not fall out even with no references.
293
self.transaction.register_clean(self.transaction.map.find_weave("id"),
295
self.assertEqual(DummyWeave('a weave'),
296
self.transaction.map.find_weave("id"))
298
def test_writable(self):
299
self.assertTrue(self.transaction.writeable())