Commit 87699fc9 authored by Barry Warsaw's avatar Barry Warsaw

Since the Unix and Windows tests share a common infrastructure now,

rename UnixTests to FileStorageTests, and add a simple BDBTests class
to use Berkeley Full storage.

Rewrite to use the new forker module api.

WindowsTests: Removed.
parent 7468a492
...@@ -126,27 +126,26 @@ class GenericTests( ...@@ -126,27 +126,26 @@ class GenericTests(
ReadOnlyStorage.ReadOnlyStorage.checkWriteMethods(self) ReadOnlyStorage.ReadOnlyStorage.checkWriteMethods(self)
class UnixTests(GenericTests): class FileStorageTests(GenericTests):
"""Test ZEO backed by a FileStorage."""
"""Add Unix-specific scaffolding to the generic test suite."""
def setUp(self): def setUp(self):
zLOG.LOG("testZEO", zLOG.INFO, "setUp() %s" % self.id()) 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._pids = [pid]
self._servers = [exit] self._servers = [adminaddr]
self._storage = client self._storage = ClientStorage(zeoport, '1', cache_size=20000000,
client.registerDB(DummyDB(), None) min_disconnect_poll=0.5, wait=1)
self._storage.registerDB(DummyDB(), None)
def tearDown(self): def tearDown(self):
self._storage.close() self._storage.close()
for server in self._servers: for server in self._servers:
server.close() forker.shutdown_zeo_server(server)
for pid in self._pids: for pid in self._pids:
os.waitpid(pid, 0) os.waitpid(pid, 0)
self.delStorage()
def getStorage(self): def getConfig(self):
filename = self.__fs_base = tempfile.mktemp() filename = self.__fs_base = tempfile.mktemp()
# Return a 1-tuple # Return a 1-tuple
return """\ return """\
...@@ -155,16 +154,11 @@ class UnixTests(GenericTests): ...@@ -155,16 +154,11 @@ class UnixTests(GenericTests):
file_name %s file_name %s
create yes create yes
</Storage> </Storage>
""" % filename, """ % filename
def delStorage(self):
from ZODB.FileStorage import cleanup
cleanup(self.__fs_base)
class BDBTests(UnixTests):
"""Add Berkeley storage tests (not sure if these are Unix specific).""" class BDBTests(FileStorageTests):
"""ZEO backed by a Berkeley Full storage."""
def getStorage(self): def getStorage(self):
self._envdir = tempfile.mktemp() self._envdir = tempfile.mktemp()
...@@ -174,45 +168,11 @@ class BDBTests(UnixTests): ...@@ -174,45 +168,11 @@ class BDBTests(UnixTests):
type Full type Full
name %s name %s
</Storage> </Storage>
""" % self._envdir, """ % self._envdir
def delStorage(self):
from bsddb3Storage.BerkeleyBase import cleanup
cleanup(self._envdir)
class WindowsTests(UnixTests):
"""Add Windows-specific scaffolding to the generic test suite."""
def setUp(self):
zLOG.LOG("testZEO", zLOG.INFO, "setUp() %s" % self.id())
args = self.getStorage()
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()
if os.name == "posix":
test_classes = [UnixTests]
elif os.name == "nt":
test_classes = [WindowsTests]
else:
raise RuntimeError, "unsupported os: %s" % os.name
test_classes = [FileStorageTests]
try: try:
from bsddb3Storage.Full import Full from bsddb3Storage.Full import Full
except ImportError: except ImportError:
......
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