Add support for ZEO-based unit tests with parallel execution of activities

The most simple way to use this feature is to use --activity_node option only,
but it is also possible to:
- run only a ZEO server (--activity_node=0)
- run only ZEO clients
- run only activity nodes, by specifying no test
- specify HOST:PORT to listen/connect for ZEO storage and ZServer

Load/save of catalog is done by the process running the ZEO server.
Load of static files is done by all processes. Save of static files is done by
the process running unit test.

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@35374 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/CMFActivity/ActivityTool.py

@@ -540,7 +540,7 @@ class ActivityTool (Folder, UniqueObject):
     # Filter content (ZMI))
     def filtered_meta_types(self, user=None):
         # Filters the list of available meta types.
-        all = ActivityTool.inheritedAttribute('filtered_meta_types')(self)
+        all = Folder.filtered_meta_types(self)
         meta_types = []
         for meta_type in self.all_meta_types():
             if meta_type['name'] in self.allowed_types:

product/ERP5Type/ZopePatch.py

@@ -59,6 +59,7 @@ from Products.ERP5Type.patches import StateChangeInfoPatch
 from Products.ERP5Type.patches import transforms
 from Products.ERP5Type.patches import OFSPdata
 from Products.ERP5Type.patches import make_hidden_input
+from Products.ERP5Type.patches import DemoStorage
 # BACK: Forward Compatibility with Zope 2.12 or CMF 2.2. Remove when we've
 # dropped support for older versions.
 from Products.ERP5Type.patches import TransactionAddBeforeCommitHook

product/ERP5Type/patches/DemoStorage.py

+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# Copyright (c) 2010 Nexedi SARL and Contributors. All Rights Reserved.
+# 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
+#
+##############################################################################
+
+from ZODB.DemoStorage import DemoStorage
+
+try:
+    loadEx = DemoStorage.loadEx
+except AttributeError:
+    pass # XXX Zope 2.12 ?
+else:
+    ##
+    # Fix bug in DemoStorage.loadEx (it uses 'load' instead of 'loadEx')
+    #
+    DemoStorage.loadEx = lambda *args: (loadEx(*args) + ('',))[:3]
+
+    ##
+    # Implemenent conflict resolution for DemoStorage
+    #
+    from ZODB import POSException
+    from ZODB.ConflictResolution import tryToResolveConflict, ResolvedSerial
+
+    # copied from ZODB/DemoStorage.py and patched
+    def store(self, oid, serial, data, version, transaction):
+        if transaction is not self._transaction:
+            raise POSException.StorageTransactionError(self, transaction)
+
+        self._lock_acquire()
+        try:
+            old = self._index.get(oid, None)
+            if old is None:
+                # Hm, nothing here, check the base version:
+                if self._base:
+                    try:
+                        p, tid = self._base.load(oid, '')
+                    except KeyError:
+                        pass
+                    else:
+                        old = oid, None, None, p, tid
+
+            nv=None
+            if old:
+                oid, pre, vdata, p, tid = old
+
+                if vdata:
+                    if vdata[0] != version:
+                        raise POSException.VersionLockError, oid
+
+                    nv=vdata[1]
+                else:
+                    nv=old
+
+                if serial != tid:
+                  # <patch>
+                  rdata = tryToResolveConflict(self, oid, tid, serial, data)
+                  if rdata is None:
+                    raise POSException.ConflictError(
+                        oid=oid, serials=(tid, serial), data=data)
+                  data = rdata
+                  # </patch>
+
+            r = [oid, old, version and (version, nv) or None, data, self._tid]
+            self._tindex.append(r)
+
+            s=self._tsize
+            s=s+72+(data and (16+len(data)) or 4)
+            if version: s=s+32+len(version)
+
+            if self._quota is not None and s > self._quota:
+                raise POSException.StorageError, (
+                    '''<b>Quota Exceeded</b><br>
+                    The maximum quota for this demonstration storage
+                    has been exceeded.<br>Have a nice day.''')
+
+        finally: self._lock_release()
+        # <patch>
+        if old and serial != tid:
+            return ResolvedSerial
+        # </patch>
+        return self._tid
+
+    DemoStorage.store = store
+
+    def loadSerial(self, oid, serial):
+        # XXX should I use self._lock_acquire and self._lock_release ?
+        pre = self._index.get(oid)
+        while pre:
+            oid, pre, vdata, p, tid = pre
+            if tid == serial:
+                return p
+        return self._base.loadSerial(oid, serial)
+
+    DemoStorage.loadSerial = loadSerial
+
+    def loadBefore(self, oid, tid):
+        # XXX should I use self._lock_acquire and self._lock_release ?
+        end_time = None
+        pre = self._index.get(oid)
+        while pre:
+            oid, pre, vdata, p, start_time = pre
+            if start_time < tid:
+                return p, start_time, end_time
+            end_time = start_time
+        base = self._base.loadBefore(oid, tid)
+        if base:
+            p, start_time, base_end_time = base
+            return p, start_time, base_end_time or end_time
+
+    DemoStorage.loadBefore = loadBefore
+
+    def history(self, oid, version=None, length=1, filter=None):
+        assert not version
+        self._lock_acquire()
+        try:
+            r = []
+            pre = self._index.get(oid)
+            while length and pre:
+                oid, pre, vdata, p, tid = pre
+                assert vdata is None
+                d = {'tid': tid, 'size': len(p), 'version': ''}
+                if filter is None or filter(d):
+                    r.append(d)
+                    length -= 1
+            if length:
+                r += self._base.history(oid, version, length, filter)
+            return r
+        finally:
+            self._lock_release()
+
+    DemoStorage.history = history

