1
# Copyright (C) 2005 by Canonical Ltd
2
# Authors: Robert Collins <robert.collins@canonical.com>
4
# This program is free software; you can redistribute it and/or modify
5
# it under the terms of the GNU General Public License as published by
6
# the Free Software Foundation; either version 2 of the License, or
7
# (at your option) any later version.
9
# This program is distributed in the hope that it will be useful,
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12
# GNU General Public License for more details.
14
# You should have received a copy of the GNU General Public License
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
18
"""Tests for the behaviour of the Transaction concept in bzr."""
20
# import system imports here
24
#import bzrlib specific imports here
25
import bzrlib.errors as errors
26
from bzrlib.tests import TestCase, TestCaseInTempDir
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
36
def __eq__(self, other):
39
return self._message == other._message
42
class TestSymbols(TestCase):
44
def test_public_symbols(self):
45
from bzrlib.transactions import ReadOnlyTransaction
46
from bzrlib.transactions import PassThroughTransaction
49
class TestReadOnlyTransaction(TestCase):
52
self.transaction = transactions.ReadOnlyTransaction()
53
super(TestReadOnlyTransaction, self).setUp()
55
def test_register_clean(self):
56
self.transaction.register_clean("anobject")
58
def test_register_dirty_raises(self):
59
self.assertRaises(errors.ReadOnlyError,
60
self.transaction.register_dirty,"anobject")
62
def test_commit_raises(self):
63
self.assertRaises(errors.CommitNotPossible, self.transaction.commit)
66
self.assertNotEqual(None, getattr(self.transaction, "map", None))
68
def test_add_and_get(self):
70
self.transaction.map.add_weave("id", weave)
71
self.assertEqual(weave, self.transaction.map.find_weave("id"))
73
def test_rollback_returns(self):
74
self.transaction.rollback()
76
def test_finish_returns(self):
77
self.transaction.finish()
79
def test_zero_size_cache(self):
80
self.transaction.set_cache_size(0)
81
weave = DummyWeave('a weave')
82
self.transaction.map.add_weave("id", weave)
83
self.assertEqual(weave, self.transaction.map.find_weave("id"))
85
# add an object, should fall right out if there are no references
86
self.transaction.register_clean(self.transaction.map.find_weave("id"))
87
self.assertEqual(None, self.transaction.map.find_weave("id"))
88
# but if we have a reference it should stick around
89
weave = DummyWeave("another weave")
90
self.transaction.map.add_weave("id", weave)
91
self.transaction.register_clean(self.transaction.map.find_weave("id"))
92
self.assertEqual(weave, self.transaction.map.find_weave("id"))
94
# its not a weakref system
95
self.assertEqual(DummyWeave("another weave"),
96
self.transaction.map.find_weave("id"))
98
def test_small_cache(self):
99
self.transaction.set_cache_size(1)
100
# add an object, should not fall right out if there are no references
101
#sys.getrefcounts(foo)
102
self.transaction.map.add_weave("id", DummyWeave("a weave"))
103
self.transaction.register_clean(self.transaction.map.find_weave("id"))
104
self.assertEqual(DummyWeave("a weave"),
105
self.transaction.map.find_weave("id"))
106
self.transaction.map.add_weave("id2", DummyWeave("a weave also"))
107
self.transaction.register_clean(self.transaction.map.find_weave("id2"))
109
self.assertEqual(None, self.transaction.map.find_weave("id"))
110
self.assertEqual(DummyWeave("a weave also"),
111
self.transaction.map.find_weave("id2"))
113
def test_small_cache_with_references(self):
114
# if we have a reference it should stick around
116
weave2 = "another weave"
117
self.transaction.map.add_weave("id", weave)
118
self.transaction.map.add_weave("id2", weave2)
119
self.assertEqual(weave, self.transaction.map.find_weave("id"))
120
self.assertEqual(weave2, self.transaction.map.find_weave("id2"))
122
# its not a weakref system
123
self.assertEqual("a weave", self.transaction.map.find_weave("id"))
125
def test_precious_with_zero_size_cache(self):
126
self.transaction.set_cache_size(0)
127
weave = DummyWeave('a weave')
128
self.transaction.map.add_weave("id", weave)
129
self.assertEqual(weave, self.transaction.map.find_weave("id"))
131
# add an object, should not fall out even with no references.
132
self.transaction.register_clean(self.transaction.map.find_weave("id"),
134
self.assertEqual(DummyWeave('a weave'),
135
self.transaction.map.find_weave("id"))
137
def test_precious_revision_history(self):
138
"""Disabled test until revision-history is a real object."""
139
print "Disabled: test_precious_revision_history"
141
self.transaction.set_cache_size(0)
143
self.transaction.map.add_revision_history(history)
144
self.assertEqual(history, self.transaction.map.find_revision_history())
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())
152
class TestPassThroughTransaction(TestCase):
154
def test_construct(self):
155
transactions.PassThroughTransaction()
157
def test_register_clean(self):
158
transaction = transactions.PassThroughTransaction()
159
transaction.register_clean("anobject")
161
def test_register_dirty(self):
162
transaction = transactions.PassThroughTransaction()
163
transaction.register_dirty("anobject")
165
def test_commit_nothing_returns(self):
166
transaction = transactions.PassThroughTransaction()
170
transaction = transactions.PassThroughTransaction()
171
self.assertNotEqual(None, getattr(transaction, "map", None))
173
def test_add_and_get(self):
174
transaction = transactions.PassThroughTransaction()
176
transaction.map.add_weave("id", weave)
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)
183
def test_finish_returns(self):
184
transaction = transactions.PassThroughTransaction()
187
def test_cache_is_ignored(self):
188
transaction = transactions.PassThroughTransaction()
189
transaction.set_cache_size(100)
191
transaction.map.add_weave("id", weave)
192
self.assertEqual(None, transaction.map.find_weave("id"))
added
removed
bzrlib/tests/test_transactions.py
