[Zope-Checkins] CVS: StandaloneZODB/ZODB/tests - IteratorStorage.py:1.5.4.2
Jeremy Hylton
jeremy@zope.com
Thu, 24 Jan 2002 13:03:22 -0500
Update of /cvs-repository/StandaloneZODB/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv19410
Modified Files:
Tag: Recovery
IteratorStorage.py
Log Message:
Add new tests from trunk.
=== StandaloneZODB/ZODB/tests/IteratorStorage.py 1.5.4.1 => 1.5.4.2 ===
from ZODB.tests.StorageTestBase import zodb_unpickle
from ZODB.utils import U64, p64
-
+from ZODB.Transaction import Transaction
class IteratorCompare:
@@ -37,6 +37,66 @@
# Now iterate over all the transactions and compare carefully
txniter = self._storage.iterator()
self.iter_verify(txniter, [revid1, revid2, revid3], 11)
+
+ def checkVersionIterator(self):
+ if not self._storage.supportsVersions():
+ return
+ self._dostore()
+ self._dostore(version='abort')
+ self._dostore()
+ self._dostore(version='abort')
+ t = Transaction()
+ self._storage.tpc_begin(t)
+ self._storage.abortVersion('abort', t)
+ self._storage.tpc_vote(t)
+ self._storage.tpc_finish(t)
+
+ self._dostore(version='commit')
+ self._dostore()
+ self._dostore(version='commit')
+ t = Transaction()
+ self._storage.tpc_begin(t)
+ self._storage.commitVersion('commit', '', t)
+ self._storage.tpc_vote(t)
+ self._storage.tpc_finish(t)
+
+ # XXX extend these checks. right now, just iterating with CVS
+ # FS or Berkeley will fail here, but once fixed we should
+ # check that the right data is returned.
+ txniter = self._storage.iterator()
+ for trans in txniter:
+ for data in trans:
+ pass
+
+ def checkTransactionalUndoIterator(self):
+ if not hasattr(self._storage, 'supportsTransactionalUndo'):
+ return
+ if not self._storage.supportsTransactionalUndo():
+ return
+
+ oid = self._storage.new_oid()
+ revid = self._dostore(oid)
+ self._dostore(oid, revid)
+
+ self.undoLastTrans()
+ self.undoLastTrans()
+
+ # XXX extend these checks. right now, just iterating with CVS
+ # FS or Berkeley will fail here, but once fixed we should
+ # check that the right data is returned.
+ txniter = self._storage.iterator()
+ for trans in txniter:
+ for data in trans:
+ print repr(data.oid), repr(data.data)
+
+ def undoLastTrans(self):
+ info = self._storage.undoInfo()
+ tid = info[0]['id']
+ t = Transaction()
+ self._storage.tpc_begin(t)
+ oids = self._storage.transactionalUndo(tid, t)
+ self._storage.tpc_vote(t)
+ self._storage.tpc_finish(t)
class ExtendedIteratorStorage(IteratorCompare):