product/ERP5Type/tests/ERP5TypeTestCase.py

@@ -61,14 +61,15 @@ except ImportError:
 
 import transaction
 from Testing import ZopeTestCase
-from Testing.ZopeTestCase.PortalTestCase import PortalTestCase, user_name
+from Testing.ZopeTestCase import PortalTestCase, user_name
 from Products.CMFCore.utils import getToolByName
 from Products.DCWorkflow.DCWorkflow import ValidationFailed
 from Products.ERP5Type.Base import _aq_reset
 from Products.ERP5Type.Accessor.Constant import PropertyGetter as ConstantGetter
 from zLOG import LOG, DEBUG
 
-import backportUnittest
+from Products.ERP5Type.tests.backportUnittest import SetupSiteError
+from Products.ERP5Type.tests.utils import DummyMailHost, parseListeningAddress
 
 # Quiet messages when installing products
 install_product_quiet = 1
@@ -153,6 +154,9 @@ try:
 except ImportError:
   pass
 
+from Products.ERP5Type.tests.ProcessingNodeTestCase import \
+  ProcessingNodeTestCase
+
 ZopeTestCase.installProduct('TimerService', quiet=install_product_quiet)
 
 # CMF
@@ -268,7 +272,7 @@ def profile_if_environ(environment_var_name):
       # No profiling, return identity decorator
       return lambda self, method: method
 
-class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
+class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
     """TestCase for ERP5 based tests.
 
     This TestCase setups an ERP5Site and installs business templates.
@@ -566,7 +570,6 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
     def _setUpDummyMailHost(self):
       """Replace Original Mail Host by Dummy Mail Host.
       """
-      from Products.ERP5Type.tests.utils import DummyMailHost
       if 'MailHost' in self.portal.objectIds():
         self.portal.manage_delObjects(['MailHost'])
         self.portal._setObject('MailHost', DummyMailHost('MailHost'))
@@ -713,50 +716,6 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
         if rule.getValidationState() != 'validated':
           rule.validate()
 
-    def tic(self, verbose=0):
-      """
-      Start all messages
-      """
-      portal_activities = getattr(self.getPortal(),'portal_activities',None)
-      if portal_activities is not None:
-        if verbose:
-          ZopeTestCase._print('Executing pending activities ...')
-          old_message_count = 0
-          start = time.time()
-        count = 1000
-        message_count = len(portal_activities.getMessageList())
-        while message_count:
-          if verbose and old_message_count != message_count:
-            ZopeTestCase._print(' %i' % message_count)
-            old_message_count = message_count
-          portal_activities.process_timer(None, None)
-          message_count = len(portal_activities.getMessageList())
-          # This prevents an infinite loop.
-          count -= 1
-          if count == 0:
-            # Get the last error message from error_log.
-            error_message = ''
-            error_log = self.getPortal().error_log._getLog()
-            if len(error_log):
-              last_log = error_log[-1]
-              error_message = '\nLast error message:\n%s\n%s\n%s\n' % (
-                last_log['type'],
-                last_log['value'],
-                last_log['tb_text'],
-                )
-            raise RuntimeError,\
-              'tic is looping forever. These messages are pending: %r %s' % (
-            [('/'.join(m.object_path), m.method_id, m.processing_node, m.retry)
-            for m in portal_activities.getMessageList()],
-            error_message
-            )
-          # This give some time between messages
-          if count % 10 == 0:
-            from Products.CMFActivity.Activity.Queue import VALIDATION_ERROR_DELAY
-            portal_activities.timeShift(3 * VALIDATION_ERROR_DELAY)
-        if verbose:
-          ZopeTestCase._print(' done (%.3fs)\n' % (time.time() - start))
-
     def createSimpleUser(self, title, reference, function):
       """
         Helper function to create a Simple ERP5 User.
