[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')