← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransactions.py

  • Committer: Robert Collins
  • Date: 2005-10-09 23:12:35 UTC
  • Revision ID: robertc@robertcollins.net-20051009231235-93626e72cac71b78
clean up test dirs on make clean

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransactions.py
1
 
# Copyright (C) 2005 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
 
23
23
 
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
28
28
 
29
29
 
30
 
class DummyWeave(object):
31
 
    """A class that can be instantiated and compared."""
32
 
 
33
 
    def __init__(self, message):
34
 
        self._message = message
35
 
        self.finished = False
36
 
 
37
 
    def __eq__(self, other):
38
 
        if other is None:
39
 
            return False
40
 
        return self._message == other._message
41
 
 
42
 
    def transaction_finished(self):
43
 
        self.finished = True
44
 
 
45
 
 
46
 
class TestSymbols(TestCase):
47
 
 
48
 
    def test_public_symbols(self):
 
30
class TestReadOnlyTransaction(TestCase):
 
31
 
 
32
    def test_symbols(self):
49
33
        from bzrlib.transactions import ReadOnlyTransaction
50
 
        from bzrlib.transactions import PassThroughTransaction
51
 
 
52
 
 
53
 
class TestReadOnlyTransaction(TestCase):
54
 
 
55
 
    def setUp(self):
56
 
        self.transaction = transactions.ReadOnlyTransaction()
57
 
        super(TestReadOnlyTransaction, self).setUp()
 
34
 
 
35
    def test_construct(self):
 
36
        transactions.ReadOnlyTransaction()
58
37
 
59
38
    def test_register_clean(self):
60
 
        self.transaction.register_clean("anobject")
 
39
        transaction = transactions.ReadOnlyTransaction()
 
40
        transaction.register_clean("anobject")
61
41
 
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")
 
46
    
 
47
    def test_commit_raises(self):
 
48
        transaction = transactions.ReadOnlyTransaction()
 
49
        self.assertRaises(errors.CommitNotPossible, transaction.commit)
65
50
 
66
51
    def test_map(self):
67
 
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
68
 
 
 
52
        transaction = transactions.ReadOnlyTransaction()
 
53
        self.assertNotEqual(None, getattr(transaction, "map", None))
 
54
    
69
55
    def test_add_and_get(self):
 
56
        transaction = transactions.ReadOnlyTransaction()
70
57
        weave = "a weave"
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"))
 
60
 
 
61
    def test_rollback_returns(self):
 
62
        transaction = transactions.ReadOnlyTransaction()
 
63
        transaction.rollback()
73
64
 
74
65
    def test_finish_returns(self):
75
 
        self.transaction.finish()
76
 
 
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)
83
 
 
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"))
89
 
        weave = None
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"))
98
 
        del weave
99
 
        # its not a weakref system
100
 
        self.assertEqual(DummyWeave("another weave"),
101
 
                         self.transaction.map.find_weave("id"))
102
 
 
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"))
113
 
        # currently a fifo
114
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
115
 
        self.assertEqual(DummyWeave("a weave also"),
116
 
                         self.transaction.map.find_weave("id2"))
117
 
 
118
 
    def test_small_cache_with_references(self):
119
 
        # if we have a reference it should stick around
120
 
        weave = "a weave"
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"))
126
 
        weave = None
127
 
        # its not a weakref system
128
 
        self.assertEqual("a weave", self.transaction.map.find_weave("id"))
129
 
 
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"))
135
 
        weave = None
136
 
        # add an object, should not fall out even with no references.
137
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"),
138
 
                                        precious=True)
139
 
        self.assertEqual(DummyWeave('a weave'),
140
 
                         self.transaction.map.find_weave("id"))
141
 
 
142
 
    def test_writable(self):
143
 
        self.assertFalse(self.transaction.writeable())
144
 
 
145
 
 
 
66
        transaction = transactions.ReadOnlyTransaction()
 
67
        transaction.finish()
 
68
 
 
69
        
146
70
class TestPassThroughTransaction(TestCase):
147
71
 
 
72
    def test_symbols(self):
 
73
        from bzrlib.transactions import PassThroughTransaction
 
74
 
148
75
    def test_construct(self):
149
76
        transactions.PassThroughTransaction()
150
77
 
151
78
    def test_register_clean(self):
152
79
        transaction = transactions.PassThroughTransaction()
153
80
        transaction.register_clean("anobject")
154
 
 
 
81
    
155
82
    def test_register_dirty(self):
156
83
        transaction = transactions.PassThroughTransaction()
