[Zodb-checkins] CVS: Packages/ZEO - testZEO.py:1.1.2.1

[jeremy@digicool.com]

Update of /cvs-repository/Packages/ZEO/tests
In directory korak:/tmp/cvs-serv22621

Added Files:
      Tag: ZEO-ZRPC-Dev
	testZEO.py 
Log Message:
An set of tests for ZEO.  Only some of the tests succeed.



--- Added File testZEO.py in package Packages/ZEO ---
"""Test suite for ZEO based on ZODB.tests"""

import asyncore
import os
import tempfile
import threading
import unittest

import ZEO.ClientStorage, ZEO.StorageServer
import ThreadedAsync, ZEO.trigger

import zLOG

# XXX The ZODB.tests package contains a grab bad things, including,
# apparently, a collection of modules that define mixin classes
# containing tests cases.

from ZODB.tests import StorageTestBase, BasicStorage, VersionStorage

class MainloopThread(threading.Thread):
    __super_init = threading.Thread.__init__

    def __init__(self, *args, **kw):
        self.ready_lock = threading.Lock()
        self.ready_lock.acquire()
        self.__super_init(*args, **kw)
    
    def run(self):
        ThreadedAsync.register_loop_callback(self.callback)
        asyncore.loop(timeout=1.0)

    def callback(self, map):
        self.ready_lock.release()

class GenericTests(StorageTestBase.StorageTestBase,
                   BasicStorage.BasicStorage,
##                   VersionStorage.VersionStorage
                   ):
    """An abstract base class for ZEO tests

    A specific ZEO test run depends on having a real storage that the
    StorageServer provides access to.  The GenericTests must be
    subclassed to provide an implementation of getStorage() that
    returns a specific storage, e.g. FileStorage.

    The storage server mainloop is run in a separate thread so that
    it does not interfere with the thread running the tests.  The
    tearDown() method clears asyncore's socket_map dict, which causes
    asyncore.loop() to return and the thread to exit.
    """

    __super_setUp = StorageTestBase.StorageTestBase.setUp
    __super_tearDown = StorageTestBase.StorageTestBase.tearDown

    def setUp(self):
        """Start a ZEO server using a Unix domain socket

        The ZEO server uses the storage object returned by the
        getStorage() method.
        """
        self.running = 1
        self.__trigger = ZEO.trigger.trigger()
        self.__map = asyncore.socket_map
        ThreadedAsync.register_loop_callback(self.start_loop)
        self.__storage = self.getStorage()
        self.__sock = tempfile.mktemp()
        self.__server = ZEO.StorageServer.StorageServer(self.__sock,
                                                        {'1': self.__storage})
        t = MainloopThread()
        t.start()
        # wait for mainloop to start
        t.ready_lock.acquire()
        self._storage = ZEO.ClientStorage.ClientStorage(self.__sock)
        self.__super_setUp()

    def tearDown(self):
        """Try to cause the tests to halt"""
        zLOG.LOG("test", zLOG.INFO, "tearDown()")
        self.running = 0
        self.stop_loop()
        self.__storage.close()
        os.unlink(self.__sock)
        self.__super_tearDown()

    def start_loop(self, map):
##        zLOG.LOG("test", zLOG.INFO, "start_loop(%s)" % repr(map))
        self.__map = map
        if not self.running:
            self.stop_loop()

    def stop_loop(self):
##        zLOG.LOG("test", zLOG.INFO, "map = %s" % id(self.__map))
        self.__map.clear()
        self.__trigger.pull_trigger()

    def checkFirst(self):
        self._storage.tpc_begin(self._transaction)
        self._storage.tpc_abort(self._transaction)

class ZEOFileStorageTests(GenericTests):
    __super_setUp = GenericTests.setUp
    __super_tearDown = GenericTests.tearDown
    
    from ZODB.FileStorage import FileStorage

    def setUp(self):
        self.__fs_base = tempfile.mktemp()
        self.__super_setUp()

    def tearDown(self):
        self.__super_tearDown()
        # file storage appears to create three files
        for ext in '', '.index', '.lock', '.tmp':
            path = self.__fs_base + ext
            os.unlink(path)
    
    def getStorage(self):
        return self.FileStorage(self.__fs_base, create=1)
        
def main():
    tests = unittest.makeSuite(ZEOFileStorageTests, 'check')
    runner = unittest.TextTestRunner()
    runner.run(tests)

if __name__ == "__main__":
    main()