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
30
class DummyWeave(object):
31
"""A class that can be instantiated and compared."""
33
def __init__(self, message):
34
self._message = message
37
def __eq__(self, other):
40
return self._message == other._message
42
def transaction_finished(self):
46
class TestSymbols(TestCase):
48
def test_public_symbols(self):
30
class TestReadOnlyTransaction(TestCase):
32
def test_symbols(self):
49
33
from bzrlib.transactions import ReadOnlyTransaction
50
from bzrlib.transactions import PassThroughTransaction
53
class TestReadOnlyTransaction(TestCase):
56
self.transaction = transactions.ReadOnlyTransaction()
57
super(TestReadOnlyTransaction, self).setUp()
35
def test_construct(self):
36
transactions.ReadOnlyTransaction()
59
38
def test_register_clean(self):
60
self.transaction.register_clean("anobject")
39
transaction = transactions.ReadOnlyTransaction()
40
transaction.register_clean("anobject")
62
42
def test_register_dirty_raises(self):
63
self.assertRaises(errors.ReadOnlyError,
64
self.transaction.register_dirty,"anobject")
43
transaction = transactions.ReadOnlyTransaction()
44
self.assertRaises(errors.ReadOnlyError,
45
transaction.register_dirty,"anobject")
47
def test_commit_raises(self):
48
transaction = transactions.ReadOnlyTransaction()
49
self.assertRaises(errors.CommitNotPossible, transaction.commit)
66
51
def test_map(self):
67
self.assertNotEqual(None, getattr(self.transaction, "map", None))
52
transaction = transactions.ReadOnlyTransaction()
53
self.assertNotEqual(None, getattr(transaction, "map", None))
69
55
def test_add_and_get(self):
56
transaction = transactions.ReadOnlyTransaction()
71
self.transaction.map.add_weave("id", weave)
72
self.assertEqual(weave, self.transaction.map.find_weave("id"))
58
transaction.map.add_weave("id", weave)
59
self.assertEqual(weave, transaction.map.find_weave("id"))
61
def test_rollback_returns(self):
62
transaction = transactions.ReadOnlyTransaction()
63
transaction.rollback()
74
65
def test_finish_returns(self):
75
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
def test_zero_size_cache(self):
85
self.transaction.set_cache_size(0)
86
weave = DummyWeave('a weave')
87
self.transaction.map.add_weave("id", weave)
88
self.assertEqual(weave, self.transaction.map.find_weave("id"))
90
# add an object, should fall right out if there are no references
91
self.transaction.register_clean(self.transaction.map.find_weave("id"))
92
self.assertEqual(None, self.transaction.map.find_weave("id"))
93
# but if we have a reference it should stick around
94
weave = DummyWeave("another weave")
95
self.transaction.map.add_weave("id", weave)
96
self.transaction.register_clean(self.transaction.map.find_weave("id"))
97
self.assertEqual(weave, self.transaction.map.find_weave("id"))
99
# its not a weakref system
100
self.assertEqual(DummyWeave("another weave"),
101
self.transaction.map.find_weave("id"))
103
def test_small_cache(self):
104
self.transaction.set_cache_size(1)
105
# add an object, should not fall right out if there are no references
106
#sys.getrefcounts(foo)
107
self.transaction.map.add_weave("id", DummyWeave("a weave"))
108
self.transaction.register_clean(self.transaction.map.find_weave("id"))
109
self.assertEqual(DummyWeave("a weave"),
110
self.transaction.map.find_weave("id"))
111
self.transaction.map.add_weave("id2", DummyWeave("a weave also"))
112
self.transaction.register_clean(self.transaction.map.find_weave("id2"))
114
self.assertEqual(None, self.transaction.map.find_weave("id"))
115
self.assertEqual(DummyWeave("a weave also"),
116
self.transaction.map.find_weave("id2"))
118
def test_small_cache_with_references(self):
119
# if we have a reference it should stick around
121
weave2 = "another weave"
122
self.transaction.map.add_weave("id", weave)
123
self.transaction.map.add_weave("id2", weave2)
124
self.assertEqual(weave, self.transaction.map.find_weave("id"))
125
self.assertEqual(weave2, self.transaction.map.find_weave("id2"))
127
# its not a weakref system
128
self.assertEqual("a weave", self.transaction.map.find_weave("id"))
130
def test_precious_with_zero_size_cache(self):
131
self.transaction.set_cache_size(0)
132
weave = DummyWeave('a weave')
133
self.transaction.map.add_weave("id", weave)
134
self.assertEqual(weave, self.transaction.map.find_weave("id"))
136
# add an object, should not fall out even with no references.
137
self.transaction.register_clean(self.transaction.map.find_weave("id"),
139
self.assertEqual(DummyWeave('a weave'),
140
self.transaction.map.find_weave("id"))
142
def test_writable(self):
143
self.assertFalse(self.transaction.writeable())
66
transaction = transactions.ReadOnlyTransaction()
146
70
class TestPassThroughTransaction(TestCase):
72
def test_symbols(self):
73
from bzrlib.transactions import PassThroughTransaction
148
75
def test_construct(self):
149
76
transactions.PassThroughTransaction()
151
78
def test_register_clean(self):
152
79
transaction = transactions.PassThroughTransaction()
153
80
transaction.register_clean("anobject")
155
82
def test_register_dirty(self):
156
83
transaction = transactions.PassThroughTransaction()
157
84
transaction.register_dirty("anobject")
86
def test_commit_nothing_returns(self):
87
transaction = transactions.PassThroughTransaction()
159
90
def test_map(self):
160
91
transaction = transactions.PassThroughTransaction()
161
92
self.assertNotEqual(None, getattr(transaction, "map", None))
163
def test_add_and_get(self):
164
transaction = transactions.PassThroughTransaction()
166
transaction.map.add_weave("id", weave)
167
self.assertEqual(None, transaction.map.find_weave("id"))
169
def test_finish_returns(self):
170
transaction = transactions.PassThroughTransaction()
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
def test_cache_is_ignored(self):
183
transaction = transactions.PassThroughTransaction()
184
transaction.set_cache_size(100)
186
transaction.map.add_weave("id", weave)
187
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())
94
def test_add_and_get(self):
95
transaction = transactions.PassThroughTransaction()
97
transaction.map.add_weave("id", weave)
98
self.assertEqual(None, transaction.map.find_weave("id"))
100
def test_rollback_asserts(self):
101
transaction = transactions.PassThroughTransaction()
102
self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
104
def test_finish_returns(self):
105
transaction = transactions.PassThroughTransaction()