@@ -832,37 +791,6 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
       self.assertEqual(method(), reference_workflow_state)
       return workflow_error_message
 
-    def startZServer(self):
-      """Starts an HTTP ZServer thread."""
-      from Testing.ZopeTestCase import threadutils, utils
-      if utils._Z2HOST is None:
-        randint = random.Random(hash(os.environ['INSTANCE_HOME'])).randint
-        def zserverRunner():
-          try:
-            threadutils.zserverRunner(utils._Z2HOST, utils._Z2PORT)
-          except socket.error, e:
-            if e.args[0] != errno.EADDRINUSE:
-              raise
-            utils._Z2HOST = None
-        from ZServer import setNumberOfThreads
-        setNumberOfThreads(1)
-        port_list = []
-        for i in range(3):
-          utils._Z2HOST = '127.0.0.1'
-          utils._Z2PORT = randint(55000, 55500)
-          t = threadutils.QuietThread(target=zserverRunner)
-          t.setDaemon(1)
-          t.start()
-          time.sleep(0.1)
-          if utils._Z2HOST:
-            ZopeTestCase._print("Running ZServer on port %i\n" % utils._Z2PORT)
-            break
-          port_list.append(str(utils._Z2PORT))
-        else:
-          ZopeTestCase._print("Can't find free port to start ZServer"
-                              " (tried ports %s)\n" % ', '.join(port_list))
-      return utils._Z2HOST, utils._Z2PORT
-
     def _installBusinessTemplateList(self, business_template_list,
                                      light_install=True,
                                      quiet=True):
@@ -921,7 +849,7 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
       title = self.getTitle()
       from Products.ERP5Type.Base import _aq_reset
       if portal_name in failed_portal_installation:
