← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransactions.py

  • Committer: John Arbash Meinel
  • Date: 2005-11-18 16:50:56 UTC
  • mto: (1185.82.108 w-changeset)
  • mto: This revision was merged to the branch mainline in revision 1738.
  • Revision ID: john@arbash-meinel.com-20051118165056-97f164d958a1886b
Moved testdata code to subdirectory

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransactions.py
1
 
# Copyright (C) 2005, 2006, 2009, 2011 Canonical Ltd
 
1
# Copyright (C) 2005 by 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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
 
20
24
#import bzrlib specific imports here
21
25
import bzrlib.errors as errors
22
 
from bzrlib.tests import TestCase
 
26
from bzrlib.selftest import TestCase, TestCaseInTempDir
23
27
import bzrlib.transactions as transactions
24
28
 
25
29
 
28
32
 
29
33
    def __init__(self, message):
30
34
        self._message = message
31
 
        self.finished = False
32
35
 
33
36
    def __eq__(self, other):
34
37
        if other is None:
35
38
            return False
36
39
        return self._message == other._message
37
40
 
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)
61
64
 
62
65
    def test_map(self):
63
66
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
64
 
 
 
67
    
65
68
    def test_add_and_get(self):
66
69
        weave = "a weave"
67
70
        self.transaction.map.add_weave("id", weave)
68
71
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
69
72
 
 
73
    def test_rollback_returns(self):
 
74
        self.transaction.rollback()
 
75
 
70
76
    def test_finish_returns(self):
71
77
        self.transaction.finish()
72
78
 
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
 
 
80
79
    def test_zero_size_cache(self):
81
80
        self.transaction.set_cache_size(0)
82
81
        weave = DummyWeave('a weave')
95
94
        # its not a weakref system
96
95
        self.assertEqual(DummyWeave("another weave"),
97
96
                         self.transaction.map.find_weave("id"))
98
 
 
 
97
        
99
98
    def test_small_cache(self):
100
99
        self.transaction.set_cache_size(1)
101
100
        # add an object, should not fall right out if there are no references
135
134
        self.assertEqual(DummyWeave('a weave'),
136
135
                         self.transaction.map.find_weave("id"))
137
136
 
138
 
    def test_writable(self):
139
 
        self.assertFalse(self.transaction.writeable())
 
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())
140
150
 
141
151
 
142
152
class TestPassThroughTransaction(TestCase):
147
157
    def test_register_clean(self):
148
158
        transaction = transactions.PassThroughTransaction()
149
159
        transaction.register_clean("anobject")
150
 
 
 
160
    
151
161
    def test_register_dirty(self):
152
162
        transaction = transactions.PassThroughTransaction()
153
163
        transaction.register_dirty("anobject")
 
164
    
 
165
    def test_commit_nothing_returns(self):
 
166
        transaction = transactions.PassThroughTransaction()
 
167
        transaction.commit()
154
168
 
155
169
    def test_map(self):
156
170
        transaction = transactions.PassThroughTransaction()
157
171
        self.assertNotEqual(None, getattr(transaction, "map", None))
158
 
 
 
172
    
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"))
 
178
        
 
179
    def test_rollback_asserts(self):
 
180
        transaction = transactions.PassThroughTransaction()
 
181
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
164
182
 
165
183
    def test_finish_returns(self):
166
184
        transaction = transactions.PassThroughTransaction()
167
185
        transaction.finish()
168
186
 
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
 
 
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"))
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())