[Zodb-checkins] CVS: ZODB4/ZEO/tests - testConnection.py:1.2

[Barry Warsaw]

Update of /cvs-repository/ZODB4/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv8419

Modified Files:
	testConnection.py 
Log Message:
Forward port from zodb3, use one true way to spawn the zeo server
subproc for both windows and unix.


=== ZODB4/ZEO/tests/testConnection.py 1.1 => 1.2 ===
--- ZODB4/ZEO/tests/testConnection.py:1.1	Fri Nov 22 16:24:53 2002
+++ ZODB4/ZEO/tests/testConnection.py	Mon Dec 16 16:16:10 2002
@@ -18,80 +18,74 @@
 """
 
 # System imports
-import os
-import time
-import socket
 import unittest
-
-# Zope/ZODB3 imports
-import zLOG
-
-# ZEO test support
-from ZEO.tests import forker
-
 # Import the actual test class
-from ZEO.tests.ConnectionTests import ConnectionTests
-
-class UnixConnectionTests(ConnectionTests):
+from ZEO.tests import ConnectionTests
 
-    """Add Unix-specific scaffolding to the generic test suite."""
 
-    def startServer(self, create=1, index=0, read_only=0, ro_svr=0):
-        zLOG.LOG("testZEO", zLOG.INFO,
-                 "startServer(create=%d, index=%d, read_only=%d)" %
-                 (create, index, read_only))
-        path = "%s.%d" % (self.file, index)
-        addr = self.addr[index]
-        pid, server = forker.start_zeo_server(
-            'FileStorage', (path, create, read_only), addr, ro_svr)
-        self._pids.append(pid)
-        self._servers.append(server)
-
-    def shutdownServer(self, index=0):
-        zLOG.LOG("testZEO", zLOG.INFO, "shutdownServer(index=%d)" % index)
-        self._servers[index].close()
-        if self._pids[index] is not None:
-            try:
-                os.waitpid(self._pids[index], 0)
-                self._pids[index] = None
-            except os.error, err:
-                print err
-
-class WindowsConnectionTests(ConnectionTests):
-
-    """Add Windows-specific scaffolding to the generic test suite."""
-
-    def startServer(self, create=1, index=0, read_only=0, ro_svr=0):
-        zLOG.LOG("testZEO", zLOG.INFO,
-                 "startServer(create=%d, index=%d, read_only=%d)" %
-                 (create, index, read_only))
-        path = "%s.%d" % (self.file, index)
-        addr = self.addr[index]
-        args = (path, '='+str(create), '='+str(read_only))
-        _addr, test_addr, test_pid = forker.start_zeo_server(
-            'FileStorage', args, addr, ro_svr)
-        self._pids.append(test_pid)
-        self._servers.append(test_addr)
-
-    def shutdownServer(self, index=0):
-        zLOG.LOG("testZEO", zLOG.INFO, "shutdownServer(index=%d)" % index)
-        if self._servers[index] is not None:
-            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            s.connect(self._servers[index])
-            s.close()
-            self._servers[index] = None
-            # XXX waitpid() isn't available until Python 2.3
-            time.sleep(0.5)
-
-if os.name == "posix":
-    test_classes = [UnixConnectionTests]
-elif os.name == "nt":
-    test_classes = [WindowsConnectionTests]
+class FileStorageConfig:
+    def getConfig(self, path, create, read_only):
+        s = """\
+        <Storage>
+            type FileStorage
+            file_name %s
+            create %s
+            read_only %s
+        </Storage>""" % (path,
+                         create and 'yes' or 'no',
+                         read_only and 'yes' or 'no')
+        return s
+
+class BerkeleyStorageConfig:
+    def getConfig(self, path, create, read_only):
+        # Full always creates and doesn't have a read_only flag
+        return """\
+        <Storage>
+            type BDBFullStorage
+            name %s
+            read_only %s
+        </Storage>""" % (path, read_only)
+
+
+class FileStorageConnectionTests(
+    FileStorageConfig,
+    ConnectionTests.ConnectionTests
+    ):
+    """FileStorage-specific connection tests."""
+
+
+class FileStorageReconnectionTests(
+    FileStorageConfig,
+    ConnectionTests.ReconnectionTests
+    ):
+    """FileStorage-specific re-connection tests."""
+
+
+class BDBConnectionTests(
+    BerkeleyStorageConfig,
+    ConnectionTests.ConnectionTests
+    ):
+    """Berkeley storage connection tests."""
+
+
+class BDBReconnectionTests(
+    BerkeleyStorageConfig,
+    ConnectionTests.ReconnectionTests
+    ):
+    """Berkeley storage re-connection tests."""
+
+
+test_classes = [FileStorageConnectionTests, FileStorageReconnectionTests]
+try:
+    from BDBStorage.BDBFullStorage import BDBFullStorage
+except ImportError:
+    pass
 else:
-    raise RuntimeError, "unsupported os: %s" % os.name
+    test_classes.append(BDBConnectionTests)
+    test_classes.append(BDBReconnectionTests)
 
-def test_suite():
 
+def test_suite():
     # shutup warnings about mktemp
     import warnings
     warnings.filterwarnings("ignore", "mktemp")
@@ -101,6 +95,7 @@
         sub = unittest.makeSuite(klass, 'check')
         suite.addTest(sub)
     return suite
+
 
 if __name__ == "__main__":
     unittest.main(defaultTest='test_suite')