-        raise backportUnittest.SetupSiteError(
+        raise SetupSiteError(
             'Installation of %s already failed, giving up' % portal_name)
       try:
         if app is None:
@@ -930,6 +858,7 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
         # make it's REQUEST available during setup
         global current_app
         current_app = app
+        app.test_portal_name = portal_name
 
         global setup_done
         if not (hasattr(aq_base(app), portal_name) and
@@ -992,6 +921,7 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
               except ImportError:
                 pass
               self.serverhost, self.serverport = self.startZServer()
+              self._registerNode(distributing=1, processing=1)
 
             self._updateConversionServerConfiguration()
             self._updateConnectionStrings()
@@ -1073,11 +1003,6 @@ class ERP5TypeTestCase(backportUnittest.TestCase, PortalTestCase):
         if count:
           LOG('Products.ERP5Type.tests.ERP5TypeTestCase.beforeClose', DEBUG,
               'dropped %d left-over activity messages' % (count,))
-        # portal_activities.process_timer automatically registers current node
-        # (localhost:<random_port>). We must unregister it so that Data.fs can
-        # be reused without reconfiguring portal_activities.
-        del portal_activities.distributingNode
-        del portal_activities._nodes
         transaction.commit()
       except AttributeError:
         pass
@@ -1248,11 +1173,6 @@ class ERP5ReportTestCase(ERP5TypeTestCase):
     if diff_list:
       self.fail('Lines differs:\n' + '\n'.join(diff_list))
 
-from unittest import _makeLoader, TestSuite
-
-def dummy_makeSuite(testCaseClass, prefix='dummy_test', sortUsing=cmp, suiteClass=TestSuite):
-  return _makeLoader(prefix, sortUsing, suiteClass).loadTestsFromTestCase(testCaseClass)
-
 def dummy_setUp(self):
   '''
   This one is overloaded so that it dos not execute beforeSetUp and afterSetUp
@@ -1276,6 +1196,34 @@ def dummy_tearDown(self):
   '''
   self._clear(1)
 
+class ZEOServerTestCase(ERP5TypeTestCase):
+  """TestCase class to run a ZEO storage
+
+  Main method is 'asyncore_loop' (inherited) since there is nothing to do
+  except processing I/O.
+  """
+
+  def setUp(self):
+    # Start ZEO storage and send address to parent process if any.
+    from Zope2.custom_zodb import zeo_client, Storage
+    from ZEO.StorageServer import StorageServer
+    storage = {'1': Storage}
+    for host_port in parseListeningAddress(os.environ.get('zeo_server')):
+      try:
+        self.zeo_server = StorageServer(host_port, storage)
+        break
+      except socket.error, e:
+        if e[0] != errno.EADDRINUSE:
+          raise
+    if zeo_client:
+      os.write(zeo_client, repr(host_port))
+      os.close(zeo_client)
+    ZopeTestCase._print("\nZEO Storage started at %s:%s ... " % host_port)
+
+  def tearDown(self):
+    self.zeo_server.close_server()
+
+
 @onsetup
 def optimize():
   '''Significantly reduces portal creation time.'''

product/ERP5Type/tests/ProcessingNodeTestCase.py

+# This module must be imported before CMFActivity product is installed.
+
+import base64, errno, select, socket, time
+from threading import Thread
+import Lifetime
+import transaction
+from BTrees.OIBTree import OIBTree
+from Testing import ZopeTestCase
+from Products.CMFActivity import ActivityTool as _ActivityTool
+from Products.CMFActivity.Activity.Queue import VALIDATION_ERROR_DELAY
+from Products.ERP5Type.tests import backportUnittest
+from Products.ERP5Type.tests.utils import createZServer
+
+
+class ActivityTool(_ActivityTool.ActivityTool):
+  """Class redefining CMFActivity.ActivityTool.ActivityTool for unit tests
+  """
+
+  # When a ZServer can't be started, the node name ends with ':' (no port).
+  def _isValidNodeName(self, node_name):
+    return True
+
+  # Divert location to register processing and distributing nodes.
+  # Load balancing is configured at the root instead of the activity tool,
+  # so that additional can register even if there is no portal set up yet.
+  # Properties at the root are:
+  # - 'test_processing_nodes' to list processing nodes
+  # - 'test_distributing_node' to select the distributing node
+  def getNodeDict(self):
+    app = self.getPhysicalRoot()
+    if getattr(app, 'test_processing_nodes', None) is None:
+      app.test_processing_nodes = OIBTree()
+    return app.test_processing_nodes
+
+  def getDistributingNode(self):
+    return self.getPhysicalRoot().test_distributing_node
+
+  def manage_setDistributingNode(self, distributingNode, REQUEST=None):
+    # A property to catch setattr on 'distributingNode' doesn't work
+    # because self would lose all acquisition wrappers.
+    previous_node = self.distributingNode
+    try:
+      super(ActivityTool, self).manage_setDistributingNode(distributingNode,
+                                                           REQUEST=REQUEST)
+      self.getPhysicalRoot().test_distributing_node = self.distributingNode
+    finally:
+      self.distributingNode = previous_node
+
+  # When there is more than 1 node, prevent the distributing node from
+  # processing activities.
+  def tic(self, processing_node=1, force=0):
+    processing_node_list = self.getProcessingNodeList()
+    if len(processing_node_list) > 1 and \
+       self.getCurrentNode() == self.getDistributingNode():
+      # Sleep between each distribute.
+      time.sleep(0.3)
+      transaction.commit()
+    else:
+      super(ActivityTool, self).tic(processing_node, force)
+
+_ActivityTool.ActivityTool = ActivityTool
+
+
+class ProcessingNodeTestCase(backportUnittest.TestCase, ZopeTestCase.TestCase):
+  """Minimal ERP5 TestCase class to process activities
+
+  When a processing node starts, the portal may not exist yet, or its name is
+  unknown, so an additional 'test_portal_name' property at the root is set by
+  the node running the unit tests to tell other nodes on which portal activities
+  should be processed.
+  """
+
+  @staticmethod
+  def asyncore_loop():
+    try:
+      Lifetime.lifetime_loop()
+    except KeyboardInterrupt:
+      pass
+    Lifetime.graceful_shutdown_loop()
+
+  def startZServer(self):
+    """Start HTTP ZServer in background"""
+    utils = ZopeTestCase.utils
+    if utils._Z2HOST is None:
+      try:
+        hs = createZServer()
+      except RuntimeError, e:
+        ZopeTestCase._print(str(e))
+      else:
+        utils._Z2HOST, utils._Z2PORT = hs.server_name, hs.server_port
+        t = Thread(target=Lifetime.loop)
+        t.setDaemon(1)
+        t.start()
+    return utils._Z2HOST, utils._Z2PORT
+
+  def _registerNode(self, distributing, processing):
+    """Register node to process and/or distribute activities"""
+    try:
+      activity_tool = self.portal.portal_activities
+    except AttributeError:
+      activity_tool = ActivityTool().__of__(self.app)
+    currentNode = activity_tool.getCurrentNode()
+    if distributing:
+      activity_tool.manage_setDistributingNode(currentNode)
+    if processing:
+      activity_tool.manage_addToProcessingList((currentNode,))
+    else:
+      activity_tool.manage_removeFromProcessingList((currentNode,))
+
+  def tic(self, verbose=0):
+    """Execute pending activities"""
+    portal_activities = self.portal.portal_activities
+    if 1:
+      if verbose:
+        ZopeTestCase._print('Executing pending activities ...')
+        old_message_count = 0
+        start = time.time()
+      count = 1000
+      getMessageList = portal_activities.getMessageList
+      message_count = len(getMessageList(include_processing=1))
+      while message_count:
+        if verbose and old_message_count != message_count:
+          ZopeTestCase._print(' %i' % message_count)
+          old_message_count = message_count
+        portal_activities.process_timer(None, None)
+        if Lifetime._shutdown_phase:
+          # XXX CMFActivity contains bare excepts
+          raise KeyboardInterrupt
+        message_count = len(getMessageList(include_processing=1))
+        # This prevents an infinite loop.
+        count -= 1
+        if count == 0:
+          # Get the last error message from error_log.
+          error_message = ''
+          error_log = self.portal.error_log._getLog()
+          if len(error_log):
+            last_log = error_log[-1]
+            error_message = '\nLast error message:\n%s\n%s\n%s\n' % (
+              last_log['type'],
+              last_log['value'],
+              last_log['tb_text'],
+              )
+          raise RuntimeError,\
+            'tic is looping forever. These messages are pending: %r %s' % (
+          [('/'.join(m.object_path), m.method_id, m.processing_node, m.retry)
+          for m in portal_activities.getMessageList()],
+          error_message
+          )
+        # This give some time between messages
+        if count % 10 == 0:
+          portal_activities.timeShift(3 * VALIDATION_ERROR_DELAY)
+      if verbose:
+        ZopeTestCase._print(' done (%.3fs)\n' % (time.time() - start))
+
+  def afterSetUp(self):
+    """Initialize a node that will only process activities"""
+    createZServer() #self.startZServer()
+    self._registerNode(distributing=0, processing=1)
+    transaction.commit()
+
+  def processing_node(self):
+    """Main loop for nodes that process activities"""
+    try:
+      while not Lifetime._shutdown_phase:
+        time.sleep(.3)
+        transaction.begin()
+        try:
+          portal = self.app[self.app.test_portal_name]
+        except AttributeError:
+          continue
+        portal.portal_activities.process_timer(None, None)
+    except KeyboardInterrupt:
+      pass

product/ERP5Type/tests/backportUnittest.py

@@ -107,8 +107,6 @@ class TestCase(unittest.TestCase):
       _testMethodDoc = property(lambda self: self.__testMethodDoc)
 
     def run(self, result=None):
-        import pdb
-        #pdb.set_trace()
         orig_result = result
         if result is None:
             result = self.defaultTestResult()
@@ -138,6 +136,8 @@ class TestCase(unittest.TestCase):
                 result.addSkip(self, str(e))
             except SetupSiteError, e:
                 result.errors.append(None)
+            except (KeyboardInterrupt, SystemExit): # BACK: Not needed for
+                raise                               #       Python >= 2.5
             except Exception:
                 result.addError(self, sys.exc_info())
             else:
@@ -151,6 +151,8 @@ class TestCase(unittest.TestCase):
                     result.addUnexpectedSuccess(self)
                 except SkipTest, e:
                     result.addSkip(self, str(e))
+                except (KeyboardInterrupt, SystemExit): # BACK: Not needed for
+                    raise                               #       Python >= 2.5
                 except Exception:
                     result.addError(self, sys.exc_info())
                 else:
@@ -158,6 +160,8 @@ class TestCase(unittest.TestCase):
 
                 try:
                     self.tearDown()
+                except (KeyboardInterrupt, SystemExit): # BACK: Not needed for
+                    raise                               #       Python >= 2.5
                 except Exception:
                     result.addError(self, sys.exc_info())
                     success = False
@@ -252,7 +256,10 @@ class TextTestRunner(unittest.TextTestRunner):
         result = self._makeResult()
         startTime = time.time()
         # BACK: 2.7 implementation wraps run with result.(start|stop)TestRun
-        test(result)
+        try:
+          test(result)
+        except KeyboardInterrupt:
+          pass
         stopTime = time.time()
         timeTaken = stopTime - startTime
         result.printErrors()

product/ERP5Type/tests/custom_zodb.py

 import os
 import shutil
+import socket
 import sys
 import glob
+import threading
 import ZODB
+from asyncore import socket_map
 from ZODB.DemoStorage import DemoStorage
 from ZODB.FileStorage import FileStorage
-from Products.ERP5Type.tests.utils import getMySQLArguments
+from ZEO.ClientStorage import ClientStorage
+from Products.ERP5Type.tests.utils import getMySQLArguments, instance_random
 from Products.ERP5Type.tests.runUnitTest import instance_home, static_dir_list
 
 def _print(message):
   sys.stderr.write(message + "\n")
 
+zserver_list = os.environ.get('zserver', '').split(',')
+os.environ['zserver'] = zserver_list[0]
+
+zeo_client = os.environ.get('zeo_client')
+if zeo_client:
+  zeo_client = zeo_client.rsplit(':', 1)
+  zeo_client = (len(zeo_client) == 1 and 'localhost' or zeo_client[0],
+                int(zeo_client[-1]))
+try:
+  activity_node = int(os.environ['activity_node'])
+except KeyError:
+  activity_node = (zeo_client or 'zeo_server' in os.environ) and 1 or None
+
 data_fs_path = os.environ.get('erp5_tests_data_fs_path',
                               os.path.join(instance_home, 'Data.fs'))
 load = int(os.environ.get('erp5_load_data_fs', 0))
 save = int(os.environ.get('erp5_save_data_fs', 0))
 
+save_mysql = None
+if not zeo_client:
+  def save_mysql(verbosity=1):
+    # The output of mysqldump needs to merge many lines at a time
+    # for performance reasons (merging lines is at most 10 times
+    # faster, so this produce somewhat not nice to read sql
+    command = 'mysqldump %s > dump.sql' % getMySQLArguments()
+    if verbosity:
+      _print('Dumping MySQL database with %s...' % command)
+    os.system(command)
+
 _print("Cleaning static files ... ")
 for dir in static_dir_list:
   for f in glob.glob(os.path.join(instance_home, dir, '*')):
     os.remove(f)
 
 if load:
-  dump_sql = os.path.join(instance_home, 'dump.sql')
-  if os.path.exists(dump_sql):
-    _print("Restoring MySQL database ... ")
-    ret = os.system("mysql %s < %s" % (getMySQLArguments(), dump_sql))
-    assert not ret
-  else:
-    os.environ['erp5_tests_recreate_catalog'] = '1'
+  if save_mysql:
+    dump_sql = os.path.join(instance_home, 'dump.sql')
+    if os.path.exists(dump_sql):
+      _print("Restoring MySQL database ... ")
+      ret = os.system("mysql %s < %s" % (getMySQLArguments(), dump_sql))
+      assert not ret
+    else:
+      os.environ['erp5_tests_recreate_catalog'] = '1'
   _print("Restoring static files ... ")
   for dir in static_dir_list:
     full_path = os.path.join(instance_home, dir)
     if os.path.exists(full_path + '.bak'):
       os.rmdir(full_path)
       shutil.copytree(full_path + '.bak', full_path, symlinks=True)
-elif save and os.path.exists(data_fs_path):
+elif save and not zeo_client and os.path.exists(data_fs_path):
   os.remove(data_fs_path)
 
-if save:
-  Storage = FileStorage(data_fs_path)
-elif load:
-  Storage = DemoStorage(base=FileStorage(data_fs_path))
+zeo_server_pid = None
+zeo_client_pid_list = []
+
+ZEvent = sys.modules.get('ZServer.PubCore.ZEvent')
+def fork():
+  pid = os.fork()
+  if pid:
+    # recreate the event pipe if it already exists
+    for obj in socket_map.values():
+      assert obj is ZEvent.the_trigger
+      obj.close()
+      ZEvent.the_trigger = ZEvent.simple_trigger()
+    # make sure parent and child have 2 different RNG
+    instance_random.seed(instance_random.random())
+  return pid
+
+while not zeo_client:
+  if activity_node:
+    r, zeo_client = os.pipe()
+    zeo_server_pid = fork()
+    if zeo_server_pid:
+      save_mysql = None
+      os.close(zeo_client)
+      zeo_client = eval(os.fdopen(r).read())
+      continue
+    else:
+      zeo_client_pid_list = activity_node = None
+      os.close(r)
+  elif activity_node is not None:
+    # run ZEO server but no need to fork
+    zeo_server_pid = 0
+
+  if save:
+    Storage = FileStorage(data_fs_path)
+  elif load:
+    Storage = DemoStorage(base=FileStorage(data_fs_path))
+  else:
+    Storage = DemoStorage()
+  break
 else:
-  Storage = DemoStorage()
+  for i in xrange(1, activity_node):
+    pid = fork()
+    if not pid:
+      zeo_client_pid_list = None
+      os.environ['zserver'] = i < len(zserver_list) and zserver_list[i] or ''
+      break
+    zeo_client_pid_list.append(pid)
+  Storage = ClientStorage(zeo_client)
 
-_print("Instance at %r loaded ... " % instance_home)
+if zeo_client_pid_list is not None:
+  _print("Instance at %r loaded ... " % instance_home)

product/ERP5Type/tests/utils.py

@@ -28,10 +28,12 @@
 
 """Utility functions and classes for unit testing
 """
-
+import errno
 import os
 import logging
-
+import random
+import socket
+import sys
 import transaction
 import zLOG
 import Products.ERP5Type
@@ -249,6 +251,58 @@ def getExtraSqlConnectionStringList():
   return os.environ.get('extra_sql_connection_string_list',
                         'test2 test2:test3 test3').split(':')
 
+instance_random = random.Random(hash(os.environ['INSTANCE_HOME']))
+
+def parseListeningAddress(host_port=None, default_host='127.0.0.1'):
+  """Parse string specifying the address to bind to
+
+  If the specified address is incomplete or missing, several (host, random_port)
+  will be returned. It must be used as follows (an appropriate error is raised
+  if all returned values failed):
+
+    for host, port in parseListeningAddress(os.environ.get('some_address')):
+      try:
+        s.bind((host, port))
+        break
+      except socket.error, e:
+        if e[0] != errno.EADDRINUSE:
+          raise
+  """
+  if host_port:
+    host_port = host_port.rsplit(':', 1)
+    if len(host_port) == 1:
+      host_port = default_host, host_port[0]
+    try:
+      yield host_port[0], int(host_port[1])
+      raise RuntimeError("Can't bind to %s:%s" % host_port)
+    except ValueError:
+      default_host = host_port[1]
+  port_list = []
+  for i in xrange(3):
+    port_list.append(instance_random.randint(55000, 55500))
+    yield default_host, port_list[-1]
+  raise RuntimeError("Can't find free port (tried ports %s)\n"
+                     % ', '.join(map(str, port_list)))
+
+def createZServer(log=os.devnull):
+  from ZServer import logger, zhttp_server, zhttp_handler
+  lg = logger.file_logger(log)
+  class new_zhttp_server:
+    # I can't use __new__ because zhttp_handler is an old-style class :(
+    def __init__(self):
+      self.__class__ = zhttp_server
+  for ip, port in parseListeningAddress(os.environ.get('zserver')):
+    hs = new_zhttp_server()
+    try:
+      hs.__init__(ip, port, resolver=None, logger_object=lg)
+      hs.install_handler(zhttp_handler(module='Zope2', uri_base=''))
+      sys.stderr.write("Running ZServer at %s:%s\n" % (ip, port))
+      return hs
+    except socket.error, e:
+      if e[0] != errno.EADDRINUSE:
+        raise
+      hs.close()
+
 # decorators
 class reindex(object):
   """Decorator to commit transaction and flush activities after the method is