[Zodb-checkins] CVS: StandaloneZODB/ZODB/tests - BasicStorage.py:1.15 Corruption.py:1.3 IteratorStorage.py:1.5 StorageTestBase.py:1.10 testDemoStorage.py:1.3 testFileStorage.py:1.14 testMappingStorage.py:1.2 testPersistentMapping.py:1.3 testTransaction.py:1.3
Jeremy Hylton
jeremy@zope.com
Mon, 21 Jan 2002 11:48:03 -0500
Update of /cvs-repository/StandaloneZODB/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv10073
Modified Files:
BasicStorage.py Corruption.py IteratorStorage.py
StorageTestBase.py testDemoStorage.py testFileStorage.py
testMappingStorage.py testPersistentMapping.py
testTransaction.py
Log Message:
Merge tests changes from the Standby-branch branch
=== StandaloneZODB/ZODB/tests/BasicStorage.py 1.14 => 1.15 ===
from ZODB.tests.MinPO import MinPO
-from ZODB.tests.StorageTestBase import zodb_unpickle, zodb_pickle
+from ZODB.tests.StorageTestBase \
+ import zodb_unpickle, zodb_pickle, handle_serials
ZERO = '\0'*8
@@ -77,7 +78,7 @@
'', txn)
r2 = self._storage.tpc_vote(txn)
self._storage.tpc_finish(txn)
- newrevid = self._handle_serials(oid, r1, r2)
+ newrevid = handle_serials(oid, r1, r2)
data, revid = self._storage.load(oid, '')
value = zodb_unpickle(data)
eq(value, MinPO(11))
@@ -173,3 +174,14 @@
# And another one
revid2 = self._dostore(oid, revid=revid1, data=p42)
eq(revid2, self._storage.getSerial(oid))
+
+ def checkTwoArgBegin(self):
+ # XXX how standard is three-argument tpc_begin()?
+ t = Transaction()
+ tid = chr(42) * 8
+ self._storage.tpc_begin(t, tid)
+ oid = self._storage.new_oid()
+ data = zodb_pickle(MinPO(8))
+ self._storage.store(oid, None, data, '', t)
+ self._storage.tpc_vote(t)
+ self._storage.tpc_finish(t)
=== StandaloneZODB/ZODB/tests/Corruption.py 1.2 => 1.3 ===
def setUp(self):
+ self.__super_setUp()
self.path = tempfile.mktemp()
self._storage = ZODB.FileStorage.FileStorage(self.path, create=1)
- self.__super_setUp()
def tearDown(self):
self.__super_tearDown()
=== StandaloneZODB/ZODB/tests/IteratorStorage.py 1.4 => 1.5 ===
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
+from ZODB.utils import U64, p64
-class IteratorStorage:
- def checkSimpleIteration(self):
+class IteratorCompare:
+
+ def iter_verify(self, txniter, revids, val0):
eq = self.assertEqual
- # Store a bunch of revisions of a single object
- oid = self._storage.new_oid()
- revid1 = self._dostore(oid, data=MinPO(11))
- revid2 = self._dostore(oid, revid=revid1, data=MinPO(12))
- revid3 = self._dostore(oid, revid=revid2, data=MinPO(13))
- # Now iterate over all the transactions
- val = 11
- txniter = self._storage.iterator()
- for reciter, revid in zip(txniter, (revid1, revid2, revid3)):
+ oid = self._oid
+ val = val0
+ for reciter, revid in zip(txniter, revids + [None]):
eq(reciter.tid, revid)
for rec in reciter:
eq(rec.oid, oid)
@@ -28,3 +24,54 @@
eq(rec.version, '')
eq(zodb_unpickle(rec.data), MinPO(val))
val = val + 1
+ eq(val, val0 + len(revids))
+
+class IteratorStorage(IteratorCompare):
+
+ def checkSimpleIteration(self):
+ # Store a bunch of revisions of a single object
+ self._oid = oid = self._storage.new_oid()
+ revid1 = self._dostore(oid, data=MinPO(11))
+ revid2 = self._dostore(oid, revid=revid1, data=MinPO(12))
+ revid3 = self._dostore(oid, revid=revid2, data=MinPO(13))
+ # Now iterate over all the transactions and compare carefully
+ txniter = self._storage.iterator()
+ self.iter_verify(txniter, [revid1, revid2, revid3], 11)
+
+class ExtendedIteratorStorage(IteratorCompare):
+
+ def checkExtendedIteration(self):
+ # Store a bunch of revisions of a single object
+ self._oid = oid = self._storage.new_oid()
+ revid1 = self._dostore(oid, data=MinPO(11))
+ revid2 = self._dostore(oid, revid=revid1, data=MinPO(12))
+ revid3 = self._dostore(oid, revid=revid2, data=MinPO(13))
+ revid4 = self._dostore(oid, revid=revid3, data=MinPO(14))
+ # Note that the end points are included
+ # Iterate over all of the transactions with explicit start/stop
+ txniter = self._storage.iterator(revid1, revid4)
+ self.iter_verify(txniter, [revid1, revid2, revid3, revid4], 11)
+ # Iterate over some of the transactions with explicit start
+ txniter = self._storage.iterator(revid3)
+ self.iter_verify(txniter, [revid3, revid4], 13)
+ # Iterate over some of the transactions with explicit stop
+ txniter = self._storage.iterator(None, revid2)
+ self.iter_verify(txniter, [revid1, revid2], 11)
+ # Iterate over some of the transactions with explicit start+stop
+ txniter = self._storage.iterator(revid2, revid3)
+ self.iter_verify(txniter, [revid2, revid3], 12)
+ # Specify an upper bound somewhere in between values
+ revid3a = p64((U64(revid3) + U64(revid4)) / 2)
+ txniter = self._storage.iterator(revid2, revid3a)
+ self.iter_verify(txniter, [revid2, revid3], 12)
+ # Specify a lower bound somewhere in between values
+ revid1a = p64((U64(revid1) + U64(revid2)) / 2)
+ txniter = self._storage.iterator(revid1a, revid3a)
+ self.iter_verify(txniter, [revid2, revid3], 12)
+ # Specify an empty range
+ txniter = self._storage.iterator(revid3, revid2)
+ self.iter_verify(txniter, [], 13)
+ # Specify a singleton range
+ txniter = self._storage.iterator(revid3, revid3)
+ self.iter_verify(txniter, [revid3], 13)
+
=== StandaloneZODB/ZODB/tests/StorageTestBase.py 1.9 => 1.10 ===
return inst
+def handle_all_serials(oid, *args):
+ """Return dict of oid to serialno from store() and tpc_vote().
+
+ Raises an exception if one of the calls raised an exception.
+
+ The storage interface got complicated when ZEO was introduced.
+ Any individual store() call can return None or a sequence of
+ 2-tuples where the 2-tuple is either oid, serialno or an
+ exception to be raised by the client.
+
+ The original interface just returned the serialno for the
+ object.
+ """
+ d = {}
+ for arg in args:
+ if isinstance(arg, types.StringType):
+ d[oid] = arg
+ elif arg is None:
+ pass
+ else:
+ for oid, serial in arg:
+ if not isinstance(serial, types.StringType):
+ raise arg
+ d[oid] = serial
+ return d
+
+def handle_serials(oid, *args):
+ """Return the serialno for oid based on multiple return values.
+
+ A helper for function _handle_all_serials().
+ """
+ args = (oid,) + args
+ return apply(handle_all_serials, args)[oid]
+
def import_helper(name):
mod = __import__(name)
- for part in string.split(name, ".")[1:]:
- mod = getattr(mod, part)
- return mod
+ return sys.modules[name]
class StorageTestBase(unittest.TestCase):
def setUp(self):
# You need to override this with a setUp that creates self._storage
self._transaction = Transaction()
+ self._storage = None
def _close(self):
# You should override this if closing your storage requires additional
# shutdown operations.
- self._transaction.abort()
- self._storage.close()
+ if self._transaction:
+ self._transaction.abort()
+ if self._storage is not None:
+ self._storage.close()
def tearDown(self):
self._close()
- def _handle_all_serials(self, oid, *args):
- """Return dict of oid to serialno from store() and tpc_vote().
-
- Raises an exception if one of the calls raised an exception.
-
- The storage interface got complicated when ZEO was introduced.
- Any individual store() call can return None or a sequence of
- 2-tuples where the 2-tuple is either oid, serialno or an
- exception to be raised by the client.
-
- The original interface just returned the serialno for the
- object.
- """
- d = {}
- for arg in args:
- if isinstance(arg, types.StringType):
- d[oid] = arg
- elif arg is None:
- pass
- else:
- for oid, serial in arg:
- if not isinstance(serial, types.StringType):
- raise arg
- d[oid] = serial
- return d
-
- def _handle_serials(self, oid, *args):
- """Return the serialno for oid based on multiple return values.
-
- A helper for function _handle_all_serials().
- """
- args = (oid,) + args
- return apply(self._handle_all_serials, args)[oid]
-
def _dostore(self, oid=None, revid=None, data=None, version=None,
already_pickled=0):
"""Do a complete storage transaction. The defaults are:
@@ -146,6 +147,7 @@
if version is None:
version = ''
# Begin the transaction
+ self._transaction = Transaction()
self._storage.tpc_begin(self._transaction)
# Store an object
r1 = self._storage.store(oid, revid, data, version,
@@ -153,7 +155,7 @@
# Finish the transaction
r2 = self._storage.tpc_vote(self._transaction)
self._storage.tpc_finish(self._transaction)
- return self._handle_serials(oid, r1, r2)
+ return handle_serials(oid, r1, r2)
def _dostoreNP(self, oid=None, revid=None, data=None, version=None):
return self._dostore(oid, revid, data, version, already_pickled=1)
=== StandaloneZODB/ZODB/tests/testDemoStorage.py 1.2 => 1.3 ===
def setUp(self):
- self._storage = ZODB.DemoStorage.DemoStorage()
StorageTestBase.StorageTestBase.setUp(self)
+ self._storage = ZODB.DemoStorage.DemoStorage()
def test_suite():
suite = unittest.makeSuite(DemoStorageTests, 'check')
=== StandaloneZODB/ZODB/tests/testFileStorage.py 1.13 => 1.14 ===
TransactionalUndoVersionStorage, PackableStorage, \
Synchronization, ConflictResolution, HistoryStorage, \
- IteratorStorage, Corruption, RevisionStorage, PersistentStorage
+ IteratorStorage, Corruption, RevisionStorage, PersistentStorage, \
+ MTStorage, ReadOnlyStorage
class FileStorageTests(
StorageTestBase.StorageTestBase,
@@ -19,7 +20,10 @@
ConflictResolution.ConflictResolvingStorage,
HistoryStorage.HistoryStorage,
IteratorStorage.IteratorStorage,
+ IteratorStorage.ExtendedIteratorStorage,
PersistentStorage.PersistentStorage,
+ MTStorage.MTStorage,
+ ReadOnlyStorage.ReadOnlyStorage
):
def open(self, **kwargs):
@@ -31,8 +35,8 @@
'FileStorageTests.fs', **kwargs)
def setUp(self):
- self.open(create=1)
StorageTestBase.StorageTestBase.setUp(self)
+ self.open(create=1)
def tearDown(self):
StorageTestBase.StorageTestBase.tearDown(self)
=== StandaloneZODB/ZODB/tests/testMappingStorage.py 1.1 => 1.2 ===
def setUp(self):
- self._storage = ZODB.MappingStorage.MappingStorage()
StorageTestBase.StorageTestBase.setUp(self)
+ self._storage = ZODB.MappingStorage.MappingStorage()
def test_suite():
suite = unittest.makeSuite(MappingStorageTests, 'check')
=== StandaloneZODB/ZODB/tests/testPersistentMapping.py 1.2 => 1.3 ===
"""Helper for this test suite to get special PersistentMapping"""
- #print modulename, classname
if classname == "PersistentMapping":
class PersistentMapping:
def __setstate__(self, state):
=== StandaloneZODB/ZODB/tests/testTransaction.py 1.2 => 1.3 ===
meth = getattr(obj, meth_name)
meth(1)
- get_transaction().commit()
-
+ get_transaction().commit()
+
checkSubSingleCommit = lambda self:\
self.wrap_test(BasicTests, "checkSingleCommit")