157
84
        transaction.register_dirty("anobject")
 
85
    
 
86
    def test_commit_nothing_returns(self):
 
87
        transaction = transactions.PassThroughTransaction()
 
88
        transaction.commit()
158
89
 
159
90
    def test_map(self):
160
91
        transaction = transactions.PassThroughTransaction()
161
92
        self.assertNotEqual(None, getattr(transaction, "map", None))
162
 
 
163
 
    def test_add_and_get(self):
164
 
        transaction = transactions.PassThroughTransaction()
165
 
        weave = "a weave"
166
 
        transaction.map.add_weave("id", weave)
167
 
        self.assertEqual(None, transaction.map.find_weave("id"))
168
 
 
169
 
    def test_finish_returns(self):
170
 
        transaction = transactions.PassThroughTransaction()
171
 
        transaction.finish()
172
 
 
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)
179
 
        transaction.finish()
180
 
        self.assertTrue(weave.finished)
181
 
 
182
 
    def test_cache_is_ignored(self):
183
 
        transaction = transactions.PassThroughTransaction()
184
 
        transaction.set_cache_size(100)
185
 
        weave = "a weave"
186
 
        transaction.map.add_weave("id", weave)
187
 
        self.assertEqual(None, transaction.map.find_weave("id"))
188
 
 
189
 
    def test_writable(self):
190
 
        transaction = transactions.PassThroughTransaction()
191
 
        self.assertTrue(transaction.writeable())
192
 
 
193
 
 
194
 
class TestWriteTransaction(TestCase):
195
 
 
196
 
    def setUp(self):
197
 
        self.transaction = transactions.WriteTransaction()
198
 
        super(TestWriteTransaction, self).setUp()
199
 
 
200
 
    def test_register_clean(self):
201
 
        self.transaction.register_clean("anobject")
202
 
 
203
 
    def test_register_dirty(self):
204
 
        self.transaction.register_dirty("anobject")
205
 
 
206
 
    def test_map(self):
207
 
        self.assertNotEqual(None, getattr(self.transaction, "map", None))
208
 
 
209
 
    def test_add_and_get(self):
210
 
        weave = "a weave"
211
 
        self.transaction.map.add_weave("id", weave)
212
 
        self.assertEqual(weave, self.transaction.map.find_weave("id"))
213
 
 
214
 
    def test_finish_returns(self):
215
 
        self.transaction.finish()
216
 
 
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)
224
 
 
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"))
231
 
        weave = None
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"))
239
 
        del weave
240
 
        # its not a weakref system
241
 
        self.assertEqual(DummyWeave("another weave"),
242
 
                         self.transaction.map.find_weave("id"))
243
 
 
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"))
250
 
        weave = None
251
 
        self.transaction.register_dirty(self.transaction.map.find_weave("id"))
252
 
        self.assertNotEqual(None, self.transaction.map.find_weave("id"))
253
 
 
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))
262
 
 
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"))
273
 
        # currently a fifo
274
 
        self.assertEqual(None, self.transaction.map.find_weave("id"))
275
 
        self.assertEqual(DummyWeave("a weave also"),
276
 
                         self.transaction.map.find_weave("id2"))
277
 
 
278
 
    def test_small_cache_with_references(self):
279
 
        # if we have a reference it should stick around
280
 
        weave = "a weave"
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"))
286
 
        weave = None
287
 
        # its not a weakref system
288
 
        self.assertEqual("a weave", self.transaction.map.find_weave("id"))
289
 
 
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"))
295
 
        weave = None
296
 
        # add an object, should not fall out even with no references.
297
 
        self.transaction.register_clean(self.transaction.map.find_weave("id"),
298
 
                                        precious=True)
299
 
        self.assertEqual(DummyWeave('a weave'),
300
 
                         self.transaction.map.find_weave("id"))
301
 
 
302
 
    def test_writable(self):
303
 
        self.assertTrue(self.transaction.writeable())
 
93
    
 
94
    def test_add_and_get(self):
 
95
        transaction = transactions.PassThroughTransaction()
 
96
        weave = "a weave"
 
97
        transaction.map.add_weave("id", weave)
 
98
        self.assertEqual(None, transaction.map.find_weave("id"))
 
99
        
 
100
    def test_rollback_asserts(self):
 
101
        transaction = transactions.PassThroughTransaction()
 
102
        self.assertRaises(errors.AlreadyCommitted, transaction.rollback)
 
103
 
 
104
    def test_finish_returns(self):
 
105
        transaction = transactions.PassThroughTransaction()
 
106
        transaction.finish()