Rename two tests that had identical names to distinguish them.

One of these (testExecuteFailuresEngineService_0) fails often, though
not always.  I spent hours trying to debug what's going on but failed,
and it's non-deterministic.

IPython/kernel/tests/engineservicetest.py

@@ -1,372 +1,373 b''
 # encoding: utf-8
 
 """Test template for complete engine object"""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import cPickle as pickle
 
 from twisted.internet import defer, reactor
 from twisted.python import failure
 from twisted.application import service
 import zope.interface as zi
 
 from IPython.kernel import newserialized
 from IPython.kernel import error
 from IPython.kernel.pickleutil import can, uncan
 import IPython.kernel.engineservice as es
 from IPython.kernel.core.interpreter import Interpreter
 from IPython.testing.parametric import Parametric, parametric
 
 #-------------------------------------------------------------------------------
 # Tests
 #-------------------------------------------------------------------------------
 
     
 # A sequence of valid commands run through execute
 validCommands = ['a=5',
                  'b=10',
                  'a=5; b=10; c=a+b',
                  'import math; 2.0*math.pi',
                  """def f():
     result = 0.0
     for i in range(10):
         result += i
 """,
                  'if 1<2: a=5',
                  """import time
 time.sleep(0.1)""",
                 """from math import cos;
 x = 1.0*cos(0.5)""", # Semicolons lead to Discard ast nodes that should be discarded
                 """from sets import Set
 s = Set()
    """, # Trailing whitespace should be allowed.
                 """import math
 math.cos(1.0)""", # Test a method call with a discarded return value
                 """x=1.0234
 a=5; b=10""", # Test an embedded semicolon
                 """x=1.0234
 a=5; b=10;""" # Test both an embedded and trailing semicolon
                  ]
                  
 # A sequence of commands that raise various exceptions
 invalidCommands = [('a=1/0',ZeroDivisionError),
                    ('print v',NameError),
                    ('l=[];l[0]',IndexError),
                    ("d={};d['a']",KeyError),
                    ("assert 1==0",AssertionError),
                    ("import abababsdbfsbaljasdlja",ImportError),
                    ("raise Exception()",Exception)]
 
 def testf(x):
     return 2.0*x
 
 globala = 99
 
 def testg(x):
     return  globala*x
 
 class IEngineCoreTestCase(object):
     """Test an IEngineCore implementer."""
 
     def createShell(self):
         return Interpreter()
 
     def catchQueueCleared(self, f):
         try:
             f.raiseException()
         except error.QueueCleared:
             pass
         
     def testIEngineCoreInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(es.IEngineCore.providedBy(self.engine))
         
     def testIEngineCoreInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngineCore."""
         for m in list(es.IEngineCore):
             self.assert_(hasattr(self.engine, m))
     
     def testIEngineCoreDeferreds(self):
         d = self.engine.execute('a=5')
         d.addCallback(lambda _: self.engine.pull('a'))
         d.addCallback(lambda _: self.engine.get_result())
         d.addCallback(lambda _: self.engine.keys())
         d.addCallback(lambda _: self.engine.push(dict(a=10)))
         return d
     
     def runTestExecute(self, cmd):
         self.shell = Interpreter()
         actual = self.shell.execute(cmd)
         def compare(computed):
             actual['id'] = computed['id']
             self.assertEquals(actual, computed)
         d = self.engine.execute(cmd)
         d.addCallback(compare)
         return d
     
     @parametric
     def testExecute(cls):
         return [(cls.runTestExecute, cmd) for cmd in validCommands]
 
     def runTestExecuteFailures(self, cmd, exc):
         def compare(f):
             self.assertRaises(exc, f.raiseException)
         d = self.engine.execute(cmd)
         d.addErrback(compare)
         return d
     
     @parametric
-    def testExecuteFailures(cls):
-        return [(cls.runTestExecuteFailures, cmd, exc) for cmd, exc in invalidCommands]
+    def testExecuteFailuresEngineService(cls):
+        return [(cls.runTestExecuteFailures, cmd, exc)
+                for cmd, exc in invalidCommands]
     
     def runTestPushPull(self, o):
         d = self.engine.push(dict(a=o))
         d.addCallback(lambda r: self.engine.pull('a'))
         d.addCallback(lambda r: self.assertEquals(o,r))
         return d
     
     @parametric
     def testPushPull(cls):
         objs = [10,"hi there",1.2342354,{"p":(1,2)},None]
         return [(cls.runTestPushPull, o) for o in objs]
         
     def testPullNameError(self):
         d = self.engine.push(dict(a=5))
         d.addCallback(lambda _:self.engine.reset())
         d.addCallback(lambda _: self.engine.pull("a"))
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         return d        
     
     def testPushPullFailures(self):
         d = self.engine.pull('a')
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         d.addCallback(lambda _: self.engine.execute('l = lambda x: x'))
         d.addCallback(lambda _: self.engine.pull('l'))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         d.addCallback(lambda _: self.engine.push(dict(l=lambda x: x)))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         return d
         
     def testPushPullArray(self):
         try:
             import numpy
         except:
             return
         a = numpy.random.random(1000)
         d = self.engine.push(dict(a=a))
         d.addCallback(lambda _: self.engine.pull('a'))
         d.addCallback(lambda b: b==a)
         d.addCallback(lambda c: c.all())
         return self.assertDeferredEquals(d, True)
         
     def testPushFunction(self):
                     
         d = self.engine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.engine.execute('result = f(10)'))
         d.addCallback(lambda _: self.engine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, testf(10)))
         return d
 
     def testPullFunction(self):
         d = self.engine.push_function(dict(f=testf, g=testg))
         d.addCallback(lambda _: self.engine.pull_function(('f','g')))
         d.addCallback(lambda r: self.assertEquals(r[0](10), testf(10)))
         return d
         
     def testPushFunctionGlobal(self):
         """Make sure that pushed functions pick up the user's namespace for globals."""
         d = self.engine.push(dict(globala=globala))
         d.addCallback(lambda _: self.engine.push_function(dict(g=testg)))
         d.addCallback(lambda _: self.engine.execute('result = g(10)'))
         d.addCallback(lambda _: self.engine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, testg(10)))
         return d
         
     def testGetResultFailure(self):
         d = self.engine.get_result(None)
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         d.addCallback(lambda _: self.engine.get_result(10))
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         return d
 
     def runTestGetResult(self, cmd):
         self.shell = Interpreter()
         actual = self.shell.execute(cmd)
         def compare(computed):
             actual['id'] = computed['id']
             self.assertEquals(actual, computed)
         d = self.engine.execute(cmd)
         d.addCallback(lambda r: self.engine.get_result(r['number']))
         d.addCallback(compare)
         return d
     
     @parametric
     def testGetResult(cls):
         return [(cls.runTestGetResult, cmd) for cmd in validCommands]
     
     def testGetResultDefault(self):
         cmd = 'a=5'
         shell = self.createShell()
         shellResult = shell.execute(cmd)
         def popit(dikt, key):
             dikt.pop(key)
             return dikt
         d = self.engine.execute(cmd)
         d.addCallback(lambda _: self.engine.get_result())
         d.addCallback(lambda r: self.assertEquals(shellResult, popit(r,'id')))
         return d
 
     def testKeys(self):
         d = self.engine.keys()
         d.addCallback(lambda s: isinstance(s, list))
         d.addCallback(lambda r: self.assertEquals(r, True))
         return d
             
 Parametric(IEngineCoreTestCase)
             
 class IEngineSerializedTestCase(object):
     """Test an IEngineCore implementer."""
         
     def testIEngineSerializedInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(es.IEngineSerialized.providedBy(self.engine))
         
     def testIEngineSerializedInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireCore."""
         for m in list(es.IEngineSerialized):
             self.assert_(hasattr(self.engine, m))
        
     def testIEngineSerializedDeferreds(self):
         dList = []
         d = self.engine.push_serialized(dict(key=newserialized.serialize(12345)))
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         d = self.engine.pull_serialized('key')
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         D = defer.DeferredList(dList)
         return D
                 
     def testPushPullSerialized(self):
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d = defer.succeed(None)
         for o in objs:
             self.engine.push_serialized(dict(key=newserialized.serialize(o)))
             value = self.engine.pull_serialized('key')
             value.addCallback(lambda serial: newserialized.IUnSerialized(serial).getObject())
             d = self.assertDeferredEquals(value,o,d)
         return d
 
     def testPullSerializedFailures(self):
         d = self.engine.pull_serialized('a')
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         d.addCallback(lambda _: self.engine.execute('l = lambda x: x'))
         d.addCallback(lambda _: self.engine.pull_serialized('l'))
         d.addErrback(lambda f: self.assertRaises(pickle.PicklingError, f.raiseException))
         return d
 
 Parametric(IEngineSerializedTestCase)
 
 class IEngineQueuedTestCase(object):
     """Test an IEngineQueued implementer."""
         
     def testIEngineQueuedInterface(self):
         """Does self.engine claim to implement IEngineQueued?"""
         self.assert_(es.IEngineQueued.providedBy(self.engine))
         
     def testIEngineQueuedInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireQueued."""
         for m in list(es.IEngineQueued):
             self.assert_(hasattr(self.engine, m))
             
     def testIEngineQueuedDeferreds(self): 
         dList = []
         d = self.engine.clear_queue()
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         d = self.engine.queue_status()
         self.assert_(isinstance(d, defer.Deferred))
         dList.append(d)
         D = defer.DeferredList(dList)
         return D
             
     def testClearQueue(self):
         result = self.engine.clear_queue()
         d1 = self.assertDeferredEquals(result, None)
         d1.addCallback(lambda _: self.engine.queue_status())
         d2 = self.assertDeferredEquals(d1, {'queue':[], 'pending':'None'})
         return d2
         
     def testQueueStatus(self):
         result = self.engine.queue_status()
         result.addCallback(lambda r: 'queue' in r and 'pending' in r)
         d = self.assertDeferredEquals(result, True)
         return d
 
 Parametric(IEngineQueuedTestCase)
 
 class IEnginePropertiesTestCase(object):
     """Test an IEngineProperties implementor."""
     
     def testIEnginePropertiesInterface(self):
         """Does self.engine claim to implement IEngineProperties?"""
         self.assert_(es.IEngineProperties.providedBy(self.engine))
     
     def testIEnginePropertiesInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngireProperties."""
         for m in list(es.IEngineProperties):
             self.assert_(hasattr(self.engine, m))
     
     def testGetSetProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.get_properties())
         d = self.assertDeferredEquals(d, dikt)
         d.addCallback(lambda r: self.engine.get_properties(('c',)))
         d = self.assertDeferredEquals(d, {'c': dikt['c']})
         d.addCallback(lambda r: self.engine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.engine.get_properties(('c', 'd')))
         d = self.assertDeferredEquals(d, dict(c=False, d=None))
         return d
     
     def testClearProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.clear_properties())
         d.addCallback(lambda r: self.engine.get_properties())
         d = self.assertDeferredEquals(d, {})
         return d
     
     def testDelHasProperties(self):
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d = self.engine.set_properties(dikt)
         d.addCallback(lambda r: self.engine.del_properties(('b','e')))
         d.addCallback(lambda r: self.engine.has_properties(('a','b','c','d','e')))
         d = self.assertDeferredEquals(d, [True, False, True, True, False])
         return d
     
     def testStrictDict(self):
         s = """from IPython.kernel.engineservice import get_engine
 p = get_engine(%s).properties"""%self.engine.id
         d = self.engine.execute(s)
         d.addCallback(lambda r: self.engine.execute("p['a'] = lambda _:None"))
         d = self.assertDeferredRaises(d, error.InvalidProperty)
         d.addCallback(lambda r: self.engine.execute("p['a'] = range(5)"))
         d.addCallback(lambda r: self.engine.execute("p['a'].append(5)"))
         d.addCallback(lambda r: self.engine.get_properties('a'))
         d = self.assertDeferredEquals(d, dict(a=range(5)))
         return d
         
 Parametric(IEnginePropertiesTestCase)

IPython/kernel/tests/multienginetest.py

@@ -1,828 +1,828 b''
 # encoding: utf-8
 
 """"""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 from twisted.internet import defer
 
 from IPython.kernel import engineservice as es
 from IPython.kernel import multiengine as me
 from IPython.kernel import newserialized
 from IPython.kernel.error import NotDefined
 from IPython.testing import util
 from IPython.testing.parametric import parametric, Parametric
 from IPython.kernel import newserialized
 from IPython.kernel.util import printer
 from IPython.kernel.error import (InvalidEngineID,
     NoEnginesRegistered,
     CompositeError,
     InvalidDeferredID)
 from IPython.kernel.tests.engineservicetest import validCommands, invalidCommands
 from IPython.kernel.core.interpreter import Interpreter
 
 
 #-------------------------------------------------------------------------------
 # Base classes and utilities
 #-------------------------------------------------------------------------------
 
 class IMultiEngineBaseTestCase(object):
     """Basic utilities for working with multiengine tests.
 
     Some subclass should define:
 
     * self.multiengine
     * self.engines to keep track of engines for clean up"""
 
     def createShell(self):
         return Interpreter()
 
     def addEngine(self, n=1):
         for i in range(n):
             e = es.EngineService()
             e.startService()
             regDict = self.controller.register_engine(es.QueuedEngine(e), None)
             e.id = regDict['id']
             self.engines.append(e)
 
 
 def testf(x):
     return 2.0*x
 
 
 globala = 99
 
 
 def testg(x):
     return  globala*x
 
 
 def isdid(did):
     if not isinstance(did, str):
         return False
     if not len(did)==40:
         return False
     return True
 
 
 def _raise_it(f):
     try:
         f.raiseException()
     except CompositeError, e:
         e.raise_exception()
 
 #-------------------------------------------------------------------------------
 # IMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class IMultiEngineTestCase(IMultiEngineBaseTestCase):
     """A test for any object that implements IEngineMultiplexer.
 
     self.multiengine must be defined and implement IEngineMultiplexer.
     """
 
     def testIMultiEngineInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(me.IEngineMultiplexer.providedBy(self.multiengine))
         self.assert_(me.IMultiEngine.providedBy(self.multiengine))
 
     def testIEngineMultiplexerInterfaceMethods(self):
         """Does self.engine have the methods and attributes in IEngineCore."""
         for m in list(me.IEngineMultiplexer):
             self.assert_(hasattr(self.multiengine, m))
 
     def testIEngineMultiplexerDeferreds(self):
         self.addEngine(1)
         d= self.multiengine.execute('a=5', targets=0)
         d.addCallback(lambda _: self.multiengine.push(dict(a=5),targets=0))
         d.addCallback(lambda _: self.multiengine.push(dict(a=5, b='asdf', c=[1,2,3]),targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b','c'),targets=0))
         d.addCallback(lambda _: self.multiengine.get_result(targets=0))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)),targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a',targets=0))
         d.addCallback(lambda _: self.multiengine.clear_queue(targets=0))
         d.addCallback(lambda _: self.multiengine.queue_status(targets=0))
         return d
 
     def testInvalidEngineID(self):
          self.addEngine(1)
          badID = 100
          d = self.multiengine.execute('a=5', targets=badID)
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.push(dict(a=5), targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.pull('a', targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.reset(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.keys(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)), targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          d.addCallback(lambda _: self.multiengine.queue_status(targets=badID))
          d.addErrback(lambda f: self.assertRaises(InvalidEngineID, f.raiseException))
          return d
 
     def testNoEnginesRegistered(self):
         badID = 'all'
         d= self.multiengine.execute('a=5', targets=badID)
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.push(dict(a=5), targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.pull('a', targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_result(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.reset(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.keys(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(a=newserialized.serialize(10)), targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         d.addCallback(lambda _: self.multiengine.queue_status(targets=badID))
         d.addErrback(lambda f: self.assertRaises(NoEnginesRegistered, f.raiseException))
         return d
 
     def runExecuteAll(self, d, cmd, shell):
         actual = shell.execute(cmd)
         d.addCallback(lambda _: self.multiengine.execute(cmd))
         def compare(result):
             for r in result:
                 actual['id'] = r['id']
                 self.assertEquals(r, actual)
         d.addCallback(compare)
 
     def testExecuteAll(self):
         self.addEngine(4)
         d= defer.Deferred()
         shell = Interpreter()
         for cmd in validCommands:
                 self.runExecuteAll(d, cmd, shell)
         d.callback(None)
         return d
 
     # The following two methods show how to do parametrized
     # tests.  This is really slick!  Same is used above.
     def runExecuteFailures(self, cmd, exc):
         self.addEngine(4)
         d= self.multiengine.execute(cmd)
         d.addErrback(lambda f: self.assertRaises(exc, _raise_it, f))
         return d
 
     @parametric
-    def testExecuteFailures(cls):
+    def testExecuteFailuresMultiEng(cls):
         return [(cls.runExecuteFailures,cmd,exc) for
                 cmd,exc in invalidCommands]
 
     def testPushPull(self):
         self.addEngine(1)
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= self.multiengine.push(dict(key=objs[0]), targets=0)
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[0]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[1]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[1]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[2]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[2]]))
         d.addCallback(lambda _: self.multiengine.push(dict(key=objs[3]), targets=0))
         d.addCallback(lambda _: self.multiengine.pull('key', targets=0))
         d.addCallback(lambda r: self.assertEquals(r, [objs[3]]))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10,b=20)))
         d.addCallback(lambda _: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, [[10,20]]))
         return d
 
     def testPushPullAll(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10))
         d.addCallback(lambda _: self.multiengine.pull('a'))
         d.addCallback(lambda r: self.assert_(r==[10,10,10,10]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20)))
         d.addCallback(lambda _: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assert_(r==4*[[10,20]]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20), targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'), targets=0))
         d.addCallback(lambda r: self.assert_(r==[[10,20]]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=None, b=None), targets=0))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'), targets=0))
         d.addCallback(lambda r: self.assert_(r==[[None,None]]))
         return d
 
     def testPushPullSerialized(self):
         self.addEngine(1)
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[0])), targets=0)
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[0]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[1])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[1]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[2])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[2]))
         d.addCallback(lambda _: self.multiengine.push_serialized(dict(key=newserialized.serialize(objs[3])), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('key', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, objs[3]))
         d.addCallback(lambda _: self.multiengine.push(dict(a=10,b=range(5)), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized(('a','b'), targets=0))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, [10, range(5)]))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
         objs = [10,"hi there",1.2342354,{"p":(1,2)}]
         d= defer.succeed(None)
         for o in objs:
             self.multiengine.push_serialized(0, key=newserialized.serialize(o))
             value = self.multiengine.pull_serialized(0, 'key')
             value.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
             d = self.assertDeferredEquals(value,o,d)
         return d
 
     def runGetResultAll(self, d, cmd, shell):
         actual = shell.execute(cmd)
         d.addCallback(lambda _: self.multiengine.execute(cmd))
         d.addCallback(lambda _: self.multiengine.get_result())
         def compare(result):
             for r in result:
                 actual['id'] = r['id']
                 self.assertEquals(r, actual)
         d.addCallback(compare)
 
     def testGetResultAll(self):
         self.addEngine(4)
         d= defer.Deferred()
         shell = Interpreter()
         for cmd in validCommands:
                 self.runGetResultAll(d, cmd, shell)
         d.callback(None)
         return d
 
     def testGetResultDefault(self):
         self.addEngine(1)
         target = 0
         cmd = 'a=5'
         shell = self.createShell()
         shellResult = shell.execute(cmd)
         def popit(dikt, key):
             dikt.pop(key)
             return dikt
         d= self.multiengine.execute(cmd, targets=target)
         d.addCallback(lambda _: self.multiengine.get_result(targets=target))
         d.addCallback(lambda r: self.assertEquals(shellResult, popit(r[0],'id')))
         return d
 
     def testGetResultFailure(self):
         self.addEngine(1)
         d= self.multiengine.get_result(None, targets=0)
         d.addErrback(lambda f: self.assertRaises(IndexError, _raise_it, f))
         d.addCallback(lambda _: self.multiengine.get_result(10, targets=0))
         d.addErrback(lambda f: self.assertRaises(IndexError, _raise_it, f))
         return d
 
     def testPushFunction(self):
         self.addEngine(1)
         d= self.multiengine.push_function(dict(f=testf), targets=0)
         d.addCallback(lambda _: self.multiengine.execute('result = f(10)', targets=0))
         d.addCallback(lambda _: self.multiengine.pull('result', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testf(10)))
         d.addCallback(lambda _: self.multiengine.push(dict(globala=globala), targets=0))
         d.addCallback(lambda _: self.multiengine.push_function(dict(g=testg), targets=0))
         d.addCallback(lambda _: self.multiengine.execute('result = g(10)', targets=0))
         d.addCallback(lambda _: self.multiengine.pull('result', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testg(10)))
         return d
 
     def testPullFunction(self):
         self.addEngine(1)
         d= self.multiengine.push(dict(a=globala), targets=0)
         d.addCallback(lambda _: self.multiengine.push_function(dict(f=testf), targets=0))
         d.addCallback(lambda _: self.multiengine.pull_function('f', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0](10), testf(10)))
         d.addCallback(lambda _: self.multiengine.execute("def g(x): return x*x", targets=0))
         d.addCallback(lambda _: self.multiengine.pull_function(('f','g'),targets=0))
         d.addCallback(lambda r: self.assertEquals((r[0][0](10),r[0][1](10)), (testf(10), 100)))
         return d
 
     def testPushFunctionAll(self):
         self.addEngine(4)
         d= self.multiengine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.multiengine.execute('result = f(10)'))
         d.addCallback(lambda _: self.multiengine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, 4*[testf(10)]))
         d.addCallback(lambda _: self.multiengine.push(dict(globala=globala)))
         d.addCallback(lambda _: self.multiengine.push_function(dict(testg=testg)))
         d.addCallback(lambda _: self.multiengine.execute('result = testg(10)'))
         d.addCallback(lambda _: self.multiengine.pull('result'))
         d.addCallback(lambda r: self.assertEquals(r, 4*[testg(10)]))
         return d
 
     def testPullFunctionAll(self):
         self.addEngine(4)
         d= self.multiengine.push_function(dict(f=testf))
         d.addCallback(lambda _: self.multiengine.pull_function('f'))
         d.addCallback(lambda r: self.assertEquals([func(10) for func in r], 4*[testf(10)]))
         return d
 
     def testGetIDs(self):
         self.addEngine(1)
         d= self.multiengine.get_ids()
         d.addCallback(lambda r: self.assertEquals(r, [0]))
         d.addCallback(lambda _: self.addEngine(3))
         d.addCallback(lambda _: self.multiengine.get_ids())
         d.addCallback(lambda r: self.assertEquals(r, [0,1,2,3]))
         return d
 
     def testClearQueue(self):
         self.addEngine(4)
         d= self.multiengine.clear_queue()
         d.addCallback(lambda r: self.assertEquals(r,4*[None]))
         return d
 
     def testQueueStatus(self):
         self.addEngine(4)
         d= self.multiengine.queue_status(targets=0)
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         return d
 
     def testGetSetProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',)))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
         return d
 
     def testClearProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.clear_properties())
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
         return d
 
     def testDelHasProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e')))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
         return d
 
 Parametric(IMultiEngineTestCase)
 
 #-------------------------------------------------------------------------------
 # ISynchronousMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class ISynchronousMultiEngineTestCase(IMultiEngineBaseTestCase):
 
     def testISynchronousMultiEngineInterface(self):
         """Does self.engine claim to implement IEngineCore?"""
         self.assert_(me.ISynchronousEngineMultiplexer.providedBy(self.multiengine))
         self.assert_(me.ISynchronousMultiEngine.providedBy(self.multiengine))
 
     def testExecute(self):
         self.addEngine(4)
         execute = self.multiengine.execute
         d= execute('a=5', targets=0, block=True)
         d.addCallback(lambda r: self.assert_(len(r)==1))
         d.addCallback(lambda _: execute('b=10'))
         d.addCallback(lambda r: self.assert_(len(r)==4))
         d.addCallback(lambda _: execute('c=30', block=False))
         d.addCallback(lambda did: self.assert_(isdid(did)))
         d.addCallback(lambda _: execute('d=[0,1,2]', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(len(r)==4))
         return d
 
     def testPushPull(self):
         data = dict(a=10, b=1.05, c=range(10), d={'e':(1,2),'f':'hi'})
         self.addEngine(4)
         push = self.multiengine.push
         pull = self.multiengine.pull
         d= push({'data':data}, targets=0)
         d.addCallback(lambda r: pull('data', targets=0))
         d.addCallback(lambda r: self.assertEqual(r,[data]))
         d.addCallback(lambda _: push({'data':data}))
         d.addCallback(lambda r: pull('data'))
         d.addCallback(lambda r: self.assertEqual(r,4*[data]))
         d.addCallback(lambda _: push({'data':data}, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: pull('data', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEqual(r,4*[data]))
         d.addCallback(lambda _: push(dict(a=10,b=20)))
         d.addCallback(lambda _: pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,20]]))
         return d
 
     def testPushPullFunction(self):
         self.addEngine(4)
         pushf = self.multiengine.push_function
         pullf = self.multiengine.pull_function
         push = self.multiengine.push
         pull = self.multiengine.pull
         execute = self.multiengine.execute
         d= pushf({'testf':testf}, targets=0)
         d.addCallback(lambda r: pullf('testf', targets=0))
         d.addCallback(lambda r: self.assertEqual(r[0](1.0), testf(1.0)))
         d.addCallback(lambda _: execute('r = testf(10)', targets=0))
         d.addCallback(lambda _: pull('r', targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], testf(10)))
         d.addCallback(lambda _: pushf({'testf':testf}, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: pullf('testf', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEqual(r[0](1.0), testf(1.0)))
         d.addCallback(lambda _: execute("def g(x): return x*x", targets=0))
         d.addCallback(lambda _: pullf(('testf','g'),targets=0))
         d.addCallback(lambda r: self.assertEquals((r[0][0](10),r[0][1](10)), (testf(10), 100)))
         return d
 
     def testGetResult(self):
         shell = Interpreter()
         result1 = shell.execute('a=10')
         result1['id'] = 0
         result2 = shell.execute('b=20')
         result2['id'] = 0
         execute= self.multiengine.execute
         get_result = self.multiengine.get_result
         self.addEngine(1)
         d= execute('a=10')
         d.addCallback(lambda _: get_result())
         d.addCallback(lambda r: self.assertEquals(r[0], result1))
         d.addCallback(lambda _: execute('b=20'))
         d.addCallback(lambda _: get_result(1))
         d.addCallback(lambda r: self.assertEquals(r[0], result1))
         d.addCallback(lambda _: get_result(2, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r[0], result2))
         return d
 
     def testResetAndKeys(self):
         self.addEngine(1)
 
         #Blocking mode
         d= self.multiengine.push(dict(a=10, b=20, c=range(10)), targets=0)
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         def keys_found(keys):
             self.assert_('a' in keys[0])
             self.assert_('b' in keys[0])
             self.assert_('b' in keys[0])
         d.addCallback(keys_found)
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0))
         def keys_not_found(keys):
             self.assert_('a' not in keys[0])
             self.assert_('b' not in keys[0])
             self.assert_('b' not in keys[0])
         d.addCallback(keys_not_found)
 
         #Non-blocking mode
         d.addCallback(lambda _: self.multiengine.push(dict(a=10, b=20, c=range(10)), targets=0))
         d.addCallback(lambda _: self.multiengine.keys(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         def keys_found(keys):
             self.assert_('a' in keys[0])
             self.assert_('b' in keys[0])
             self.assert_('b' in keys[0])
         d.addCallback(keys_found)
         d.addCallback(lambda _: self.multiengine.reset(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.keys(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         def keys_not_found(keys):
             self.assert_('a' not in keys[0])
             self.assert_('b' not in keys[0])
             self.assert_('b' not in keys[0])
         d.addCallback(keys_not_found)
 
         return d
 
     def testPushPullSerialized(self):
         self.addEngine(1)
         dikt = dict(a=10,b='hi there',c=1.2345,d={'p':(1,2)})
         sdikt = {}
         for k,v in dikt.iteritems():
             sdikt[k] = newserialized.serialize(v)
         d= self.multiengine.push_serialized(dict(a=sdikt['a']), targets=0)
         d.addCallback(lambda _: self.multiengine.pull('a',targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], dikt['a']))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, dikt['a']))
         d.addCallback(lambda _: self.multiengine.push_serialized(sdikt, targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized(sdikt.keys(), targets=0))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, dikt.values()))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
 
         #Non-blocking mode
         d.addCallback(lambda r: self.multiengine.push_serialized(dict(a=sdikt['a']), targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.pull('a',targets=0))
         d.addCallback(lambda r: self.assertEquals(r[0], dikt['a']))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda serial: newserialized.IUnSerialized(serial[0]).getObject())
         d.addCallback(lambda r: self.assertEquals(r, dikt['a']))
         d.addCallback(lambda _: self.multiengine.push_serialized(sdikt, targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.multiengine.pull_serialized(sdikt.keys(), targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda serial: [newserialized.IUnSerialized(s).getObject() for s in serial[0]])
         d.addCallback(lambda r: self.assertEquals(r, dikt.values()))
         d.addCallback(lambda _: self.multiengine.reset(targets=0))
         d.addCallback(lambda _: self.multiengine.pull_serialized('a', targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
     def testClearQueue(self):
         self.addEngine(4)
         d= self.multiengine.clear_queue()
         d.addCallback(lambda r: self.multiengine.clear_queue(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r,4*[None]))
         return d
 
     def testQueueStatus(self):
         self.addEngine(4)
         d= self.multiengine.queue_status(targets=0)
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         d.addCallback(lambda r: self.multiengine.queue_status(targets=0, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(isinstance(r[0],tuple)))
         return d
 
     def testGetIDs(self):
         self.addEngine(1)
         d= self.multiengine.get_ids()
         d.addCallback(lambda r: self.assertEquals(r, [0]))
         d.addCallback(lambda _: self.addEngine(3))
         d.addCallback(lambda _: self.multiengine.get_ids())
         d.addCallback(lambda r: self.assertEquals(r, [0,1,2,3]))
         return d
 
     def testGetSetProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',)))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False)))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dikt]))
         d.addCallback(lambda r: self.multiengine.get_properties(('c',), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{'c': dikt['c']}]))
         d.addCallback(lambda r: self.multiengine.set_properties(dict(c=False), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(('c', 'd'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[dict(c=False, d=None)]))
         return d
 
     def testClearProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.clear_properties())
         d.addCallback(lambda r: self.multiengine.get_properties())
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.clear_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.get_properties(block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[{}]))
         return d
 
     def testDelHasProperties(self):
         self.addEngine(4)
         dikt = dict(a=5, b='asdf', c=True, d=None, e=range(5))
         d= self.multiengine.set_properties(dikt)
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e')))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
 
         #Non-blocking
         d.addCallback(lambda r: self.multiengine.set_properties(dikt, block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.del_properties(('b','e'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.has_properties(('a','b','c','d','e'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[True, False, True, True, False]]))
         return d
 
     def test_clear_pending_deferreds(self):
         self.addEngine(4)
         did_list = []
         d= self.multiengine.execute('a=10',block=False)
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.push(dict(b=10),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.pull(('a','b'),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.clear_pending_deferreds())
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[0],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[1],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[2],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         return d
 
 #-------------------------------------------------------------------------------
 # Coordinator test cases
 #-------------------------------------------------------------------------------
 
 class IMultiEngineCoordinatorTestCase(object):
 
     def testScatterGather(self):
         self.addEngine(4)
         d= self.multiengine.scatter('a', range(16))
         d.addCallback(lambda r: self.multiengine.gather('a'))
         d.addCallback(lambda r: self.assertEquals(r, range(16)))
         d.addCallback(lambda _: self.multiengine.gather('asdf'))
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
 
     def testScatterGatherNumpy(self):
         try:
             import numpy
             from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         except:
             return
         else:
             self.addEngine(4)
             a = numpy.arange(16)
             d = self.multiengine.scatter('a', a)
             d.addCallback(lambda r: self.multiengine.gather('a'))
             d.addCallback(lambda r: assert_array_equal(r, a))
             return d
 
     def testMap(self):
         self.addEngine(4)
         def f(x):
             return x**2
         data = range(16)
         d= self.multiengine.map(f, data)
         d.addCallback(lambda r: self.assertEquals(r,[f(x) for x in data]))
         return d
 
 
 class ISynchronousMultiEngineCoordinatorTestCase(IMultiEngineCoordinatorTestCase):
 
     def testScatterGatherNonblocking(self):
         self.addEngine(4)
         d= self.multiengine.scatter('a', range(16), block=False)
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.gather('a', block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r, range(16)))
         return d
     
     def testScatterGatherNumpyNonblocking(self):
         try:
             import numpy
             from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         except:
             return
         else:
             self.addEngine(4)
             a = numpy.arange(16)
             d = self.multiengine.scatter('a', a, block=False)
             d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
             d.addCallback(lambda r: self.multiengine.gather('a', block=False))
             d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
             d.addCallback(lambda r: assert_array_equal(r, a))
             return d
     
     def test_clear_pending_deferreds(self):
         self.addEngine(4)
         did_list = []
         d= self.multiengine.scatter('a',range(16),block=False)
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.gather('a',block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.map(lambda x: x, range(16),block=False))
         d.addCallback(lambda did: did_list.append(did))
         d.addCallback(lambda _: self.multiengine.clear_pending_deferreds())
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[0],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[1],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         d.addCallback(lambda _: self.multiengine.get_pending_deferred(did_list[2],True))
         d.addErrback(lambda f: self.assertRaises(InvalidDeferredID, f.raiseException))
         return d
 
 #-------------------------------------------------------------------------------
 # Extras test cases
 #-------------------------------------------------------------------------------
 
 class IMultiEngineExtrasTestCase(object):
 
     def testZipPull(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10,b=20))
         d.addCallback(lambda r: self.multiengine.zip_pull(('a','b')))
         d.addCallback(lambda r: self.assert_(r, [4*[10],4*[20]]))
         return d
 
     def testRun(self):
         self.addEngine(4)
         import tempfile
         fname = tempfile.mktemp('foo.py')
         f= open(fname, 'w')
         f.write('a = 10\nb=30')
         f.close()
         d= self.multiengine.run(fname)
         d.addCallback(lambda r: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,30]]))
         return d
 
 
 class ISynchronousMultiEngineExtrasTestCase(IMultiEngineExtrasTestCase):
 
     def testZipPullNonblocking(self):
         self.addEngine(4)
         d= self.multiengine.push(dict(a=10,b=20))
         d.addCallback(lambda r: self.multiengine.zip_pull(('a','b'), block=False))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assert_(r, [4*[10],4*[20]]))
         return d
 
     def testRunNonblocking(self):
         self.addEngine(4)
         import tempfile
         fname = tempfile.mktemp('foo.py')
         f= open(fname, 'w')
         f.write('a = 10\nb=30')
         f.close()
         d= self.multiengine.run(fname, block=False)
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.multiengine.pull(('a','b')))
         d.addCallback(lambda r: self.assertEquals(r, 4*[[10,30]]))
         return d
 
 
 #-------------------------------------------------------------------------------
 # IFullSynchronousMultiEngineTestCase
 #-------------------------------------------------------------------------------
 
 class IFullSynchronousMultiEngineTestCase(ISynchronousMultiEngineTestCase,
     ISynchronousMultiEngineCoordinatorTestCase,
     ISynchronousMultiEngineExtrasTestCase):
     pass