############################################################################## # # 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.1 (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 suite for ZEO based on ZODB.tests.""" # System imports import os import random import socket import asyncore import tempfile import unittest import logging # ZODB test support import ZODB from ZODB.tests.MinPO import MinPO from ZODB.tests.StorageTestBase import zodb_unpickle # ZODB test mixin classes from ZODB.tests import StorageTestBase, BasicStorage, VersionStorage, \ TransactionalUndoStorage, TransactionalUndoVersionStorage, \ PackableStorage, Synchronization, ConflictResolution, RevisionStorage, \ MTStorage, ReadOnlyStorage from ZEO.ClientStorage import ClientStorage from ZEO.tests import forker, Cache, CommitLockTests, ThreadTests logger = logging.getLogger('ZEO.tests.testZEO') class DummyDB: def invalidate(self, *args): pass class MiscZEOTests: """ZEO tests that don't fit in elsewhere.""" def checkLargeUpdate(self): obj = MinPO("X" * (10 * 128 * 1024)) self._dostore(data=obj) def checkZEOInvalidation(self): addr = self._storage._addr storage2 = ClientStorage(addr, wait=1, min_disconnect_poll=0.1) try: oid = self._storage.new_oid() ob = MinPO('first') revid1 = self._dostore(oid, data=ob) data, serial = storage2.load(oid, '') self.assertEqual(zodb_unpickle(data), MinPO('first')) self.assertEqual(serial, revid1) revid2 = self._dostore(oid, data=MinPO('second'), revid=revid1) for n in range(3): # Let the server and client talk for a moment. # Is there a better way to do this? asyncore.poll(0.1) data, serial = storage2.load(oid, '') self.assertEqual(zodb_unpickle(data), MinPO('second'), 'Invalidation message was not sent!') self.assertEqual(serial, revid2) finally: storage2.close() def get_port(): """Return a port that is not in use. Checks if a port is in use by trying to connect to it. Assumes it is not in use if connect raises an exception. Raises RuntimeError after 10 tries. """ for i in range(10): port = random.randrange(20000, 30000) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: try: s.connect(('localhost', port)) except socket.error: # XXX check value of error? return port finally: s.close() raise RuntimeError, "Can't find port" class GenericTests( # Base class for all ZODB tests StorageTestBase.StorageTestBase, # ZODB test mixin classes (in the same order as imported) BasicStorage.BasicStorage, PackableStorage.PackableStorage, Synchronization.SynchronizedStorage, MTStorage.MTStorage, ReadOnlyStorage.ReadOnlyStorage, # ZEO test mixin classes (in the same order as imported) CommitLockTests.CommitLockVoteTests, ThreadTests.ThreadTests, # Locally defined (see above) MiscZEOTests ): """Combine tests from various origins in one class.""" def setUp(self): logger.info("setUp() %s", self.id()) # XXX is this really needed? port = get_port() zconf = forker.ZEOConfig(('', port)) zport, adminaddr, pid, path = forker.start_zeo_server(self.getConfig(), zconf, port) self._pids = [pid] self._servers = [adminaddr] self._conf_path = path self._storage = ClientStorage(zport, '1', cache_size=20000000, min_disconnect_poll=0.5, wait=1, wait_timeout=60) self._storage.registerDB(DummyDB(), None) def tearDown(self): self._storage.close() os.remove(self._conf_path) for server in self._servers: 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 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) def checkWriteMethods(self): # ReadOnlyStorage defines checkWriteMethods. The decision # about where to raise the read-only error was changed after # Zope 2.5 was released. So this test needs to detect Zope # of the 2.5 vintage and skip the test. # The __version__ attribute was not present in Zope 2.5. if hasattr(ZODB, "__version__"): ReadOnlyStorage.ReadOnlyStorage.checkWriteMethods(self) def checkSortKey(self): key = '%s:%s' % (self._storage._storage, self._storage._server_addr) self.assertEqual(self._storage.sortKey(), key) class FullGenericTests( GenericTests, Cache.StorageWithCache, Cache.TransUndoStorageWithCache, CommitLockTests.CommitLockUndoTests, ConflictResolution.ConflictResolvingStorage, ConflictResolution.ConflictResolvingTransUndoStorage, PackableStorage.PackableUndoStorage, RevisionStorage.RevisionStorage, TransactionalUndoStorage.TransactionalUndoStorage, TransactionalUndoVersionStorage.TransactionalUndoVersionStorage, VersionStorage.VersionStorage, ): """Extend GenericTests with tests that MappingStorage can't pass.""" class FileStorageTests(FullGenericTests): """Test ZEO backed by a FileStorage.""" level = 2 def getConfig(self): filename = self.__fs_base = tempfile.mktemp() return """\ <filestorage 1> path %s </filestorage> """ % filename class BDBTests(FullGenericTests): """ZEO backed by a Berkeley full storage.""" level = 2 def getConfig(self): self._envdir = tempfile.mktemp() return """\ <fullstorage 1> envdir %s </fullstorage> """ % self._envdir class MappingStorageTests(GenericTests): """ZEO backed by a Mapping storage.""" def getConfig(self): return """<mappingstorage 1/>""" # XXX There are still a bunch of tests that fail. Are there # still test classes in GenericTests that shouldn't be there? # XXX Is the above comment still relevant? test_classes = [FileStorageTests, MappingStorageTests] def test_suite(): 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")