[Zodb-checkins] CVS: Zope3/src/zodb/zeo/tests - test_zeo.py:1.7.2.2
Jeremy Hylton
jeremy@zope.com
Fri, 14 Feb 2003 15:51:58 -0500
Update of /cvs-repository/Zope3/src/zodb/zeo/tests
In directory cvs.zope.org:/tmp/cvs-serv1900/zeo/tests
Modified Files:
Tag: ZODB3-2-integration-branch
test_zeo.py
Log Message:
Merge in variant of changes from ZODB 3.2.
Split the tests into those that all the storages can support and those
that mapping storage can't support.
=== Zope3/src/zodb/zeo/tests/test_zeo.py 1.7.2.1 => 1.7.2.2 ===
--- Zope3/src/zodb/zeo/tests/test_zeo.py:1.7.2.1 Mon Feb 10 17:08:29 2003
+++ Zope3/src/zodb/zeo/tests/test_zeo.py Fri Feb 14 15:51:57 2003
@@ -30,25 +30,18 @@
# ZODB test mixin classes
-from zodb.storage.tests import base, basic, version, \
- undo, undoversion, \
- packable, synchronization, conflict, revision, \
- mt, readonly
+from zodb.storage.tests import base, basic, version, undo, undoversion, \
+ packable, synchronization, conflict, revision, mt, readonly
-# ZEO imports
from zodb.zeo.client import ClientStorage
-
-# ZEO test support
from zodb.zeo.tests import forker, cache
-
-# ZEO test mixin classes
from zodb.zeo.tests import commitlock, threadtests
+from zodb.zeo.tests.common import TestClientStorage, DummyDB
class DummyDB:
def invalidate(self, *args):
pass
-
class MiscZEOTests:
"""ZEO tests that don't fit in elsewhere."""
@@ -58,7 +51,7 @@
def checkZEOInvalidation(self):
addr = self._storage._addr
- storage2 = ClientStorage(addr, wait=True, min_disconnect_poll=0.1)
+ storage2 = TestClientStorage(addr, wait=True, min_disconnect_poll=0.1)
try:
oid = self._storage.newObjectId()
ob = MinPO('first')
@@ -78,59 +71,40 @@
finally:
storage2.close()
+class ZEOConflictTests(
+ conflict.ConflictResolvingStorage,
+ conflict.ConflictResolvingTransUndoStorage):
+
+ def unresolvable(self, klass):
+ # This helper method is used to test the implementation of
+ # conflict resolution. That code runs in the server, and there
+ # is no way for the test suite (a client) to inquire about it.
+ return False
-class GenericTests(
+class StorageTests(
# Base class for all ZODB tests
base.StorageTestBase,
# ZODB test mixin classes
basic.BasicStorage,
- version.VersionStorage,
- undo.TransactionalUndoStorage,
- undoversion.TransactionalUndoVersionStorage,
- packable.PackableStorage,
- synchronization.SynchronizedStorage,
- conflict.ConflictResolvingStorage,
- conflict.ConflictResolvingTransUndoStorage,
- revision.RevisionStorage,
- mt.MTStorage,
readonly.ReadOnlyStorage,
+ revision.RevisionStorage,
+ synchronization.SynchronizedStorage,
# ZEO test mixin classes
cache.StorageWithCache,
- cache.TransUndoStorageWithCache,
commitlock.CommitLockTests,
threadtests.ThreadTests,
# Locally defined (see above)
MiscZEOTests
):
-
- """Combine tests from various origins in one class."""
-
- def open(self, read_only=0):
- # XXX Needed to support ReadOnlyStorage tests. Ought to be a
- # cleaner way.
- addr = self._storage._addr
- self._storage.close()
- self._storage = ClientStorage(addr, read_only=read_only, wait=1)
-
- _open = open
-
- def unresolvable(self, klass):
- # This helper method is used to test the implementation of
- # conflict resolution. That code runs in the server, and there
- # is no way for the test suite (a client) to inquire about it.
- pass
-
-
-class FileStorageTests(GenericTests):
- """Test ZEO backed by a FileStorage."""
+ """Tests for storage that supports IStorage."""
def setUp(self):
logging.info("testZEO: setUp() %s", self.id())
zeoport, adminaddr, pid = forker.start_zeo_server(self.getConfig())
self._pids = [pid]
self._servers = [adminaddr]
- self._storage = ClientStorage(zeoport, '1', cache_size=20000000,
- min_disconnect_poll=0.5, wait=1)
+ self._storage = TestClientStorage(zeoport, '1', cache_size=20000000,
+ min_disconnect_poll=0.5, wait=1)
self._storage.registerDB(DummyDB())
def tearDown(self):
@@ -142,9 +116,36 @@
for pid in self._pids:
os.waitpid(pid, 0)
+ def open(self, read_only=False):
+ # XXX Needed to support ReadOnlyStorage tests. Ought to be a
+ # cleaner way.
+ addr = self._storage._addr
+ self._storage.close()
+ self._storage = TestClientStorage(addr, read_only=read_only, wait=True)
+
+class UndoVersionStorageTests(
+ StorageTests,
+ ZEOConflictTests,
+ cache.TransUndoStorageWithCache,
+ mt.MTStorage,
+ packable.PackableStorage,
+ undo.TransactionalUndoStorage,
+ undoversion.TransactionalUndoVersionStorage,
+ version.VersionStorage,
+ ):
+ """Tests for storage that supports IUndoStorage and IVersionStorage."""
+
+ # XXX Some of the pack tests should really be run for the mapping
+ # storage, but the pack tests assume that the storage also supports
+ # multiple revisions.
+
+class FileStorageTests(UndoVersionStorageTests):
+ """Test ZEO backed by a FileStorage."""
+
+ level = 2
+
def getConfig(self):
filename = self.__fs_base = tempfile.mktemp()
- # Return a 1-tuple
return """\
<Storage>
type FileStorage
@@ -153,13 +154,13 @@
</Storage>
""" % filename
-
-class BDBTests(FileStorageTests):
+class BDBTests(UndoVersionStorageTests):
"""ZEO backed by a Berkeley Full storage."""
- def getStorage(self):
+ level = 2
+
+ def getConfig(self):
self._envdir = tempfile.mktemp()
- # Return a 1-tuple
return """\
<Storage>
type BDBFullStorage
@@ -167,25 +168,26 @@
</Storage>
""" % self._envdir
+class MappingStorageTests(StorageTests):
+
+ def getConfig(self):
+ self._envdir = tempfile.mktemp()
+ return """\
+ <Storage>
+ type MappingStorage
+ name %s
+ </Storage>
+ """ % self._envdir
-test_classes = [FileStorageTests]
+test_classes = [FileStorageTests, MappingStorageTests]
from zodb.storage.base import berkeley_is_available
if berkeley_is_available:
test_classes.append(BDBTests)
-
def test_suite():
- # shutup warnings about mktemp
- import warnings
- warnings.filterwarnings("ignore", "mktemp")
-
suite = unittest.TestSuite()
for klass in test_classes:
sub = unittest.makeSuite(klass, 'check')
suite.addTest(sub)
return suite
-
-
-if __name__ == "__main__":
- unittest.main(defaultTest='test_suite')