[Zope3-checkins] CVS: Zope3/src/zodb/zeo/tests - thread.py:1.2.6.1 common.py:1.2.6.1 zeoserver.py:1.7.8.1 threadtests.py:1.4.4.1 test_zeo.py:1.7.4.2 test_conn.py:1.3.8.1 test_cache.py:1.2.12.1 forker.py:1.4.8.1 connection.py:1.5.4.1 commitlock.py:1.3.4.1
Jeremy Hylton
jeremy@zope.com
Wed, 12 Mar 2003 16:41:57 -0500
Update of /cvs-repository/Zope3/src/zodb/zeo/tests
In directory cvs.zope.org:/tmp/cvs-serv29357/zeo/tests
Modified Files:
Tag: opaque-pickles-branch
zeoserver.py threadtests.py test_zeo.py test_conn.py
test_cache.py forker.py connection.py commitlock.py
Added Files:
Tag: opaque-pickles-branch
thread.py common.py
Log Message:
Update from trunk.
Resolve a few import conflicts.
=== Added File Zope3/src/zodb/zeo/tests/thread.py ===
##############################################################################
#
# Copyright (c) 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
#
##############################################################################
"""A Thread base class for use with unittest."""
from cStringIO import StringIO
import threading
import traceback
class TestThread(threading.Thread):
__super_init = threading.Thread.__init__
__super_run = threading.Thread.run
def __init__(self, testcase, group=None, target=None, name=None,
args=(), kwargs={}, verbose=None):
self.__super_init(group, target, name, args, kwargs, verbose)
self.setDaemon(1)
self._testcase = testcase
def run(self):
try:
self.testrun()
except Exception:
s = StringIO()
traceback.print_exc(file=s)
self._testcase.fail("Exception in thread %s:\n%s\n" %
(self, s.getvalue()))
def cleanup(self, timeout=15):
self.join(timeout)
if self.isAlive():
self._testcase.fail("Thread did not finish: %s" % self)
=== Added File Zope3/src/zodb/zeo/tests/common.py ===
##############################################################################
#
# Copyright (c) 2003 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
#
##############################################################################
"""Common helper classes for testing."""
import threading
from zodb.zeo.client import ClientStorage
class TestClientStorage(ClientStorage):
def verify_cache(self, stub):
self.end_verify = threading.Event()
self.verify_result = super(TestClientStorage, self).verify_cache(stub)
def endVerify(self):
super(TestClientStorage, self).endVerify()
self.end_verify.set()
class DummyDB:
def invalidate(self, *args, **kws):
pass
=== Zope3/src/zodb/zeo/tests/zeoserver.py 1.7 => 1.7.8.1 ===
--- Zope3/src/zodb/zeo/tests/zeoserver.py:1.7 Mon Jan 27 14:44:14 2003
+++ Zope3/src/zodb/zeo/tests/zeoserver.py Wed Mar 12 16:41:21 2003
@@ -29,24 +29,12 @@
import zodb.zeo.server
from zodb.zeo import threadedasync
-
def load_storage(fp):
context = ZConfig.Context.Context()
rootconf = context.loadFile(fp)
storageconf = rootconf.getSection('Storage')
return config.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
-
-
class ZEOTestServer(asyncore.dispatcher):
"""A server for killing the whole process at the end of a test.
@@ -64,10 +52,10 @@
"""
__super_init = asyncore.dispatcher.__init__
- def __init__(self, addr, storage, keep):
+ def __init__(self, addr, server, keep):
self.__super_init()
self._sockets = [self]
- self._storage = storage
+ self._server = server
self._keep = keep
# Count down to zero, the number of connects
self._count = 1
@@ -95,9 +83,10 @@
# the ack character until the storage is finished closing.
if self._count <= 0:
self.logger.info('closing the storage')
- self._storage.close()
+ self._server.close_server()
if not self._keep:
- cleanup(self._storage)
+ for storage in self._server.storages.values():
+ storage.cleanup()
self.logger.info('exiting')
# Close all the other sockets so that we don't have to wait
# for os._exit() to get to it before starting the next
@@ -137,8 +126,11 @@
ro_svr = False
keep = False
configfile = None
+ invalidation_queue_size = 100
+ transaction_timeout = None
+ monitor_address = None
# Parse the arguments and let getopt.error percolate
- opts, args = getopt.getopt(sys.argv[1:], 'rkC:')
+ opts, args = getopt.getopt(sys.argv[1:], 'rkC:Q:T:m:')
for opt, arg in opts:
if opt == '-r':
ro_svr = True
@@ -146,6 +138,12 @@
keep = True
elif opt == '-C':
configfile = arg
+ elif opt == '-Q':
+ invalidation_queue_size = int(arg)
+ elif opt == '-T':
+ transaction_timeout = int(arg)
+ elif opt == '-m':
+ monitor_address = '', int(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')
@@ -156,19 +154,24 @@
zeo_port = int(args[0])
test_port = zeo_port + 1
test_addr = ('', test_port)
+ addr = ('', zeo_port)
+ serv = zodb.zeo.server.StorageServer(
+ addr, {'1': storage}, ro_svr,
+ invalidation_queue_size=invalidation_queue_size,
+ transaction_timeout=transaction_timeout,
+ monitor_address=monitor_address)
try:
logger.info('creating the test server, ro: %s, keep: %s',
ro_svr, keep)
- t = ZEOTestServer(test_addr, storage, keep)
+ t = ZEOTestServer(test_addr, serv, keep)
except socket.error, e:
if e[0] <> errno.EADDRINUSE: raise
logger.info('addr in use, closing and exiting')
storage.close()
- cleanup(storage)
+ storage.cleanup()
sys.exit(2)
addr = ('', zeo_port)
logger.info('creating the storage server')
- serv = zodb.zeo.server.StorageServer(addr, {'1': storage}, ro_svr)
t.register_socket(serv.dispatcher)
# Loop for socket events
logger.info('entering threadedasync loop')
=== Zope3/src/zodb/zeo/tests/threadtests.py 1.4 => 1.4.4.1 ===
--- Zope3/src/zodb/zeo/tests/threadtests.py:1.4 Wed Feb 5 18:28:22 2003
+++ Zope3/src/zodb/zeo/tests/threadtests.py Wed Mar 12 16:41:21 2003
@@ -19,7 +19,7 @@
from zodb.storage.tests.base import zodb_pickle, MinPO
from zodb.zeo.client import ClientStorageError
-from zodb.zeo.interfaces import Disconnected
+from zodb.zeo.interfaces import ClientDisconnected
ZERO = '\0'*8
@@ -86,7 +86,7 @@
self.gotValueError = 1
try:
self.storage.tpcAbort(self.trans)
- except Disconnected:
+ except ClientDisconnected:
self.gotDisconnected = 1
class MTStoresThread(threading.Thread):
@@ -142,24 +142,6 @@
thread2.join()
self.assertEqual(thread1.gotValueError, 1)
self.assertEqual(thread2.gotValueError, 1)
-
- def checkThatFailedBeginDoesNotHaveLock(self):
- doNextEvent = threading.Event()
- threadStartedEvent = threading.Event()
- thread1 = GetsThroughVoteThread(self._storage,
- doNextEvent, threadStartedEvent)
- thread2 = AbortsAfterBeginFailsThread(self._storage,
- doNextEvent, threadStartedEvent)
- thread1.start()
- threadStartedEvent.wait(1)
- thread2.start()
- self._storage.close()
- doNextEvent.set()
- thread1.join()
- thread2.join()
- self.assertEqual(thread1.gotValueError, 1)
- self.assertEqual(thread2.gotValueError, 1)
- self.assertEqual(thread2.gotDisconnected, 1)
# Run a bunch of threads doing small and large stores in parallel
def checkMTStores(self):
=== Zope3/src/zodb/zeo/tests/test_zeo.py 1.7.4.1 => 1.7.4.2 ===
--- Zope3/src/zodb/zeo/tests/test_zeo.py:1.7.4.1 Mon Mar 10 14:40:48 2003
+++ Zope3/src/zodb/zeo/tests/test_zeo.py Wed Mar 12 16:41:21 2003
@@ -31,26 +31,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
-from zodb.zeo.interfaces import Disconnected
-
-# 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."""
@@ -60,7 +52,7 @@
def checkZEOInvalidation(self):
addr = self._storage._addr
- storage2 = ClientStorage(addr, wait=1, min_disconnect_poll=0.1)
+ storage2 = TestClientStorage(addr, wait=True, min_disconnect_poll=0.1)
try:
oid = self._storage.newObjectId()
ob = MinPO('first')
@@ -80,59 +72,38 @@
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,
+ 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):
@@ -146,9 +117,39 @@
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.StorageWithCache,
+ cache.TransUndoStorageWithCache,
+ commitlock.CommitLockUndoTests,
+ mt.MTStorage,
+ packable.PackableStorage,
+ revision.RevisionStorage,
+ 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
@@ -157,13 +158,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
@@ -171,25 +172,35 @@
</Storage>
""" % self._envdir
+ # XXX These test seems to have massive failures when I run them.
+ # I don't think they should fail, but need Barry's help to debug.
+
+ def checkCommitLockUndoClose(self):
+ pass
+
+ def checkCommitLockUndoAbort(self):
+ pass
+
+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')
=== Zope3/src/zodb/zeo/tests/test_conn.py 1.3 => 1.3.8.1 ===
--- Zope3/src/zodb/zeo/tests/test_conn.py:1.3 Wed Jan 22 16:46:53 2003
+++ Zope3/src/zodb/zeo/tests/test_conn.py Wed Mar 12 16:41:21 2003
@@ -1,6 +1,6 @@
##############################################################################
#
-# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# Copyright (c) 2001 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
@@ -17,11 +17,10 @@
platform-dependent scaffolding.
"""
-# System imports
import unittest
-# Import the actual test class
-from zodb.zeo.tests import connection
+from zodb.zeo.tests.connection import ConnectionTests, ReconnectionTests
+from zodb.storage.base import berkeley_is_available
class FileStorageConfig:
def getConfig(self, path, create, read_only):
@@ -35,7 +34,6 @@
create and 'yes' or 'no',
read_only and 'yes' or 'no')
-
class BerkeleyStorageConfig:
def getConfig(self, path, create, read_only):
# Full always creates and doesn't have a read_only flag
@@ -47,54 +45,33 @@
</Storage>""" % (path,
read_only and 'yes' or 'no')
+class MappingStorageConfig:
+ def getConfig(self, path, create, read_only):
+ return """\
+ <Storage>
+ type MappingStorage
+ name %s
+ </Storage>""" % path
-class FileStorageConnectionTests(
- FileStorageConfig,
- connection.ConnectionTests
- ):
- """FileStorage-specific connection tests."""
-
-
-class FileStorageReconnectionTests(
- FileStorageConfig,
- connection.ReconnectionTests
- ):
- """FileStorage-specific re-connection tests."""
-
-
-class BDBConnectionTests(
- BerkeleyStorageConfig,
- connection.ConnectionTests
- ):
- """Berkeley storage connection tests."""
-
-
-class BDBReconnectionTests(
- BerkeleyStorageConfig,
- connection.ReconnectionTests
- ):
- """Berkeley storage re-connection tests."""
-
-
-test_classes = [FileStorageConnectionTests, FileStorageReconnectionTests]
-
-from zodb.storage.base import berkeley_is_available
+tests = [
+ (MappingStorageConfig, ConnectionTests, 1),
+ (FileStorageConfig, ReconnectionTests, 1),
+ (FileStorageConfig, ConnectionTests, 2),
+ ]
+
if berkeley_is_available:
- test_classes.append(BDBConnectionTests)
- test_classes.append(BDBReconnectionTests)
-
+ tests += [
+ (BerkeleyStorageConfig, ConnectionTests, 2),
+ (BerkeleyStorageConfig, ReconnectionTests, 2),
+ ]
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')
+ for testclass, configclass, level in tests:
+ # synthesize a concrete class combining tests and configuration
+ name = "%s:%s" % (testclass.__name__, configclass.__name__)
+ aclass = type.__new__(type, name, (configclass, testclass, object), {})
+ aclass.level = level
+ sub = unittest.makeSuite(aclass, "check")
suite.addTest(sub)
return suite
-
-
-if __name__ == "__main__":
- unittest.main(defaultTest='test_suite')
=== Zope3/src/zodb/zeo/tests/test_cache.py 1.2 => 1.2.12.1 ===
--- Zope3/src/zodb/zeo/tests/test_cache.py:1.2 Wed Dec 25 09:12:22 2002
+++ Zope3/src/zodb/zeo/tests/test_cache.py Wed Mar 12 16:41:21 2003
@@ -16,7 +16,6 @@
At times, we do 'white box' testing, i.e. we know about the internals
of the ClientCache object.
"""
-from __future__ import nested_scopes
import os
import time
=== Zope3/src/zodb/zeo/tests/forker.py 1.4 => 1.4.8.1 ===
--- Zope3/src/zodb/zeo/tests/forker.py:1.4 Mon Jan 27 14:44:14 2003
+++ Zope3/src/zodb/zeo/tests/forker.py Wed Mar 12 16:41:21 2003
@@ -51,8 +51,8 @@
raise RuntimeError, "Can't find port"
-def start_zeo_server(conf, addr=None, ro_svr=False, keep=False,
- admin_retries=10):
+def start_zeo_server(conf, addr=None, ro_svr=False, monitor=False, keep=False,
+ invq=None, timeout=None):
"""Start a ZEO server in a separate process.
Returns the ZEO port, the test server port, and the pid.
@@ -72,11 +72,19 @@
if script.endswith('.pyc'):
script = script[:-1]
# Create a list of arguments, which we'll tuplify below
- args = [sys.executable, script, '-C', tmpfile]
+ qa = _quote_arg
+ args = [qa(sys.executable), qa(script), '-C', qa(tmpfile)]
if ro_svr:
args.append('-r')
if keep:
args.append('-k')
+ if invq:
+ args += ['-Q', str(invq)]
+ if timeout:
+ args += ['-T', str(timeout)]
+ if monitor:
+ # XXX Is it safe to reuse the port?
+ args += ['-m', '42000']
args.append(str(port))
d = os.environ.copy()
d['PYTHONPATH'] = os.pathsep.join(sys.path)
@@ -86,7 +94,7 @@
# Always do a sleep as the first thing, since we don't expect
# the spawned process to get started right away.
delay = 0.25
- for i in range(admin_retries):
+ for i in range(10):
time.sleep(delay)
logging.debug('forker: connect %s', i)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -107,6 +115,12 @@
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
def shutdown_zeo_server(adminaddr):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
=== Zope3/src/zodb/zeo/tests/connection.py 1.5 => 1.5.4.1 ===
--- Zope3/src/zodb/zeo/tests/connection.py:1.5 Wed Feb 5 18:28:22 2003
+++ Zope3/src/zodb/zeo/tests/connection.py Wed Mar 12 16:41:21 2003
@@ -1,6 +1,6 @@
##############################################################################
#
-# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# Copyright (c) 2001 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
@@ -24,9 +24,10 @@
import logging
from zodb.zeo.client import ClientStorage
-from zodb.zeo.interfaces import Disconnected
+from zodb.zeo.interfaces import ClientDisconnected
from zodb.zeo.zrpc.marshal import Marshaller
from zodb.zeo.tests import forker
+from zodb.zeo.tests.common import TestClientStorage, DummyDB
from transaction import get_transaction
from zodb.ztransaction import Transaction
@@ -36,22 +37,17 @@
from zodb.storage.tests.base import zodb_pickle, zodb_unpickle
from zodb.storage.tests.base import handle_all_serials, ZERO
-
-class DummyDB:
- def invalidate(self, *args, **kws):
- pass
-
-
class CommonSetupTearDown(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.
- """
+ """Common boilerplate"""
__super_setUp = StorageTestBase.setUp
__super_tearDown = StorageTestBase.tearDown
+ keep = 0
+ invq = None
+ timeout = None
+ monitor = 0
+
def setUp(self):
"""Test setup for connection tests.
@@ -67,7 +63,7 @@
self._pids = []
self._servers = []
self._newAddr()
- self.startServer(keep=self.keep)
+ self.startServer()
def tearDown(self):
"""Try to cause the tests to halt"""
@@ -99,21 +95,18 @@
# port+1 is also used, so only draw even port numbers
return 'localhost', random.randrange(25000, 30000, 2)
- def getConfig(self):
- raise NotImplementedError
-
def openClientStorage(self, cache='', cache_size=200000, wait=True,
read_only=False, read_only_fallback=False,
addr=None):
if addr is None:
addr = self.addr
- storage = ClientStorage(addr,
- client=cache,
- cache_size=cache_size,
- wait=wait,
- min_disconnect_poll=0.1,
- read_only=read_only,
- read_only_fallback=read_only_fallback)
+ storage = TestClientStorage(addr,
+ client=cache,
+ cache_size=cache_size,
+ wait=wait,
+ min_disconnect_poll=0.1,
+ read_only=read_only,
+ read_only_fallback=read_only_fallback)
storage.registerDB(DummyDB())
return storage
@@ -122,15 +115,18 @@
# it fails with an error.
forker_admin_retries = 10
- def startServer(self, create=True, index=0,
- read_only=False, ro_svr=False, keep=False):
+ # Concrete test classes must provide a getConfig() method
+
+ def startServer(self, create=True, index=0, read_only=False, ro_svr=False,
+ keep=False):
addr = self.addr[index]
self.logger.warn("startServer(create=%d, index=%d, read_only=%d) @ %s",
create, index, read_only, addr)
path = "%s.%d" % (self.file, index)
conf = self.getConfig(path, create, read_only)
zeoport, adminaddr, pid = forker.start_zeo_server(
- conf, addr, ro_svr, keep, self.forker_admin_retries)
+ conf, addr, ro_svr,
+ self.monitor, self.keep, self.invq, self.timeout)
self._pids.append(pid)
self._servers.append(adminaddr)
@@ -142,11 +138,13 @@
forker.shutdown_zeo_server(adminaddr)
self._servers[index] = None
- def pollUp(self, timeout=30.0):
+ def pollUp(self, timeout=30.0, storage=None):
# Poll until we're connected
+ if storage is None:
+ storage = self._storage
now = time.time()
giveup = now + timeout
- while not self._storage.is_connected():
+ while not storage.is_connected():
asyncore.poll(0.1)
now = time.time()
if now > giveup:
@@ -164,7 +162,11 @@
class ConnectionTests(CommonSetupTearDown):
- keep = False
+ """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.
+ """
def checkMultipleAddresses(self):
for i in range(4):
@@ -193,8 +195,9 @@
try:
self._dostore()
break
- except Disconnected:
+ except ClientDisconnected:
time.sleep(0.5)
+ self._storage.sync()
def checkReadOnlyClient(self):
# Open a read-only client to a read-write server; stores fail
@@ -256,7 +259,7 @@
# Poll until the client disconnects
self.pollDown()
# Stores should fail now
- self.assertRaises(Disconnected, self._dostore)
+ self.assertRaises(ClientDisconnected, self._dostore)
# Restart the server
self.startServer(create=False)
@@ -266,12 +269,29 @@
self._dostore()
def checkDisconnectionError(self):
- # Make sure we get a Disconnected when we try to read an
+ # Make sure we get a ClientDisconnected when we try to read an
# object when we're not connected to a storage server and the
# object is not in the cache.
self.shutdownServer()
self._storage = self.openClientStorage('test', 1000, wait=False)
- self.assertRaises(Disconnected, self._storage.load, 'fredwash', '')
+ self.assertRaises(ClientDisconnected,
+ self._storage.load, 'fredwash', '')
+
+ def checkDisconnectedAbort(self):
+ self._storage = self.openClientStorage()
+ self._dostore()
+ oids = [self._storage.newObjectId() for i in range(5)]
+ txn = Transaction()
+ self._storage.tpcBegin(txn)
+ for oid in oids:
+ data = zodb_pickle(MinPO(oid))
+ self._storage.store(oid, None, data, '', txn)
+ self.shutdownServer()
+ self.assertRaises(ClientDisconnected, self._storage.tpcVote, txn)
+ self._storage.tpcAbort(txn)
+ self.startServer(create=0)
+ self._storage._wait()
+ self._dostore()
def checkBasicPersistence(self):
# Verify cached data persists across client storage instances.
@@ -335,18 +355,14 @@
try:
self._dostore(oid, data=obj)
break
- except (Disconnected, select.error,
- threading.ThreadError, socket.error):
+ except ClientDisconnected:
self.logger.warn("checkReconnection: "
"Error after server restart; retrying.",
exc_info=True)
get_transaction().abort()
- time.sleep(0.1) # XXX how long to sleep
- # XXX This is a bloody pain. We're placing a heavy burden
- # on users to catch a plethora of exceptions in order to
- # write robust code. Need to think about implementing
- # John Heintz's suggestion to make sure all exceptions
- # inherit from POSException.
+ self._storage.sync()
+ else:
+ self.fail("Could not reconnect to server")
self.logger.warn("checkReconnection: finished")
def checkBadMessage1(self):
@@ -377,7 +393,7 @@
try:
self._dostore()
- except Disconnected:
+ except ClientDisconnected:
pass
else:
self._storage.close()
@@ -427,6 +443,7 @@
class ReconnectionTests(CommonSetupTearDown):
keep = True
forker_admin_retries = 20
+ invq = 2
def checkReadOnlyStorage(self):
# Open a read-only client to a read-only *storage*; stores fail
@@ -493,7 +510,7 @@
# Poll until the client disconnects
self.pollDown()
# Stores should fail now
- self.assertRaises(Disconnected, self._dostore)
+ self.assertRaises(ClientDisconnected, self._dostore)
# Restart the server
self.startServer(create=False, read_only=True)
@@ -522,7 +539,7 @@
# Poll until the client disconnects
self.pollDown()
# Stores should fail now
- self.assertRaises(Disconnected, self._dostore)
+ self.assertRaises(ClientDisconnected, self._dostore)
# Restart the server, this time read-write
self.startServer(create=False)
@@ -554,12 +571,142 @@
try:
self._dostore()
break
- except (Disconnected, ReadOnlyError,
- select.error, threading.ThreadError, socket.error):
+ except (ClientDisconnected, ReadOnlyError):
time.sleep(0.1)
+ self._storage.sync()
else:
self.fail("Couldn't store after starting a read-write server")
+ def checkNoVerificationOnServerRestart(self):
+ self._storage = self.openClientStorage()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ self._dostore()
+ self.shutdownServer()
+ self.pollDown()
+ self._storage.verify_result = None
+ self.startServer(create=0)
+ self.pollUp()
+ # There were no transactions committed, so no verification
+ # should be needed.
+ self.assertEqual(self._storage.verify_result, "no verification")
+
+ def checkNoVerificationOnServerRestartWith2Clients(self):
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+
+ self._storage = self.openClientStorage()
+ oid = self._storage.newObjectId()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ # do two storages of the object to make sure an invalidation
+ # message is generated
+ revid = self._dostore(oid)
+ self._dostore(oid, revid)
+
+ perstorage.load(oid, '')
+
+ self.shutdownServer()
+
+ self.pollDown()
+ self._storage.verify_result = None
+ perstorage.verify_result = None
+ self.startServer(create=0)
+ self.pollUp()
+ self.pollUp(storage=perstorage)
+ # There were no transactions committed, so no verification
+ # should be needed.
+ self.assertEqual(self._storage.verify_result, "no verification")
+ self.assertEqual(perstorage.verify_result, "no verification")
+ perstorage.close()
+
+ def checkQuickVerificationWith2Clients(self):
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+
+ self._storage = self.openClientStorage()
+ oid = self._storage.newObjectId()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ # do two storages of the object to make sure an invalidation
+ # message is generated
+ revid = self._dostore(oid)
+ revid = self._dostore(oid, revid)
+
+ perstorage.load(oid, '')
+ perstorage.close()
+
+ revid = self._dostore(oid, revid)
+
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "quick verification")
+
+ self.assertEqual(perstorage.load(oid, ''),
+ self._storage.load(oid, ''))
+
+
+
+ def checkVerificationWith2ClientsInvqOverflow(self):
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+
+ self._storage = self.openClientStorage()
+ oid = self._storage.newObjectId()
+ # When we create a new storage, it should always do a full
+ # verification
+ self.assertEqual(self._storage.verify_result, "full verification")
+ # do two storages of the object to make sure an invalidation
+ # message is generated
+ revid = self._dostore(oid)
+ revid = self._dostore(oid, revid)
+
+ perstorage.load(oid, '')
+ perstorage.close()
+
+ # the test code sets invq bound to 2
+ for i in range(5):
+ revid = self._dostore(oid, revid)
+
+ perstorage = self.openClientStorage(cache="test")
+ self.assertEqual(perstorage.verify_result, "full verification")
+ t = time.time() + 30
+ while not perstorage.end_verify.isSet():
+ perstorage.sync()
+ if time.time() > t:
+ self.fail("timed out waiting for endVerify")
+
+ self.assertEqual(self._storage.load(oid, '')[1], revid)
+ self.assertEqual(perstorage.load(oid, ''),
+ self._storage.load(oid, ''))
+
+ perstorage.close()
+
+class TimeoutTests(CommonSetupTearDown):
+ timeout = 1
+
+ def checkTimeout(self):
+ storage = self.openClientStorage()
+ txn = Transaction()
+ storage.tpcBegin(txn)
+ storage.tpcVote(txn)
+ time.sleep(2)
+ self.assertRaises(ClientDisconnected, storage.tpcFinish, txn)
+
+ def checkTimeoutOnAbort(self):
+ storage = self.openClientStorage()
+ txn = Transaction()
+ storage.tpcBegin(txn)
+ storage.tpcVote(txn)
+ storage.tpcAbort(txn)
+
+ def checkTimeoutOnAbortNoLock(self):
+ storage = self.openClientStorage()
+ txn = Transaction()
+ storage.tpcBegin(txn)
+ storage.tpcAbort(txn)
class MSTThread(threading.Thread):
=== Zope3/src/zodb/zeo/tests/commitlock.py 1.3 => 1.3.4.1 ===
--- Zope3/src/zodb/zeo/tests/commitlock.py:1.3 Wed Feb 5 18:28:22 2003
+++ Zope3/src/zodb/zeo/tests/commitlock.py Wed Mar 12 16:41:21 2003
@@ -21,44 +21,18 @@
from zodb.storage.tests.base import zodb_pickle, MinPO
from zodb.zeo.client import ClientStorage
-from zodb.zeo.interfaces import Disconnected
+from zodb.zeo.interfaces import ClientDisconnected
+from zodb.zeo.tests.thread import TestThread
+from zodb.zeo.tests.common import DummyDB
ZERO = '\0'*8
-class DummyDB:
- def invalidate(self, *args):
- pass
-
-class TestThread(threading.Thread):
- __super_init = threading.Thread.__init__
- __super_run = threading.Thread.run
-
- def __init__(self, testcase, group=None, target=None, name=None,
- args=(), kwargs={}, verbose=None):
- self.__super_init(group, target, name, args, kwargs, verbose)
- self.setDaemon(1)
- self._testcase = testcase
-
- def run(self):
- try:
- self.testrun()
- except Exception:
- s = StringIO()
- traceback.print_exc(file=s)
- self._testcase.fail("Exception in thread %s:\n%s\n" %
- (self, s.getvalue()))
-
- def cleanup(self, timeout=15):
- self.join(timeout)
- if self.isAlive():
- self._testcase.fail("Thread did not finish: %s" % self)
-
class WorkerThread(TestThread):
# run the entire test in a thread so that the blocking call for
# tpc_vote() doesn't hang the test suite.
- def __init__(self, testcase, storage, trans, method="tpc_finish"):
+ def __init__(self, testcase, storage, trans, method="tpcFinish"):
self.storage = storage
self.trans = trans
self.method = method
@@ -74,53 +48,45 @@
oid = self.storage.newObjectId()
p = zodb_pickle(MinPO("c"))
self.storage.store(oid, ZERO, p, '', self.trans)
- self.ready.set()
- self.storage.tpcVote(self.trans)
- if self.method == "tpc_finish":
+ self.myvote()
+ if self.method == "tpcFinish":
self.storage.tpcFinish(self.trans)
else:
self.storage.tpcAbort(self.trans)
- except Disconnected:
+ except ClientDisconnected:
pass
-class CommitLockTests:
-
- # The commit lock tests verify that the storage successfully
- # blocks and restarts transactions when there is content for a
- # single storage. There are a lot of cases to cover.
-
- # CommitLock1 checks the case where a single transaction delays
- # other transactions before they actually block. IOW, by the time
- # the other transactions get to the vote stage, the first
- # transaction has finished.
+ def myvote(self):
+ # The vote() call is synchronous, which makes it difficult to
+ # coordinate the action of multiple threads that all call
+ # vote(). This method sends the vote call, then sets the
+ # event saying vote was called, then waits for the vote
+ # response. It digs deep into the implementation of the client.
+
+ # This method is a replacement for:
+ # self.ready.set()
+ # self.storage.tpc_vote(self.trans)
+
+ rpc = self.storage._server.rpc
+ msgid = rpc._deferred_call("tpcVote", self.storage._serial)
+ self.ready.set()
+ rpc._deferred_wait(msgid)
+ self.storage._check_serials()
- def checkCommitLock1OnCommit(self):
- self._storages = []
- try:
- self._checkCommitLock("tpc_finish", self._dosetup1, self._dowork1)
- finally:
- self._cleanup()
+class CommitLockTests:
- def checkCommitLock1OnAbort(self):
- self._storages = []
- try:
- self._checkCommitLock("tpc_abort", self._dosetup1, self._dowork1)
- finally:
- self._cleanup()
+ NUM_CLIENTS = 5
- def checkCommitLock2OnCommit(self):
- self._storages = []
- try:
- self._checkCommitLock("tpc_finish", self._dosetup2, self._dowork2)
- finally:
- self._cleanup()
-
- def checkCommitLock2OnAbort(self):
- self._storages = []
- try:
- self._checkCommitLock("tpc_abort", self._dosetup2, self._dowork2)
- finally:
- self._cleanup()
+ # The commit lock tests verify that the storage successfully
+ # blocks and restarts transactions when there is contention for a
+ # single storage. There are a lot of cases to cover. transaction
+ # has finished.
+
+ # The general flow of these tests is to start a transaction by
+ # getting far enough into 2PC to acquire the commit lock. Then
+ # begin one or more other connections that also want to commit.
+ # This causes the commit lock code to be exercised. Once the
+ # other connections are started, the first transaction completes.
def _cleanup(self):
for store, trans in self._storages:
@@ -128,77 +94,74 @@
store.close()
self._storages = []
- def _checkCommitLock(self, method_name, dosetup, dowork):
- # check the commit lock when a client attemps a transaction,
- # but fails/exits before finishing the commit.
-
- # The general flow of these tests is to start a transaction by
- # calling tpc_begin(). Then begin one or more other
- # connections that also want to commit. This causes the
- # commit lock code to be exercised. Once the other
- # connections are started, the first transaction completes.
- # Either by commit or abort, depending on whether method_name
- # is "tpc_finish."
-
- # The tests are parameterized by method_name, dosetup(), and
- # dowork(). The dosetup() function is called with a
- # connectioned client storage, transaction, and timestamp.
- # Any work it does occurs after the first transaction has
- # started, but before it finishes. The dowork() function
- # executes after the first transaction has completed.
-
- # Start on transaction normally and get the lock.
- t = Transaction()
- self._storage.tpcBegin(t)
+ def _start_txn(self):
+ txn = Transaction()
+ self._storage.tpcBegin(txn)
oid = self._storage.newObjectId()
- self._storage.store(oid, ZERO, zodb_pickle(MinPO(1)), '', t)
- self._storage.tpcVote(t)
+ self._storage.store(oid, ZERO, zodb_pickle(MinPO(1)), '', txn)
+ return oid, txn
- # Start a second transaction on a different connection without
- # blocking the test thread.
- self._storages = []
- for i in range(4):
- storage2 = self._duplicate_client()
- t2 = Transaction()
- tid = self._get_timestamp()
- dosetup(storage2, t2, tid)
- if i == 0:
- storage2.close()
- else:
- self._storages.append((storage2, t2))
+ def checkCommitLockVoteFinish(self):
+ oid, txn = self._start_txn()
+ self._storage.tpcVote(txn)
+
+ self._begin_threads()
- if method_name == "tpc_finish":
- self._storage.tpcFinish(t)
- self._storage.load(oid, '')
- else:
- self._storage.tpcAbort(t)
+ self._storage.tpcFinish(txn)
+ self._storage.load(oid, '')
- dowork(method_name)
+ self._finish_threads()
- # Make sure the server is still responsive
self._dostore()
+ self._cleanup()
+
+ def checkCommitLockVoteAbort(self):
+ oid, txn = self._start_txn()
+ self._storage.tpcVote(txn)
- def _dosetup1(self, storage, trans, tid):
- storage.tpcBegin(trans, tid)
+ self._begin_threads()
- def _dowork1(self, method_name):
- for store, trans in self._storages:
- oid = store.newObjectId()
- store.store(oid, ZERO, zodb_pickle(MinPO("c")), '', trans)
- store.tpcVote(trans)
- if method_name == "tpc_finish":
- store.tpcFinish(trans)
- else:
- store.tpcAbort(trans)
+ self._storage.tpcAbort(txn)
+
+ self._finish_threads()
+
+ self._dostore()
+ self._cleanup()
+
+ def checkCommitLockVoteClose(self):
+ oid, txn = self._start_txn()
+ self._storage.tpcVote(txn)
+
+ self._begin_threads()
+
+ self._storage.close()
- def _dosetup2(self, storage, trans, tid):
+ self._finish_threads()
+ self._cleanup()
+
+ def _begin_threads(self):
+ # Start a second transaction on a different connection without
+ # blocking the test thread.
+ self._storages = []
self._threads = []
- t = WorkerThread(self, storage, trans)
- self._threads.append(t)
- t.start()
- t.ready.wait()
+
+ for i in range(self.NUM_CLIENTS):
+ storage = self._duplicate_client()
+ txn = Transaction()
+ tid = self._get_timestamp()
+
+ t = WorkerThread(self, storage, txn)
+ self._threads.append(t)
+ t.start()
+ t.ready.wait()
+
+ # Close on the connections abnormally to test server response
+ if i == 0:
+ storage.close()
+ else:
+ self._storages.append((storage, txn))
- def _dowork2(self, method_name):
+ def _finish_threads(self):
for t in self._threads:
t.cleanup()
@@ -215,5 +178,67 @@
def _get_timestamp(self):
t = time.time()
- ts = TimeStamp(*(time.gmtime(t)[:5] + (t % 60,)))
- return ts.raw()
+ t = apply(TimeStamp,(time.gmtime(t)[:5]+(t%60,)))
+ return `t`
+
+class CommitLockUndoTests(CommitLockTests):
+
+ def _get_trans_id(self):
+ self._dostore()
+ L = self._storage.undoInfo()
+ return L[0]['id']
+
+ def _begin_undo(self, trans_id):
+ rpc = self._storage._server.rpc
+ return rpc._deferred_call("undo", trans_id, self._storage._serial)
+
+ def _finish_undo(self, msgid):
+ return self._storage._server.rpc._deferred_wait(msgid)
+
+ def checkCommitLockUndoFinish(self):
+ trans_id = self._get_trans_id()
+ oid, txn = self._start_txn()
+ msgid = self._begin_undo(trans_id)
+
+ self._begin_threads()
+
+ self._finish_undo(msgid)
+ self._storage.tpcVote(txn)
+ self._storage.tpcFinish(txn)
+ self._storage.load(oid, '')
+
+ self._finish_threads()
+
+ self._dostore()
+ self._cleanup()
+
+ def checkCommitLockUndoAbort(self):
+ trans_id = self._get_trans_id()
+ oid, txn = self._start_txn()
+ msgid = self._begin_undo(trans_id)
+
+ self._begin_threads()
+
+ self._finish_undo(msgid)
+ self._storage.tpcVote(txn)
+ self._storage.tpcAbort(txn)
+
+ self._finish_threads()
+
+ self._dostore()
+ self._cleanup()
+
+ def checkCommitLockUndoClose(self):
+ trans_id = self._get_trans_id()
+ oid, txn = self._start_txn()
+ msgid = self._begin_undo(trans_id)
+
+ self._begin_threads()
+
+ self._finish_undo(msgid)
+ self._storage.tpcVote(txn)
+ self._storage.close()
+
+ self._finish_threads()
+
+ self._cleanup()