« back to all changes in this revision
Viewing changes to tests/zope/zstorm.py
- browse files at revision 276.2.2
- compare with another revision
- download diff
- download tarball
- view history from revision 276.2.2
- Committer: Gustavo Niemeyer
- Date: 2008-11-04 16:52:14 UTC
- mfrom: (280 trunk)
- mto: This revision was merged to the branch mainline in revision 282.
- Revision ID: gustavo@niemeyer.net-20081104165214-p2pn9lraefjiu91b
Merged from trunk.
added
removed
tests/zope/zstorm.py
29
29
else:
30
30
has_transaction = True
31
31
from storm.zope.interfaces import IZStorm, ZStormError
32
from storm.zope.zstorm import ZStorm
32
from storm.zope.zstorm import ZStorm, StoreDataManager
33
33
34
34
try:
35
35
from zope.component import provideUtility, getUtility
42
42
43
43
44
44
from storm.exceptions import OperationalError
45
from storm.locals import Store, Int
45
from storm.locals import Store
46
46
47
47
48
48
class ZStormTest(TestHelper):
157
157
"name2": "sqlite:2",
158
158
"name3": "sqlite:3"})
159
159
160
def _isInTransaction(self, store):
161
"""Check if a Store is part of the current transaction."""
162
for dm in transaction.get()._resources:
163
if isinstance(dm, StoreDataManager) and dm._store is store:
164
return True
165
return False
166
167
def assertInTransaction(self, store):
168
"""Check that the given store is joined to the transaction."""
169
self.assertTrue(self._isInTransaction(store),
170
"%r should be joined to the transaction" % store)
171
172
def assertNotInTransaction(self, store):
173
"""Check that the given store is not joined to the transaction."""
174
self.assertTrue(not self._isInTransaction(store),
175
"%r should not be joined to the transaction" % store)
176
177
def test_wb_store_joins_transaction_on_register_event(self):
178
"""The Store joins the transaction when register-transaction
179
is emitted.
180
181
The Store tests check the various operations that trigger this
182
event.
183
"""
184
store = self.zstorm.get("name", "sqlite:")
185
self.assertNotInTransaction(store)
186
store._event.emit("register-transaction")
187
self.assertInTransaction(store)
188
189
def test_wb_store_joins_transaction_on_use_after_commit(self):
190
store = self.zstorm.get("name", "sqlite:")
191
store.execute("SELECT 1")
192
transaction.commit()
193
self.assertNotInTransaction(store)
194
store.execute("SELECT 1")
195
self.assertInTransaction(store)
196
197
def test_wb_store_joins_transaction_on_use_after_abort(self):
198
store = self.zstorm.get("name", "sqlite:")
199
store.execute("SELECT 1")
200
transaction.abort()
201
self.assertNotInTransaction(store)
202
store.execute("SELECT 1")
203
self.assertInTransaction(store)
204
160
205
def test_remove(self):
161
206
removed_store = self.zstorm.get("name", "sqlite:")
162
207
self.zstorm.remove(removed_store)
164
209
self.assertNotEquals(store, removed_store)
165
210
self.assertRaises(ZStormError, self.zstorm.get, "name")
166
211
167
# Abort the transaction so that the currently registered
168
# resource detaches.
169
transaction.abort()
170
171
# Let's try to make storm blow up when committing, so that
172
# we can be sure that the removed store isn't linked to the
173
# transaction system by a synchronizer anymore.
174
class BadTable(object):
175
__storm_table__ = "bad_table"
176
id = Int(primary=True)
177
removed_store.add(BadTable())
178
179
# If this fails, the store is still linked to the transaction
180
# system.
181
transaction.commit()
182
183
def test_double_abort(self):
184
"""
185
Surprisingly, transaction.abort() won't detach the internal
186
transaction from the manager. This means that when
187
transaction.begin() runs, it will detect that there's still
188
a Transaction instance around, and will call abort on it once
189
more before instantiating a new Transaction. The Transaction's
190
abort() method will then call beforeCompletion() on our
191
synchronizer, which then tries to join() on the current
192
transaction, which is still the old one and blows up saying
193
that a previous transaction has failed (we know it, since we
194
*aborted* it).
195
"""
196
class BadTable(object):
197
__storm_table__ = "bad_table"
198
id = Int(primary=True, default=1)
199
store = self.zstorm.get("name", "sqlite:")
200
store.add(BadTable())
201
self.assertRaises(OperationalError, transaction.commit)
202
transaction.abort()
203
transaction.abort()
212
def test_wb_removed_store_does_not_join_transaction(self):
213
"""If a store has been removed, it will not join the transaction."""
214
store = self.zstorm.get("name", "sqlite:")
215
self.zstorm.remove(store)
216
store.execute("SELECT 1")
217
self.assertNotInTransaction(store)
218
219
def test_wb_removed_store_does_not_join_future_transactions(self):
220
"""If a store has been removed after joining a transaction, it
221
will not join new transactions."""
222
store = self.zstorm.get("name", "sqlite:")
223
store.execute("SELECT 1")
224
self.zstorm.remove(store)
225
self.assertInTransaction(store)
226
227
transaction.abort()
228
store.execute("SELECT 1")
229
self.assertNotInTransaction(store)
204
230
205
231
def test_wb_reset(self):
206
232
"""_reset is used to reset the zstorm utility between zope test runs.
207
208
We must make sure the L{StoreSynchronizer} is removed from the
209
transaction manager.
210
233
"""
211
234
store = self.zstorm.get("name", "sqlite:")
212
self.assertEquals(
213
len(transaction.manager._synchs.values()[0].data.values()), 1)
214
transaction.abort()
215
self.assertEquals(
216
len(transaction.manager._synchs.values()[0].data.values()), 1)
217
235
self.zstorm._reset()
218
self.assertEquals(
219
len(transaction.manager._synchs.values()[0].data.values()), 0)
236
self.assertEqual(list(self.zstorm.iterstores()), [])
220
237
221
238
def test_store_strong_reference(self):
222
239
"""
Loggerhead is a web-based interface for Breezy
Version: 2.0.1