← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transactions.py

(vila) Fix test failures blocking package builds. (Vincent Ladeuil)

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, 2006, 2009, 2011 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
13
13
#
14
14
# You should have received a copy of the GNU General Public License
15
15
# along with this program; if not, write to the Free Software
16
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
 
18
18
"""Tests for the behaviour of the Transaction concept in bzr."""
19
19
 
20
 
# import system imports here
21
 
import os
22
 
import sys
23
 
 
24
20
#import bzrlib specific imports here
25
21
import bzrlib.errors as errors
26
 
from bzrlib.tests import TestCase, TestCaseInTempDir
 
22
from bzrlib.tests import TestCase
27
23
import bzrlib.transactions as transactions
28
24
 
29
25
 
32
28
 
33
29
    def __init__(self, message):
34
30
        self._message = message
 
31
        self.finished = False
35
32
 
36
33
    def __eq__(self, other):
37
34
        if other is None:
38
35
            return False
39
36
        return self._message == other._message
40
37
 
 
38
    def transaction_finished(self):
 
39
        self.finished = True
 
40
 
41
41
 
42
42
class TestSymbols(TestCase):
43
43
 
56
56
        self.transaction.register_clean("anobject")
57
57
 
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")
61
 
    
62
 
    def test_commit_raises(self):
63
 
        self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
64
61
 
65
62
    def test_map(self):
66
63
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
67
 
    
 
64
 
68
65
    def test_add_and_get(self):
69
66
        weave = "a weave"
70
67
        self.transaction.map.add_weave("id", weave)
71
68
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
72
69
 
73
 
    def test_rollback_returns(self):
74
 
        self.transaction.rollback()
75
 
 
76
70
    def test_finish_returns(self):
77
71
        self.transaction.finish()
78
72
 
 
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)
 
79
 
79
80
    def test_zero_size_cache(self):
80
81
        self.transaction.set_cache_size(0)
81
82
        weave = DummyWeave('a weave')
94
95
        # its not a weakref system
95
96
        self.assertEqual(DummyWeave("another weave"),
96
97
                         self.transaction.map.find_weave("id"))
97
 
        
 
98
 
98
99
    def test_small_cache(self):
99
100
        self.transaction.set_cache_size(1)
100
101
        # add an object, should not fall right out if there are no references
134
135
        self.assertEqual(DummyWeave('a weave'),
135
136
                         self.transaction.map.find_weave("id"))
136
137
 
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())
 
138
    def test_writable(self):
 
139
        self.assertFalse(self.transaction.writeable())
150
140
 
151
141
 
152
142
class TestPassThroughTransaction(TestCase):
157
147
    def test_register_clean(self):
158
148
        transaction = transactions.PassThroughTransaction()
159
149
        transaction.register_clean("anobject")
160
 
    
 
150
 
161
151
    def test_register_dirty(self):
162
152
        transaction = transactions.PassThroughTransaction()
163
153
        transaction.register_dirty("anobject")
164
 
    
165
 
    def test_commit_nothing_returns(self):
166
 
        transaction = transactions.PassThroughTransaction()
167
 
        transaction.commit()
168
154
 
169
155
    def test_map(self):
170
156
        transaction = transactions.PassThroughTransaction()
171
157
        self.assertNotEqual(None, getattr(transaction, "map", None))
172
 
    
 
158
 
173
159
    def test_add_and_get(self):
174
160
        transaction = transactions.PassThroughTransaction()
175
161
        weave = "a weave"
176
162
        transaction.map.add_weave("id", weave)
177
163
        self.assertEqual(None, transaction.map.find_weave("id"))
178
 
        
179
 
    def test_rollback_asserts(self):
180
 
        transaction = transactions.PassThroughTransaction()
181
 
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
182
164
 
183
165
    def test_finish_returns(self):
184
166
        transaction = transactions.PassThroughTransaction()
185
167
        transaction.finish()
186
168
 
 
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)
 
175
        transaction.finish()
 
176
        self.assertTrue(weave.finished)
 
177
 
187
178
    def test_cache_is_ignored(self):
188
179
        transaction = transactions.PassThroughTransaction()
189
180
        transaction.set_cache_size(100)
190
181
        weave = "a weave"
191
182
        transaction.map.add_weave("id", weave)
192
183
        self.assertEqual(None, transaction.map.find_weave("id"))
 
184
 
 
185
    def test_writable(self):
 
186
        transaction = transactions.PassThroughTransaction()
 
187
        self.assertTrue(transaction.writeable())
 
188
 
 
189
 
 
190
class TestWriteTransaction(TestCase):
 
191
 
 
192
    def setUp(self):
 
193
        self.transaction = transactions.WriteTransaction()
 
194
        super(TestWriteTransaction, self).setUp()
 
195
 
 
196
    def test_register_clean(self):
 
197
        self.transaction.register_clean("anobject")
 
198
 
 
199
    def test_register_dirty(self):
 
200
        self.transaction.register_dirty("anobject")
 
201
 
 
202
    def test_map(self):
 
203
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
 
204
 
 
205
    def test_add_and_get(self):
 
206
        weave = "a weave"
 
207
        self.transaction.map.add_weave("id", weave)
 
208
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
 
209
 
 
210
    def test_finish_returns(self):
 
211
        self.transaction.finish()
 
212
 
 
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)
 
220
 
 
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"))
 
227
        weave = None
 
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"))
 
235
        del weave
 
236
        # its not a weakref system
 
237
        self.assertEqual(DummyWeave("another weave"),
 
238
                         self.transaction.map.find_weave("id"))
 
239
 
 
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"))
 
246
        weave = None
 
247
        self.transaction.register_dirty(self.transaction.map.find_weave("id"))
 
248
        self.assertNotEqual(None, self.transaction.map.find_weave("id"))
 
249
 
 
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))
 
258
 
 
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"))
 
269
        # currently a fifo
 
270
        self.assertEqual(None, self.transaction.map.find_weave("id"))
 
271
        self.assertEqual(DummyWeave("a weave also"),
 
272
                         self.transaction.map.find_weave("id2"))
 
273
 
 
274
    def test_small_cache_with_references(self):
 
275
        # if we have a reference it should stick around
 
276
        weave = "a weave"
 
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"))
 
282
        weave = None
 
283
        # its not a weakref system
 
284
        self.assertEqual("a weave", self.transaction.map.find_weave("id"))
 
285
 
 
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"))
 
291
        weave = None
 
292
        # add an object, should not fall out even with no references.
 
293
        self.transaction.register_clean(self.transaction.map.find_weave("id"),
 
294
                                        precious=True)
 
295
        self.assertEqual(DummyWeave('a weave'),
 
296
                         self.transaction.map.find_weave("id"))
 
297
 
 
298
    def test_writable(self):
 
299
        self.assertTrue(self.transaction.writeable())