IPython/kernel/tests/multienginetest.py

# 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):
        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 testMapNonblocking(self):
        self.addEngine(4)
        def f(x):
            return x**2
        data = range(16)
        d= self.multiengine.map(f, data, block=False)
        d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
        d.addCallback(lambda r: self.assertEquals(r,[f(x) for x in data]))
        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