[Zope-Checkins] CVS: Zope/lib/python/ZEO/tests - zeoserver.py:1.4.8.1 ConnectionTests.py:1.5.2.3 TestThread.py:1.3.8.2 forker.py:1.21.4.2 testConnection.py:1.2.2.2 testStart.py:1.12.4.3 testZEO.py:1.55.2.2
Chris McDonough
chrism@zope.com
Fri, 3 Jan 2003 01:37:07 -0500
Update of /cvs-repository/Zope/lib/python/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv28385/tests
Modified Files:
Tag: chrism-install-branch
ConnectionTests.py TestThread.py forker.py testConnection.py
testStart.py testZEO.py
Added Files:
Tag: chrism-install-branch
zeoserver.py
Log Message:
Merging chrism-install-branch with HEAD (hopefully for one of the last
times).
=== Added File Zope/lib/python/ZEO/tests/zeoserver.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Helper file used to launch a ZEO server cross platform"""
import os
import sys
import errno
import getopt
import random
import socket
import asyncore
import ThreadedAsync.LoopCallback
import ZConfig
import zLOG
from ZODB import StorageConfig
import ZEO.StorageServer
def load_storage(fp):
rootconf = ZConfig.loadfile(fp)
storageconf = rootconf.getSection('Storage')
return StorageConfig.createStorage(storageconf)
def cleanup(storage):
# FileStorage and the Berkeley storages have this method, which deletes
# all files and directories used by the storage. This prevents @-files
# from clogging up /tmp
try:
storage.cleanup()
except AttributeError:
pass
def log(label, msg, *args):
zLOG.LOG(label, zLOG.DEBUG, msg % args)
class ZEOTestServer(asyncore.dispatcher):
"""A server for killing the whole process at the end of a test.
The first time we connect to this server, we write an ack character down
the socket. The other end should block on a recv() of the socket so it
can guarantee the server has started up before continuing on.
The second connect to the port immediately exits the process, via
os._exit(), without writing data on the socket. It does close and clean
up the storage first. The other end will get the empty string from its
recv() which will be enough to tell it that the server has exited.
I think this should prevent us from ever getting a legitimate addr-in-use
error.
"""
__super_init = asyncore.dispatcher.__init__
def __init__(self, addr, storage, keep):
self.__super_init()
self._storage = storage
self._keep = keep
# Count down to zero, the number of connects
self._count = 1
# For zLOG
self._label ='zeoserver:%d @ %s' % (os.getpid(), addr)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
# Some ZEO tests attempt a quick start of the server using the same
# port so we have to set the reuse flag.
self.set_reuse_addr()
try:
self.bind(addr)
except:
# We really want to see these exceptions
import traceback
traceback.print_exc()
raise
self.listen(5)
self.log('bound and listening')
def log(self, msg, *args):
log(self._label, msg, *args)
def handle_accept(self):
sock, addr = self.accept()
self.log('in handle_accept()')
# When we're done with everything, close the storage. Do not write
# the ack character until the storage is finished closing.
if self._count <= 0:
self.log('closing the storage')
self._storage.close()
if not self._keep:
cleanup(self._storage)
self.log('exiting')
os._exit(0)
self.log('continuing')
sock.send('X')
self._count -= 1
def main():
label = 'zeoserver:%d' % os.getpid()
log(label, 'starting')
# We don't do much sanity checking of the arguments, since if we get it
# wrong, it's a bug in the test suite.
ro_svr = 0
keep = 0
configfile = None
# Parse the arguments and let getopt.error percolate
opts, args = getopt.getopt(sys.argv[1:], 'rkC:')
for opt, arg in opts:
if opt == '-r':
ro_svr = 1
elif opt == '-k':
keep = 1
elif opt == '-C':
configfile = arg
# Open the config file and let ZConfig parse the data there. Then remove
# the config file, otherwise we'll leave turds.
fp = open(configfile, 'r')
storage = load_storage(fp)
fp.close()
os.remove(configfile)
# The rest of the args are hostname, portnum
zeo_port = int(args[0])
test_port = zeo_port + 1
try:
log(label, 'creating the test server, ro: %s, keep: %s', ro_svr, keep)
t = ZEOTestServer(('', test_port), storage, keep)
except socket.error, e:
if e[0] <> errno.EADDRINUSE: raise
log(label, 'addr in use, closing and exiting')
storage.close()
cleanup(storage)
sys.exit(2)
addr = ('', zeo_port)
log(label, 'creating the storage server')
serv = ZEO.StorageServer.StorageServer(addr, {'1': storage}, ro_svr)
log(label, 'entering ThreadedAsync loop')
ThreadedAsync.LoopCallback.loop()
if __name__ == '__main__':
main()
=== Zope/lib/python/ZEO/tests/ConnectionTests.py 1.5.2.2 => 1.5.2.3 === (448/548 lines abridged)
--- Zope/lib/python/ZEO/tests/ConnectionTests.py:1.5.2.2 Sun Nov 24 18:55:26 2002
+++ Zope/lib/python/ZEO/tests/ConnectionTests.py Fri Jan 3 01:36:32 2003
@@ -11,21 +11,23 @@
# FOR A PARTICULAR PURPOSE
#
##############################################################################
-import asyncore
+
import os
+import sys
+import time
import random
import select
import socket
-import sys
+import asyncore
import tempfile
import threading
-import time
import zLOG
from ZEO.ClientStorage import ClientStorage
from ZEO.Exceptions import Disconnected
from ZEO.zrpc.marshal import Marshaller
+from ZEO.tests import forker
from ZODB.Transaction import get_transaction, Transaction
from ZODB.POSException import ReadOnlyError
@@ -34,22 +36,18 @@
from ZODB.tests.StorageTestBase import zodb_pickle, zodb_unpickle
from ZODB.tests.StorageTestBase import handle_all_serials, ZERO
+
class DummyDB:
- def invalidate(self, *args):
+ def invalidate(self, *args, **kws):
pass
-class ConnectionTests(StorageTestBase):
- """Tests that explicitly manage the server process.
- To test the cache or re-connection, these test cases explicit
- start and stop a ZEO storage server.
-
- This must be subclassed; the subclass must provide implementations
- of startServer() and shutdownServer().
- """
+class CommonSetupTearDown(StorageTestBase):
+ """Common boilerplate"""
[-=- -=- -=- 448 lines omitted -=- -=- -=-]
+ self.assertRaises(ReadOnlyError, self._dostore)
+
+ # Shut down the server
+ self.shutdownServer()
+ self._servers = []
+ # Poll until the client disconnects
+ self.pollDown()
+ # Stores should fail now
+ self.assertRaises(Disconnected, self._dostore)
+
+ # Restart the server, this time read-write
+ self.startServer(create=0)
+ # Poll until the client sconnects
+ self.pollUp()
+ # Stores should now succeed
+ self._dostore()
+
+ def checkReconnectSwitch(self):
+ # A fallback client initially connects to a read-only server,
+ # then discovers a read-write server and switches to that
+
+ # We don't want the read-write server created by setUp()
+ self.shutdownServer()
+ self._servers = []
+ # Allocate a second address (for the second server)
+ self._newAddr()
+
+ # Start a read-only server
+ self.startServer(create=0, index=0, read_only=1)
+ # Start a client in fallback mode
+ self._storage = self.openClientStorage(read_only_fallback=1)
+ # Stores should fail here
+ self.assertRaises(ReadOnlyError, self._dostore)
+
+ # Start a read-write server
+ self.startServer(index=1, read_only=0)
+ # After a while, stores should work
+ for i in range(300): # Try for 30 seconds
+ try:
+ self._dostore()
+ break
+ except (Disconnected, ReadOnlyError,
+ select.error, threading.ThreadError, socket.error):
+ time.sleep(0.1)
+ else:
+ self.fail("Couldn't store after starting a read-write server")
+
class MSTThread(threading.Thread):
=== Zope/lib/python/ZEO/tests/TestThread.py 1.3.8.1 => 1.3.8.2 ===
--- Zope/lib/python/ZEO/tests/TestThread.py:1.3.8.1 Tue Oct 8 20:41:42 2002
+++ Zope/lib/python/ZEO/tests/TestThread.py Fri Jan 3 01:36:32 2003
@@ -30,7 +30,7 @@
def run(self):
try:
self.testrun()
- except Exception, err:
+ except Exception:
s = StringIO()
traceback.print_exc(file=s)
self._testcase.fail("Exception in thread %s:\n%s\n" %
=== Zope/lib/python/ZEO/tests/forker.py 1.21.4.1 => 1.21.4.2 ===
--- Zope/lib/python/ZEO/tests/forker.py:1.21.4.1 Tue Oct 8 20:41:42 2002
+++ Zope/lib/python/ZEO/tests/forker.py Fri Jan 3 01:36:32 2003
@@ -13,20 +13,22 @@
##############################################################################
"""Library for forking storage server and connecting client storage"""
-import asyncore
import os
+import sys
+import time
+import errno
import random
import socket
-import sys
-import traceback
-import types
-import ZEO.ClientStorage
+import tempfile
+
+import zLOG
# Change value of PROFILE to enable server-side profiling
PROFILE = 0
if PROFILE:
import hotshot
+
def get_port():
"""Return a port that is not in use.
@@ -48,130 +50,71 @@
s.close()
raise RuntimeError, "Can't find port"
-if os.name == "nt":
-
- def start_zeo_server(storage_name, args, addr=None, ro_svr=0):
- """Start a ZEO server in a separate process.
- Returns the ZEO port, the test server port, and the pid.
- """
- import ZEO.tests.winserver
- if addr is None:
- port = get_port()
- else:
- port = addr[1]
- script = ZEO.tests.winserver.__file__
- if script.endswith('.pyc'):
- script = script[:-1]
- if ro_svr:
- prefix = (sys.executable, script, "-r")
- else:
- prefix = (sys.executable, script)
- args = prefix + (str(port), storage_name) + args
- d = os.environ.copy()
- d['PYTHONPATH'] = os.pathsep.join(sys.path)
- pid = os.spawnve(os.P_NOWAIT, sys.executable, args, d)
- return ('localhost', port), ('localhost', port + 1), pid
+def start_zeo_server(conf, addr=None, ro_svr=0, keep=0):
+ """Start a ZEO server in a separate process.
+ Returns the ZEO port, the test server port, and the pid.
+ """
+ # Store the config info in a temp file.
+ tmpfile = tempfile.mktemp()
+ fp = open(tmpfile, 'w')
+ fp.write(conf)
+ fp.close()
+ # Create the server
+ import ZEO.tests.zeoserver
+ if addr is None:
+ port = get_port()
+ else:
+ port = addr[1]
+ script = ZEO.tests.zeoserver.__file__
+ if script.endswith('.pyc'):
+ script = script[:-1]
+ # Create a list of arguments, which we'll tuplify below
+ qa = _quote_arg
+ args = [qa(sys.executable), qa(script), '-C', qa(tmpfile)]
+ if ro_svr:
+ args.append('-r')
+ if keep:
+ args.append('-k')
+ args.append(str(port))
+ d = os.environ.copy()
+ d['PYTHONPATH'] = os.pathsep.join(sys.path)
+ pid = os.spawnve(os.P_NOWAIT, sys.executable, tuple(args), d)
+ adminaddr = ('localhost', port+1)
+ # We need to wait until the server starts, but not forever
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ for i in range(5):
+ try:
+ zLOG.LOG('forker', zLOG.DEBUG, 'connect %s' % i)
+ s.connect(adminaddr)
+ ack = s.recv(1024)
+ zLOG.LOG('forker', zLOG.DEBUG, 'acked: %s' % ack)
+ break
+ except socket.error, e:
+ if e[0] <> errno.ECONNREFUSED: raise
+ time.sleep(1)
+ else:
+ zLOG.LOG('forker', zLOG.DEBUG, 'boo hoo')
+ raise
+ return ('localhost', port), adminaddr, pid
+
+
+if sys.platform[:3].lower() == "win":
+ def _quote_arg(s):
+ return '"%s"' % s
else:
+ def _quote_arg(s):
+ return s
- class ZEOServerExit(asyncore.file_dispatcher):
- """Used to exit ZEO.StorageServer when run is done"""
-
- def writable(self):
- return 0
-
- def readable(self):
- return 1
- def handle_read(self):
- buf = self.recv(4)
- if buf:
- assert buf == "done"
- server.close_server()
- asyncore.socket_map.clear()
-
- def handle_close(self):
- server.close_server()
- asyncore.socket_map.clear()
-
- class ZEOClientExit:
- """Used by client to cause server to exit"""
- def __init__(self, pipe):
- self.pipe = pipe
-
- def close(self):
- try:
- os.write(self.pipe, "done")
- os.close(self.pipe)
- except os.error:
- pass
-
- def start_zeo_server(storage_name, args, addr, ro_svr=0):
- assert isinstance(args, types.TupleType)
- rd, wr = os.pipe()
- pid = os.fork()
- if pid == 0:
- asyncore.socket_map.clear() # Don't service the parent's sockets
- import ZEO.zrpc.log
- reload(ZEO.zrpc.log) # Don't share the logging file object
- try:
- if PROFILE:
- p = hotshot.Profile("stats.s.%d" % os.getpid())
- p.runctx(
- "run_server(addr, rd, wr, storage_name, args, ro_svr)",
- globals(), locals())
- p.close()
- else:
- run_server(addr, rd, wr, storage_name, args, ro_svr)
- except:
- print "Exception in ZEO server process"
- traceback.print_exc()
- os._exit(0)
- else:
- os.close(rd)
- return pid, ZEOClientExit(wr)
-
- def load_storage(name, args):
- package = __import__("ZODB." + name)
- mod = getattr(package, name)
- klass = getattr(mod, name)
- return klass(*args)
-
- def run_server(addr, rd, wr, storage_name, args, ro_svr):
- # in the child, run the storage server
- global server
- os.close(wr)
- ZEOServerExit(rd)
- import ZEO.StorageServer, ZEO.zrpc.server
- storage = load_storage(storage_name, args)
- server = ZEO.StorageServer.StorageServer(addr, {'1':storage}, ro_svr)
- ZEO.zrpc.server.loop()
- storage.close()
- if isinstance(addr, types.StringType):
- os.unlink(addr)
-
- def start_zeo(storage_name, args, cache=None, cleanup=None,
- domain="AF_INET", storage_id="1", cache_size=20000000):
- """Setup ZEO client-server for storage.
-
- Returns a ClientStorage instance and a ZEOClientExit instance.
-
- XXX Don't know if os.pipe() will work on Windows.
- """
-
- if domain == "AF_INET":
- addr = '', get_port()
- elif domain == "AF_UNIX":
- import tempfile
- addr = tempfile.mktemp()
- else:
- raise ValueError, "bad domain: %s" % domain
-
- pid, exit = start_zeo_server(storage_name, args, addr)
- s = ZEO.ClientStorage.ClientStorage(addr, storage_id,
- client=cache,
- cache_size=cache_size,
- min_disconnect_poll=0.5,
- wait=1)
- return s, exit, pid
+def shutdown_zeo_server(adminaddr):
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ s.connect(adminaddr)
+ try:
+ ack = s.recv(1024)
+ except socket.error, e:
+ if e[0] <> errno.ECONNRESET: raise
+ ack = 'no ack received'
+ zLOG.LOG('shutdownServer', zLOG.DEBUG, 'acked: %s' % ack)
+ s.close()
=== Zope/lib/python/ZEO/tests/testConnection.py 1.2.2.1 => 1.2.2.2 ===
--- Zope/lib/python/ZEO/tests/testConnection.py:1.2.2.1 Tue Oct 8 20:41:42 2002
+++ Zope/lib/python/ZEO/tests/testConnection.py Fri Jan 3 01:36:33 2003
@@ -18,80 +18,73 @@
"""
# 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):
+ return """\
+ <Storage>
+ type FileStorage
+ file_name %s
+ create %s
+ read_only %s
+ </Storage>""" % (path,
+ create and 'yes' or 'no',
+ read_only and 'yes' or 'no')
+
+
+class BerkeleyStorageConfig:
+ def getConfig(self, path, create, read_only):
+ 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 +94,7 @@
sub = unittest.makeSuite(klass, 'check')
suite.addTest(sub)
return suite
+
if __name__ == "__main__":
unittest.main(defaultTest='test_suite')
=== Zope/lib/python/ZEO/tests/testStart.py 1.12.4.2 => 1.12.4.3 ===
--- Zope/lib/python/ZEO/tests/testStart.py:1.12.4.2 Sat Oct 26 15:51:49 2002
+++ Zope/lib/python/ZEO/tests/testStart.py Fri Jan 3 01:36:33 2003
@@ -48,7 +48,9 @@
if startfile[-1] == 'c':
startfile = startfile[:-1]
self.env = Environment(startfile)
- self.cmd = '%s %s' % (sys.executable, startfile)
+ # Put quotes around the arguments for (e.g.) Mac OS X,
+ # where the path names may include spaces.
+ self.cmd = '"%s" "%s"' % (sys.executable, startfile)
self.pids = {}
def tearDown(self):
@@ -185,6 +187,11 @@
pass
def test_suite():
+
+ # XXX These tests hang on the Zope trunk. Disable for now, as
+ # start.py won't be supported in ZODB 3.2 anyway.
+
+ return None
# shutup warnings about mktemp
import warnings
=== Zope/lib/python/ZEO/tests/testZEO.py 1.55.2.1 => 1.55.2.2 ===
--- Zope/lib/python/ZEO/tests/testZEO.py:1.55.2.1 Tue Oct 8 20:41:42 2002
+++ Zope/lib/python/ZEO/tests/testZEO.py Fri Jan 3 01:36:33 2003
@@ -30,22 +30,6 @@
from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_unpickle
-# Handle potential absence of removefs
-try:
- from ZODB.tests.StorageTestBase import removefs
-except ImportError:
- # for compatibility with Zope 2.5 &c.
- import errno
-
- def removefs(base):
- """Remove all files created by FileStorage with path base."""
- for ext in '', '.old', '.tmp', '.lock', '.index', '.pack':
- path = base + ext
- try:
- os.remove(path)
- except os.error, err:
- if err[0] != errno.ENOENT:
- raise
# ZODB test mixin classes
from ZODB.tests import StorageTestBase, BasicStorage, VersionStorage, \
@@ -67,8 +51,8 @@
def invalidate(self, *args):
pass
-class MiscZEOTests:
+class MiscZEOTests:
"""ZEO tests that don't fit in elsewhere."""
def checkLargeUpdate(self):
@@ -97,10 +81,10 @@
finally:
storage2.close()
+
class GenericTests(
# Base class for all ZODB tests
StorageTestBase.StorageTestBase,
-
# ZODB test mixin classes (in the same order as imported)
BasicStorage.BasicStorage,
VersionStorage.VersionStorage,
@@ -113,13 +97,13 @@
RevisionStorage.RevisionStorage,
MTStorage.MTStorage,
ReadOnlyStorage.ReadOnlyStorage,
-
# ZEO test mixin classes (in the same order as imported)
Cache.StorageWithCache,
Cache.TransUndoStorageWithCache,
CommitLockTests.CommitLockTests,
ThreadTests.ThreadTests,
- MiscZEOTests # Locally defined (see above)
+ # Locally defined (see above)
+ MiscZEOTests
):
"""Combine tests from various origins in one class."""
@@ -141,73 +125,65 @@
if hasattr(ZODB, "__version__"):
ReadOnlyStorage.ReadOnlyStorage.checkWriteMethods(self)
-class UnixTests(GenericTests):
- """Add Unix-specific scaffolding to the generic test suite."""
+class FileStorageTests(GenericTests):
+ """Test ZEO backed by a FileStorage."""
def setUp(self):
zLOG.LOG("testZEO", zLOG.INFO, "setUp() %s" % self.id())
- client, exit, pid = forker.start_zeo(*self.getStorage())
+ zeoport, adminaddr, pid = forker.start_zeo_server(self.getConfig())
self._pids = [pid]
- self._servers = [exit]
- self._storage = client
- client.registerDB(DummyDB(), None)
+ self._servers = [adminaddr]
+ self._storage = ClientStorage(zeoport, '1', cache_size=20000000,
+ min_disconnect_poll=0.5, wait=1)
+ self._storage.registerDB(DummyDB(), None)
def tearDown(self):
self._storage.close()
for server in self._servers:
- server.close()
- for pid in self._pids:
- os.waitpid(pid, 0)
- self.delStorage()
+ forker.shutdown_zeo_server(server)
+ if hasattr(os, 'waitpid'):
+ # Not in Windows Python until 2.3
+ for pid in self._pids:
+ os.waitpid(pid, 0)
+
+ def getConfig(self):
+ filename = self.__fs_base = tempfile.mktemp()
+ # Return a 1-tuple
+ return """\
+ <Storage>
+ type FileStorage
+ file_name %s
+ create yes
+ </Storage>
+ """ % filename
- def getStorage(self):
- self.__fs_base = tempfile.mktemp()
- return 'FileStorage', (self.__fs_base, '1')
- def delStorage(self):
- removefs(self.__fs_base)
+class BDBTests(FileStorageTests):
+ """ZEO backed by a Berkeley full storage."""
-class WindowsTests(GenericTests):
+ def getStorage(self):
+ self._envdir = tempfile.mktemp()
+ # Return a 1-tuple
+ return """\
+ <Storage>
+ type BDBFullStorage
+ name %s
+ </Storage>
+ """ % self._envdir
- """Add Windows-specific scaffolding to the generic test suite."""
- def setUp(self):
- zLOG.LOG("testZEO", zLOG.INFO, "setUp() %s" % self.id())
- args = self.getStorageInfo()
- name = args[0]
- args = args[1]
- zeo_addr, self.test_addr, self.test_pid = \
- forker.start_zeo_server(name, args)
- storage = ClientStorage(zeo_addr, wait=1, min_disconnect_poll=0.1)
- self._storage = storage
- storage.registerDB(DummyDB(), None)
- def tearDown(self):
- self._storage.close()
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.connect(self.test_addr)
- s.close()
- # the connection should cause the storage server to die
- time.sleep(0.5)
- self.delStorage()
-
- def getStorageInfo(self):
- self.__fs_base = tempfile.mktemp()
- return 'FileStorage', (self.__fs_base, '1') # create=1
-
- def delStorage(self):
- removefs(self.__fs_base)
-
-if os.name == "posix":
- test_classes = [UnixTests]
-elif os.name == "nt":
- test_classes = [WindowsTests]
+test_classes = [FileStorageTests]
+try:
+ from BDBStorage.BDBFullStorage import BDBFullStorage
+except ImportError:
+ pass
else:
- raise RuntimeError, "unsupported os: %s" % os.name
+ test_classes.append(BDBTests)
-def test_suite():
+def test_suite():
# shutup warnings about mktemp
import warnings
warnings.filterwarnings("ignore", "mktemp")
@@ -217,6 +193,7 @@
sub = unittest.makeSuite(klass, 'check')
suite.addTest(sub)
return suite
+
if __name__ == "__main__":
unittest.main(defaultTest='test_suite')