Commit a2f9f4e2 authored by Guido van Rossum's avatar Guido van Rossum

Refactored the tests, removing some unnecessary structure.

Got rid of PackWaitWrapper (since pack now waits by default).

Moved the connection test scaffolding to testConnection.py (new);
moved the openClientStorage() implementation into ConnectionTests.py
since it is not platform dependent.
parent b053cb74
...@@ -23,7 +23,7 @@ import time ...@@ -23,7 +23,7 @@ import time
import zLOG import zLOG
import ZEO.ClientStorage from ZEO.ClientStorage import ClientStorage
from ZEO.Exceptions import Disconnected from ZEO.Exceptions import Disconnected
from ZEO.zrpc.marshal import Marshaller from ZEO.zrpc.marshal import Marshaller
...@@ -32,6 +32,10 @@ from ZODB.POSException import ReadOnlyError ...@@ -32,6 +32,10 @@ from ZODB.POSException import ReadOnlyError
from ZODB.tests import StorageTestBase from ZODB.tests import StorageTestBase
from ZODB.tests.StorageTestBase import zodb_unpickle, MinPO from ZODB.tests.StorageTestBase import zodb_unpickle, MinPO
class DummyDB:
def invalidate(self, *args):
pass
class ConnectionTests(StorageTestBase.StorageTestBase): class ConnectionTests(StorageTestBase.StorageTestBase):
"""Tests that explicitly manage the server process. """Tests that explicitly manage the server process.
...@@ -67,7 +71,16 @@ class ConnectionTests(StorageTestBase.StorageTestBase): ...@@ -67,7 +71,16 @@ class ConnectionTests(StorageTestBase.StorageTestBase):
def openClientStorage(self, cache='', cache_size=200000, wait=1, def openClientStorage(self, cache='', cache_size=200000, wait=1,
read_only=0, read_only_fallback=0): read_only=0, read_only_fallback=0):
raise NotImplementedError base = ClientStorage(self.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 = base
storage.registerDB(DummyDB(), None)
return storage
def shutdownServer(self, index=0): def shutdownServer(self, index=0):
raise NotImplementedError raise NotImplementedError
...@@ -490,4 +503,3 @@ class ConnectionTests(StorageTestBase.StorageTestBase): ...@@ -490,4 +503,3 @@ class ConnectionTests(StorageTestBase.StorageTestBase):
self._storage = self.openClientStorage() self._storage = self.openClientStorage()
self._dostore() self._dostore()
##############################################################################
#
# 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
#
##############################################################################
"""Test setup for ZEO connection logic.
The actual tests are in ConnectionTests.py; this file provides the
platform-dependent scaffolding.
"""
# 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):
"""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]
else:
raise RuntimeError, "unsupported os: %s" % os.name
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')
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment