Merge pull request #5766 from minrk/map-length-0

View.map with empty sequence should return empty list

IPython/parallel/client/remotefunction.py

@@ -1,285 +1,276 b''
-"""Remote Functions and decorators for Views.
+"""Remote Functions and decorators for Views."""
 
-Authors:
-
-* Brian Granger
-* Min RK
-"""
-#-----------------------------------------------------------------------------
-#  Copyright (C) 2010-2011  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
-#-----------------------------------------------------------------------------
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
 
 from __future__ import division
 
 import sys
 import warnings
 
 from IPython.external.decorator import decorator
 from IPython.testing.skipdoctest import skip_doctest
 
 from . import map as Map
 from .asyncresult import AsyncMapResult
 
 #-----------------------------------------------------------------------------
 # Functions and Decorators
 #-----------------------------------------------------------------------------
 
 @skip_doctest
 def remote(view, block=None, **flags):
     """Turn a function into a remote function.
 
     This method can be used for map:
 
     In [1]: @remote(view,block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def remote_function(f):
         return RemoteFunction(view, f, block=block, **flags)
     return remote_function
 
 @skip_doctest
 def parallel(view, dist='b', block=None, ordered=True, **flags):
     """Turn a function into a parallel remote function.
 
     This method can be used for map:
 
     In [1]: @parallel(view, block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def parallel_function(f):
         return ParallelFunction(view, f, dist=dist, block=block, ordered=ordered, **flags)
     return parallel_function
 
 def getname(f):
     """Get the name of an object.
     
     For use in case of callables that are not functions, and
     thus may not have __name__ defined.
     
     Order: f.__name__ >  f.name > str(f)
     """
     try:
         return f.__name__
     except:
         pass
     try:
         return f.name
     except:
         pass
     
     return str(f)
 
 @decorator
 def sync_view_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute.
     
     This is a clone of view.sync_results, but for remote functions
     """
     view = self.view
     if view._in_sync_results:
         return f(self, *args, **kwargs)
     view._in_sync_results = True
     try:
         ret = f(self, *args, **kwargs)
     finally:
         view._in_sync_results = False
         view._sync_results()
     return ret
     
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class RemoteFunction(object):
     """Turn an existing function into a remote function.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
 
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     view = None # the remote connection
     func = None # the wrapped function
     block = None # whether to block
     flags = None # dict of extra kwargs for temp_flags
 
     def __init__(self, view, f, block=None, **flags):
         self.view = view
         self.func = f
         self.block=block
         self.flags=flags
 
     def __call__(self, *args, **kwargs):
         block = self.view.block if self.block is None else self.block
         with self.view.temp_flags(block=block, **self.flags):
             return self.view.apply(self.func, *args, **kwargs)
 
 
 class ParallelFunction(RemoteFunction):
     """Class for mapping a function to sequences.
 
     This will distribute the sequences according the a mapper, and call
     the function on each sub-sequence.  If called via map, then the function
     will be called once on each element, rather that each sub-sequence.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     dist : str [default: 'b']
         The key for which mapObject to use to distribute sequences
         options are:
 
         * 'b' : use contiguous chunks in order
         * 'r' : use round-robin striping
 
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
     chunksize : int or None
         The size of chunk to use when breaking up sequences in a load-balanced manner
     ordered : bool [default: True]
         Whether the result should be kept in order. If False,
         results become available as they arrive, regardless of submission order.
     **flags
         remaining kwargs are passed to View.temp_flags
     """
 
     chunksize = None
     ordered = None
     mapObject = None
     _mapping = False
 
     def __init__(self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
         self.ordered = ordered
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
     
     @sync_view_results
     def __call__(self, *sequences):
         client = self.view.client
         
         lens = []
         maxlen = minlen = -1
         for i, seq in enumerate(sequences):
             try:
                 n = len(seq)
             except Exception:
                 seq = list(seq)
                 if isinstance(sequences, tuple):
                     # can't alter a tuple
                     sequences = list(sequences)
                 sequences[i] = seq
                 n = len(seq)
             if n > maxlen:
                 maxlen = n
             if minlen == -1 or n < minlen:
                 minlen = n
             lens.append(n)
         
+        if maxlen == 0:
+            # nothing to iterate over
+            return []
+        
         # check that the length of sequences match
         if not self._mapping and minlen != maxlen:
             msg = 'all sequences must have equal length, but have %s' % lens
             raise ValueError(msg)
         
         balanced = 'Balanced' in self.view.__class__.__name__
         if balanced:
             if self.chunksize:
                 nparts = maxlen // self.chunksize + int(maxlen % self.chunksize > 0)
             else:
                 nparts = maxlen
             targets = [None]*nparts
         else:
             if self.chunksize:
                 warnings.warn("`chunksize` is ignored unless load balancing", UserWarning)
             # multiplexed:
             targets = self.view.targets
             # 'all' is lazily evaluated at execution time, which is now:
             if targets == 'all':
                 targets = client._build_targets(targets)[1]
             elif isinstance(targets, int):
                 # single-engine view, targets must be iterable
                 targets = [targets]
             nparts = len(targets)
 
         msg_ids = []
         for index, t in enumerate(targets):
             args = []
             for seq in sequences:
                 part = self.mapObject.getPartition(seq, index, nparts, maxlen)
                 args.append(part)
 
             if sum([len(arg) for arg in args]) == 0:
                 continue
 
             if self._mapping:
                 if sys.version_info[0] >= 3:
                     f = lambda f, *sequences: list(map(f, *sequences))
                 else:
                     f = map
                 args = [self.func] + args
             else:
                 f=self.func
 
             view = self.view if balanced else client[t]
             with view.temp_flags(block=False, **self.flags):
                 ar = view.apply(f, *args)
 
             msg_ids.extend(ar.msg_ids)
 
         r = AsyncMapResult(self.view.client, msg_ids, self.mapObject,
                             fname=getname(self.func),
                             ordered=self.ordered
                         )
 
         if self.block:
             try:
                 return r.get()
             except KeyboardInterrupt:
                 return r
         else:
             return r
 
     def map(self, *sequences):
         """call a function on each element of one or more sequence(s) remotely.
         This should behave very much like the builtin map, but return an AsyncMapResult
         if self.block is False.
         
         That means it can take generators (will be cast to lists locally),
         and mismatched sequence lengths will be padded with None.
         """
         # set _mapping as a flag for use inside self.__call__
         self._mapping = True
         try:
             ret = self(*sequences)
         finally:
             self._mapping = False
         return ret
 
 __all__ = ['remote', 'parallel', 'RemoteFunction', 'ParallelFunction']

IPython/parallel/tests/test_view.py

@@ -1,849 +1,842 b''
 # -*- coding: utf-8 -*-
-"""test View objects
+"""test View objects"""
 
-Authors:
-
-* Min RK
-"""
-#-------------------------------------------------------------------------------
-#  Copyright (C) 2011  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
-#-------------------------------------------------------------------------------
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
 
 import base64
 import sys
 import platform
 import time
 from collections import namedtuple
 from tempfile import NamedTemporaryFile
 
 import zmq
 from nose.plugins.attrib import attr
 
 from IPython.testing import decorators as dec
 from IPython.utils.io import capture_output
 from IPython.utils.py3compat import unicode_type
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult, AsyncMapResult
 from IPython.parallel.util import interactive
 
 from IPython.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, crash, wait, skip_without
 
 def setup():
     add_engines(3, total=True)
 
 point = namedtuple("point", "x y")
 
 class TestView(ClusterTestCase):
     
     def setUp(self):
         # On Win XP, wait for resource cleanup, else parallel test group fails
         if platform.system() == "Windows" and platform.win32_ver()[0] == "XP":
             # 1 sec fails. 1.5 sec seems ok. Using 2 sec for margin of safety
             time.sleep(2)
         super(TestView, self).setUp()
 
     @attr('crash')
     def test_z_crash_mux(self):
         """test graceful handling of engine death (direct)"""
         # self.add_engines(1)
         eid = self.client.ids[-1]
         ar = self.client[eid].apply_async(crash)
         self.assertRaisesRemote(error.EngineError, ar.get, 10)
         eid = ar.engine_id
         tic = time.time()
         while eid in self.client.ids and time.time()-tic < 5:
             time.sleep(.01)
             self.client.spin()
         self.assertFalse(eid in self.client.ids, "Engine should have died")
     
     def test_push_pull(self):
         """test pushing and pulling"""
         data = dict(a=10, b=1.05, c=list(range(10)), d={'e':(1,2),'f':'hi'})
         t = self.client.ids[-1]
         v = self.client[t]
         push = v.push
         pull = v.pull
         v.block=True
         nengines = len(self.client)
         push({'data':data})
         d = pull('data')
         self.assertEqual(d, data)
         self.client[:].push({'data':data})
         d = self.client[:].pull('data', block=True)
         self.assertEqual(d, nengines*[data])
         ar = push({'data':data}, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         ar = self.client[:].pull('data', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         self.assertEqual(r, nengines*[data])
         self.client[:].push(dict(a=10,b=20))
         r = self.client[:].pull(('a','b'), block=True)
         self.assertEqual(r, nengines*[[10,20]])
     
     def test_push_pull_function(self):
         "test pushing and pulling functions"
         def testf(x):
             return 2.0*x
         
         t = self.client.ids[-1]
         v = self.client[t]
         v.block=True
         push = v.push
         pull = v.pull
         execute = v.execute
         push({'testf':testf})
         r = pull('testf')
         self.assertEqual(r(1.0), testf(1.0))
         execute('r = testf(10)')
         r = pull('r')
         self.assertEqual(r, testf(10))
         ar = self.client[:].push({'testf':testf}, block=False)
         ar.get()
         ar = self.client[:].pull('testf', block=False)
         rlist = ar.get()
         for r in rlist:
             self.assertEqual(r(1.0), testf(1.0))
         execute("def g(x): return x*x")
         r = pull(('testf','g'))
         self.assertEqual((r[0](10),r[1](10)), (testf(10), 100))
     
     def test_push_function_globals(self):
         """test that pushed functions have access to globals"""
         @interactive
         def geta():
             return a
         # self.add_engines(1)
         v = self.client[-1]
         v.block=True
         v['f'] = geta
         self.assertRaisesRemote(NameError, v.execute, 'b=f()')
         v.execute('a=5')
         v.execute('b=f()')
         self.assertEqual(v['b'], 5)
     
     def test_push_function_defaults(self):
         """test that pushed functions preserve default args"""
         def echo(a=10):
             return a
         v = self.client[-1]
         v.block=True
         v['f'] = echo
         v.execute('b=f()')
         self.assertEqual(v['b'], 10)
 
     def test_get_result(self):
         """test getting results from the Hub."""
         c = pmod.Client(profile='iptest')
         # self.add_engines(1)
         t = c.ids[-1]
         v = c[t]
         v2 = self.client[t]
         ar = v.apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = v2.get_result(ar.msg_ids[0])
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get(), ar.get())
         ar2 = v2.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.spin()
         c.close()
     
     def test_run_newline(self):
         """test that run appends newline to files"""
         with NamedTemporaryFile('w', delete=False) as f:
             f.write("""def g():
                 return 5
                 """)
         v = self.client[-1]
         v.run(f.name, block=True)
         self.assertEqual(v.apply_sync(lambda f: f(), pmod.Reference('g')), 5)
 
     def test_apply_tracked(self):
         """test tracking for apply"""
         # self.add_engines(1)
         t = self.client.ids[-1]
         v = self.client[t]
         v.block=False
         def echo(n=1024*1024, **kwargs):
             with v.temp_flags(**kwargs):
                 return v.apply(lambda x: x, 'x'*n)
         ar = echo(1, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         ar = echo(track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEqual(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         
     def test_push_tracked(self):
         t = self.client.ids[-1]
         ns = dict(x='x'*1024*1024)
         v = self.client[t]
         ar = v.push(ns, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = v.push(ns, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         ar._tracker.wait()
         self.assertEqual(ar.sent, ar._tracker.done)
         self.assertTrue(ar.sent)
         ar.get()
         
     def test_scatter_tracked(self):
         t = self.client.ids
         x='x'*1024*1024
         ar = self.client[t].scatter('x', x, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = self.client[t].scatter('x', x, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEqual(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         ar.get()
     
     def test_remote_reference(self):
         v = self.client[-1]
         v['a'] = 123
         ra = pmod.Reference('a')
         b = v.apply_sync(lambda x: x, ra)
         self.assertEqual(b, 123)
 
 
     def test_scatter_gather(self):
         view = self.client[:]
         seq1 = list(range(16))
         view.scatter('a', seq1)
         seq2 = view.gather('a', block=True)
         self.assertEqual(seq2, seq1)
         self.assertRaisesRemote(NameError, view.gather, 'asdf', block=True)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy(self):
         import numpy
         from numpy.testing.utils import assert_array_equal
         view = self.client[:]
         a = numpy.arange(64)
         view.scatter('a', a, block=True)
         b = view.gather('a', block=True)
         assert_array_equal(b, a)
     
     def test_scatter_gather_lazy(self):
         """scatter/gather with targets='all'"""
         view = self.client.direct_view(targets='all')
         x = list(range(64))
         view.scatter('x', x)
         gathered = view.gather('x', block=True)
         self.assertEqual(gathered, x)
         
 
     @dec.known_failure_py3
     @skip_without('numpy')
     def test_push_numpy_nocopy(self):
         import numpy
         view = self.client[:]
         a = numpy.arange(64)
         view['A'] = a
         @interactive
         def check_writeable(x):
             return x.flags.writeable
         
         for flag in view.apply_sync(check_writeable, pmod.Reference('A')):
             self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
         
         view.push(dict(B=a))
         for flag in view.apply_sync(check_writeable, pmod.Reference('B')):
             self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
     
     @skip_without('numpy')
     def test_apply_numpy(self):
         """view.apply(f, ndarray)"""
         import numpy
         from numpy.testing.utils import assert_array_equal
         
         A = numpy.random.random((100,100))
         view = self.client[-1]
         for dt in [ 'int32', 'uint8', 'float32', 'float64' ]:
             B = A.astype(dt)
             C = view.apply_sync(lambda x:x, B)
             assert_array_equal(B,C)
     
     @skip_without('numpy')
     def test_apply_numpy_object_dtype(self):
         """view.apply(f, ndarray) with dtype=object"""
         import numpy
         from numpy.testing.utils import assert_array_equal
         view = self.client[-1]
         
         A = numpy.array([dict(a=5)])
         B = view.apply_sync(lambda x:x, A)
         assert_array_equal(A,B)
         
         A = numpy.array([(0, dict(b=10))], dtype=[('i', int), ('o', object)])
         B = view.apply_sync(lambda x:x, A)
         assert_array_equal(A,B)
     
     @skip_without('numpy')
     def test_push_pull_recarray(self):
         """push/pull recarrays"""
         import numpy
         from numpy.testing.utils import assert_array_equal
         
         view = self.client[-1]
         
         R = numpy.array([
             (1, 'hi', 0.),
             (2**30, 'there', 2.5),
             (-99999, 'world', -12345.6789),
         ], [('n', int), ('s', '|S10'), ('f', float)])
         
         view['RR'] = R
         R2 = view['RR']
         
         r_dtype, r_shape = view.apply_sync(interactive(lambda : (RR.dtype, RR.shape)))
         self.assertEqual(r_dtype, R.dtype)
         self.assertEqual(r_shape, R.shape)
         self.assertEqual(R2.dtype, R.dtype)
         self.assertEqual(R2.shape, R.shape)
         assert_array_equal(R2, R)
 
     @skip_without('pandas')
     def test_push_pull_timeseries(self):
         """push/pull pandas.TimeSeries"""
         import pandas
         
         ts = pandas.TimeSeries(list(range(10)))
         
         view = self.client[-1]
         
         view.push(dict(ts=ts), block=True)
         rts = view['ts']
         
         self.assertEqual(type(rts), type(ts))
         self.assertTrue((ts == rts).all())
     
     def test_map(self):
         view = self.client[:]
         def f(x):
             return x**2
         data = list(range(16))
         r = view.map_sync(f, data)
         self.assertEqual(r, list(map(f, data)))
     
+    def test_map_empty_sequence(self):
+        view = self.client[:]
+        r = view.map_sync(lambda x: x, [])
+        self.assertEqual(r, [])
+    
     def test_map_iterable(self):
         """test map on iterables (direct)"""
         view = self.client[:]
         # 101 is prime, so it won't be evenly distributed
         arr = range(101)
         # ensure it will be an iterator, even in Python 3
         it = iter(arr)
         r = view.map_sync(lambda x: x, it)
         self.assertEqual(r, list(arr))
 
     @skip_without('numpy')
     def test_map_numpy(self):
         """test map on numpy arrays (direct)"""
         import numpy
         from numpy.testing.utils import assert_array_equal
 
         view = self.client[:]
         # 101 is prime, so it won't be evenly distributed
         arr = numpy.arange(101)
         r = view.map_sync(lambda x: x, arr)
         assert_array_equal(r, arr)
     
     def test_scatter_gather_nonblocking(self):
         data = list(range(16))
         view = self.client[:]
         view.scatter('a', data, block=False)
         ar = view.gather('a', block=False)
         self.assertEqual(ar.get(), data)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy_nonblocking(self):
         import numpy
         from numpy.testing.utils import assert_array_equal
         a = numpy.arange(64)
         view = self.client[:]
         ar = view.scatter('a', a, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         amr = view.gather('a', block=False)
         self.assertTrue(isinstance(amr, AsyncMapResult))
         assert_array_equal(amr.get(), a)
 
     def test_execute(self):
         view = self.client[:]
         # self.client.debug=True
         execute = view.execute
         ar = execute('c=30', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         ar = execute('d=[0,1,2]', block=False)
         self.client.wait(ar, 1)
         self.assertEqual(len(ar.get()), len(self.client))
         for c in view['c']:
             self.assertEqual(c, 30)
     
     def test_abort(self):
         view = self.client[-1]
         ar = view.execute('import time; time.sleep(1)', block=False)
         ar2 = view.apply_async(lambda : 2)
         ar3 = view.apply_async(lambda : 3)
         view.abort(ar2)
         view.abort(ar3.msg_ids)
         self.assertRaises(error.TaskAborted, ar2.get)
         self.assertRaises(error.TaskAborted, ar3.get)
     
     def test_abort_all(self):
         """view.abort() aborts all outstanding tasks"""
         view = self.client[-1]
         ars = [ view.apply_async(time.sleep, 0.25) for i in range(10) ]
         view.abort()
         view.wait(timeout=5)
         for ar in ars[5:]:
             self.assertRaises(error.TaskAborted, ar.get)
     
     def test_temp_flags(self):
         view = self.client[-1]
         view.block=True
         with view.temp_flags(block=False):
             self.assertFalse(view.block)
         self.assertTrue(view.block)
     
     @dec.known_failure_py3
     def test_importer(self):
         view = self.client[-1]
         view.clear(block=True)
         with view.importer:
             import re
         
         @interactive
         def findall(pat, s):
             # this globals() step isn't necessary in real code
             # only to prevent a closure in the test
             re = globals()['re']
             return re.findall(pat, s)
         
         self.assertEqual(view.apply_sync(findall, '\w+', 'hello world'), 'hello world'.split())
     
     def test_unicode_execute(self):
         """test executing unicode strings"""
         v = self.client[-1]
         v.block=True
         if sys.version_info[0] >= 3:
             code="a='é'"
         else:
             code=u"a=u'é'"
         v.execute(code)
         self.assertEqual(v['a'], u'é')
         
     def test_unicode_apply_result(self):
         """test unicode apply results"""
         v = self.client[-1]
         r = v.apply_sync(lambda : u'é')
         self.assertEqual(r, u'é')
     
     def test_unicode_apply_arg(self):
         """test passing unicode arguments to apply"""
         v = self.client[-1]
         
         @interactive
         def check_unicode(a, check):
             assert not isinstance(a, bytes), "%r is bytes, not unicode"%a
             assert isinstance(check, bytes), "%r is not bytes"%check
             assert a.encode('utf8') == check, "%s != %s"%(a,check)
         
         for s in [ u'é', u'ßø®∫',u'asdf' ]:
             try:
                 v.apply_sync(check_unicode, s, s.encode('utf8'))
             except error.RemoteError as e:
                 if e.ename == 'AssertionError':
                     self.fail(e.evalue)
                 else:
                     raise e
     
     def test_map_reference(self):
         """view.map(<Reference>, *seqs) should work"""
         v = self.client[:]
         v.scatter('n', self.client.ids, flatten=True)
         v.execute("f = lambda x,y: x*y")
         rf = pmod.Reference('f')
         nlist = list(range(10))
         mlist = nlist[::-1]
         expected = [ m*n for m,n in zip(mlist, nlist) ]
         result = v.map_sync(rf, mlist, nlist)
         self.assertEqual(result, expected)
 
     def test_apply_reference(self):
         """view.apply(<Reference>, *args) should work"""
         v = self.client[:]
         v.scatter('n', self.client.ids, flatten=True)
         v.execute("f = lambda x: n*x")
         rf = pmod.Reference('f')
         result = v.apply_sync(rf, 5)
         expected = [ 5*id for id in self.client.ids ]
         self.assertEqual(result, expected)
     
     def test_eval_reference(self):
         v = self.client[self.client.ids[0]]
         v['g'] = list(range(5))
         rg = pmod.Reference('g[0]')
         echo = lambda x:x
         self.assertEqual(v.apply_sync(echo, rg), 0)
     
     def test_reference_nameerror(self):
         v = self.client[self.client.ids[0]]
         r = pmod.Reference('elvis_has_left')
         echo = lambda x:x
         self.assertRaisesRemote(NameError, v.apply_sync, echo, r)
 
     def test_single_engine_map(self):
         e0 = self.client[self.client.ids[0]]
         r = list(range(5))
         check = [ -1*i for i in r ]
         result = e0.map_sync(lambda x: -1*x, r)
         self.assertEqual(result, check)
     
     def test_len(self):
         """len(view) makes sense"""
         e0 = self.client[self.client.ids[0]]
         self.assertEqual(len(e0), 1)
         v = self.client[:]
         self.assertEqual(len(v), len(self.client.ids))
         v = self.client.direct_view('all')
         self.assertEqual(len(v), len(self.client.ids))
         v = self.client[:2]
         self.assertEqual(len(v), 2)
         v = self.client[:1]
         self.assertEqual(len(v), 1)
         v = self.client.load_balanced_view()
         self.assertEqual(len(v), len(self.client.ids))
 
     
     # begin execute tests
     
     def test_execute_reply(self):
         e0 = self.client[self.client.ids[0]]
         e0.block = True
         ar = e0.execute("5", silent=False)
         er = ar.get()
         self.assertEqual(str(er), "<ExecuteReply[%i]: 5>" % er.execution_count)
         self.assertEqual(er.pyout['data']['text/plain'], '5')
 
     def test_execute_reply_rich(self):
         e0 = self.client[self.client.ids[0]]
         e0.block = True
         e0.execute("from IPython.display import Image, HTML")
         ar = e0.execute("Image(data=b'garbage', format='png', width=10)", silent=False)
         er = ar.get()
         b64data = base64.encodestring(b'garbage').decode('ascii')
         self.assertEqual(er._repr_png_(), (b64data, dict(width=10)))
         ar = e0.execute("HTML('<b>bold</b>')", silent=False)
         er = ar.get()
         self.assertEqual(er._repr_html_(), "<b>bold</b>")
 
     def test_execute_reply_stdout(self):
         e0 = self.client[self.client.ids[0]]
         e0.block = True
         ar = e0.execute("print (5)", silent=False)
         er = ar.get()
         self.assertEqual(er.stdout.strip(), '5')
         
     def test_execute_pyout(self):
         """execute triggers pyout with silent=False"""
         view = self.client[:]
         ar = view.execute("5", silent=False, block=True)
         
         expected = [{'text/plain' : '5'}] * len(view)
         mimes = [ out['data'] for out in ar.pyout ]
         self.assertEqual(mimes, expected)
     
     def test_execute_silent(self):
         """execute does not trigger pyout with silent=True"""
         view = self.client[:]
         ar = view.execute("5", block=True)
         expected = [None] * len(view)
         self.assertEqual(ar.pyout, expected)
     
     def test_execute_magic(self):
         """execute accepts IPython commands"""
         view = self.client[:]
         view.execute("a = 5")
         ar = view.execute("%whos", block=True)
         # this will raise, if that failed
         ar.get(5)
         for stdout in ar.stdout:
             lines = stdout.splitlines()
             self.assertEqual(lines[0].split(), ['Variable', 'Type', 'Data/Info'])
             found = False
             for line in lines[2:]:
                 split = line.split()
                 if split == ['a', 'int', '5']:
                     found = True
                     break
             self.assertTrue(found, "whos output wrong: %s" % stdout)
     
     def test_execute_displaypub(self):
         """execute tracks display_pub output"""
         view = self.client[:]
         view.execute("from IPython.core.display import *")
         ar = view.execute("[ display(i) for i in range(5) ]", block=True)
         
         expected = [ {u'text/plain' : unicode_type(j)} for j in range(5) ]
         for outputs in ar.outputs:
             mimes = [ out['data'] for out in outputs ]
             self.assertEqual(mimes, expected)
     
     def test_apply_displaypub(self):
         """apply tracks display_pub output"""
         view = self.client[:]
         view.execute("from IPython.core.display import *")
         
         @interactive
         def publish():
             [ display(i) for i in range(5) ]
         
         ar = view.apply_async(publish)
         ar.get(5)
         expected = [ {u'text/plain' : unicode_type(j)} for j in range(5) ]
         for outputs in ar.outputs:
             mimes = [ out['data'] for out in outputs ]
             self.assertEqual(mimes, expected)
     
     def test_execute_raises(self):
         """exceptions in execute requests raise appropriately"""
         view = self.client[-1]
         ar = view.execute("1/0")
         self.assertRaisesRemote(ZeroDivisionError, ar.get, 2)
     
     def test_remoteerror_render_exception(self):
         """RemoteErrors get nice tracebacks"""
         view = self.client[-1]
         ar = view.execute("1/0")
         ip = get_ipython()
         ip.user_ns['ar'] = ar
         with capture_output() as io:
             ip.run_cell("ar.get(2)")
         
         self.assertTrue('ZeroDivisionError' in io.stdout, io.stdout)
     
     def test_compositeerror_render_exception(self):
         """CompositeErrors get nice tracebacks"""
         view = self.client[:]
         ar = view.execute("1/0")
         ip = get_ipython()
         ip.user_ns['ar'] = ar
 
         with capture_output() as io:
             ip.run_cell("ar.get(2)")
         
         count = min(error.CompositeError.tb_limit, len(view))
         
         self.assertEqual(io.stdout.count('ZeroDivisionError'), count * 2, io.stdout)
         self.assertEqual(io.stdout.count('by zero'), count, io.stdout)
         self.assertEqual(io.stdout.count(':execute'), count, io.stdout)
     
     def test_compositeerror_truncate(self):
         """Truncate CompositeErrors with many exceptions"""
         view = self.client[:]
         msg_ids = []
         for i in range(10):
             ar = view.execute("1/0")
             msg_ids.extend(ar.msg_ids)
         
         ar = self.client.get_result(msg_ids)
         try:
             ar.get()
         except error.CompositeError as _e:
             e = _e
         else:
             self.fail("Should have raised CompositeError")
         
         lines = e.render_traceback()
         with capture_output() as io:
             e.print_traceback()
         
         self.assertTrue("more exceptions" in lines[-1])
         count = e.tb_limit
         
         self.assertEqual(io.stdout.count('ZeroDivisionError'), 2 * count, io.stdout)
         self.assertEqual(io.stdout.count('by zero'), count, io.stdout)
         self.assertEqual(io.stdout.count(':execute'), count, io.stdout)
     
     @dec.skipif_not_matplotlib
     def test_magic_pylab(self):
         """%pylab works on engines"""
         view = self.client[-1]
         ar = view.execute("%pylab inline")
         # at least check if this raised:
         reply = ar.get(5)
         # include imports, in case user config
         ar = view.execute("plot(rand(100))", silent=False)
         reply = ar.get(5)
         self.assertEqual(len(reply.outputs), 1)
         output = reply.outputs[0]
         self.assertTrue("data" in output)
         data = output['data']
         self.assertTrue("image/png" in data)
     
     def test_func_default_func(self):
         """interactively defined function as apply func default"""
         def foo():
             return 'foo'
         
         def bar(f=foo):
             return f()
         
         view = self.client[-1]
         ar = view.apply_async(bar)
         r = ar.get(10)
         self.assertEqual(r, 'foo')
     def test_data_pub_single(self):
         view = self.client[-1]
         ar = view.execute('\n'.join([
             'from IPython.kernel.zmq.datapub import publish_data',
             'for i in range(5):',
             '  publish_data(dict(i=i))'
         ]), block=False)
         self.assertTrue(isinstance(ar.data, dict))
         ar.get(5)
         self.assertEqual(ar.data, dict(i=4))
 
     def test_data_pub(self):
         view = self.client[:]
         ar = view.execute('\n'.join([
             'from IPython.kernel.zmq.datapub import publish_data',
             'for i in range(5):',
             '  publish_data(dict(i=i))'
         ]), block=False)
         self.assertTrue(all(isinstance(d, dict) for d in ar.data))
         ar.get(5)
         self.assertEqual(ar.data, [dict(i=4)] * len(ar))
     
     def test_can_list_arg(self):
         """args in lists are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x: x, [rA])
         r = ar.get(5)
         self.assertEqual(r, [128])
 
     def test_can_dict_arg(self):
         """args in dicts are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x: x, dict(foo=rA))
         r = ar.get(5)
         self.assertEqual(r, dict(foo=128))
 
     def test_can_list_kwarg(self):
         """kwargs in lists are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x=5: x, x=[rA])
         r = ar.get(5)
         self.assertEqual(r, [128])
 
     def test_can_dict_kwarg(self):
         """kwargs in dicts are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x=5: x, dict(foo=rA))
         r = ar.get(5)
         self.assertEqual(r, dict(foo=128))
     
     def test_map_ref(self):
         """view.map works with references"""
         view = self.client[:]
         ranks = sorted(self.client.ids)
         view.scatter('rank', ranks, flatten=True)
         rrank = pmod.Reference('rank')
         
         amr = view.map_async(lambda x: x*2, [rrank] * len(view))
         drank = amr.get(5)
         self.assertEqual(drank, [ r*2 for r in ranks ])
         
     def test_nested_getitem_setitem(self):
         """get and set with view['a.b']"""
         view = self.client[-1]
         view.execute('\n'.join([
             'class A(object): pass',
             'a = A()',
             'a.b = 128',
             ]), block=True)
         ra = pmod.Reference('a')
 
         r = view.apply_sync(lambda x: x.b, ra)
         self.assertEqual(r, 128)
         self.assertEqual(view['a.b'], 128)
 
         view['a.b'] = 0
 
         r = view.apply_sync(lambda x: x.b, ra)
         self.assertEqual(r, 0)
         self.assertEqual(view['a.b'], 0)
     
     def test_return_namedtuple(self):
         def namedtuplify(x, y):
             from IPython.parallel.tests.test_view import point
             return point(x, y)
         
         view = self.client[-1]
         p = view.apply_sync(namedtuplify, 1, 2)
         self.assertEqual(p.x, 1)
         self.assertEqual(p.y, 2)
 
     def test_apply_namedtuple(self):
         def echoxy(p):
             return p.y, p.x
         
         view = self.client[-1]
         tup = view.apply_sync(echoxy, point(1, 2))
         self.assertEqual(tup, (2,1))
     
     def test_sync_imports(self):
         view = self.client[-1]
         with capture_output() as io:
             with view.sync_imports():
                 import IPython
         self.assertIn("IPython", io.stdout)
         
         @interactive
         def find_ipython():
             return 'IPython' in globals()
         
         assert view.apply_sync(find_ipython)
 
     def test_sync_imports_quiet(self):
         view = self.client[-1]
         with capture_output() as io:
             with view.sync_imports(quiet=True):
                 import IPython
         self.assertEqual(io.stdout, '')
         
         @interactive
         def find_ipython():
             return 'IPython' in globals()
         
         assert view.apply_sync(find_ipython)