support iterators in view.map

Some objects are not sliceable (e.g. xrange). For these, fall back
on itertools.islice.

tests included

reported on IRC by @juliantaylor

IPython/parallel/client/map.py

@@ -1,165 +1,171 b''
 # encoding: utf-8
 
 """Classes used in scattering and gathering sequences.
 
 Scattering consists of partitioning a sequence and sending the various
 pieces to individual nodes in a cluster.
 
 
 Authors:
 
 * Brian Granger
 * MinRK
 
 """
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008-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
 #-------------------------------------------------------------------------------
 
 from __future__ import division
 
 import types
+from itertools import islice
 
 from IPython.utils.data import flatten as utils_flatten
 
 #-------------------------------------------------------------------------------
 # Figure out which array packages are present and their array types
 #-------------------------------------------------------------------------------
 
 arrayModules = []
 try:
     import Numeric
 except ImportError:
     pass
 else:
     arrayModules.append({'module':Numeric, 'type':Numeric.arraytype})
 try:
     import numpy
 except ImportError:
     pass
 else:
     arrayModules.append({'module':numpy, 'type':numpy.ndarray})
 try:
     import numarray
 except ImportError:
     pass
 else:
     arrayModules.append({'module':numarray,
         'type':numarray.numarraycore.NumArray})
 
 class Map:
     """A class for partitioning a sequence using a map."""
             
     def getPartition(self, seq, p, q):
         """Returns the pth partition of q partitions of seq."""
         
         # Test for error conditions here
         if p<0 or p>=q:
           print "No partition exists."
           return
           
         remainder = len(seq)%q
         basesize = len(seq)//q
         hi = []
         lo = []
         for n in range(q):
             if n < remainder:
                 lo.append(n * (basesize + 1))
                 hi.append(lo[-1] + basesize + 1)
             else:
                 lo.append(n*basesize + remainder)
                 hi.append(lo[-1] + basesize)
-
         
-        result = seq[lo[p]:hi[p]]
+        try:
+            result = seq[lo[p]:hi[p]]
+        except TypeError:
+            # some objects (iterators) can't be sliced,
+            # use islice:
+            result = list(islice(seq, lo[p], hi[p]))
+            
         return result
            
     def joinPartitions(self, listOfPartitions):
         return self.concatenate(listOfPartitions)
                     
     def concatenate(self, listOfPartitions):
         testObject = listOfPartitions[0]
         # First see if we have a known array type
         for m in arrayModules:
             #print m
             if isinstance(testObject, m['type']):
                 return m['module'].concatenate(listOfPartitions)
         # Next try for Python sequence types
         if isinstance(testObject, (types.ListType, types.TupleType)):
             return utils_flatten(listOfPartitions)
         # If we have scalars, just return listOfPartitions
         return listOfPartitions
 
 class RoundRobinMap(Map):
     """Partitions a sequence in a roun robin fashion.
     
     This currently does not work!
     """
 
     def getPartition(self, seq, p, q):
         # if not isinstance(seq,(list,tuple)):
         #     raise NotImplementedError("cannot RR partition type %s"%type(seq))
         return seq[p:len(seq):q]
         #result = []
         #for i in range(p,len(seq),q):
         #    result.append(seq[i])
         #return result
 
     def joinPartitions(self, listOfPartitions):
         testObject = listOfPartitions[0]
         # First see if we have a known array type
         for m in arrayModules:
             #print m
             if isinstance(testObject, m['type']):
                 return self.flatten_array(m['type'], listOfPartitions)
         if isinstance(testObject, (types.ListType, types.TupleType)):
             return self.flatten_list(listOfPartitions)
         return listOfPartitions
     
     def flatten_array(self, klass, listOfPartitions):
         test = listOfPartitions[0]
         shape = list(test.shape)
         shape[0] = sum([ p.shape[0] for p in listOfPartitions])
         A = klass(shape)
         N = shape[0]
         q = len(listOfPartitions)
         for p,part in enumerate(listOfPartitions):
             A[p:N:q] = part
         return A
     
     def flatten_list(self, listOfPartitions):
         flat = []
         for i in range(len(listOfPartitions[0])):
             flat.extend([ part[i] for part in listOfPartitions if len(part) > i ])
         return flat
         #lengths = [len(x) for x in listOfPartitions]
         #maxPartitionLength = len(listOfPartitions[0])
         #numberOfPartitions = len(listOfPartitions)
         #concat = self.concatenate(listOfPartitions)
         #totalLength = len(concat)
         #result = []
         #for i in range(maxPartitionLength):
         #    result.append(concat[i:totalLength:maxPartitionLength])
         # return self.concatenate(listOfPartitions)
 
 def mappable(obj):
     """return whether an object is mappable or not."""
     if isinstance(obj, (tuple,list)):
         return True
     for m in arrayModules:
         if isinstance(obj,m['type']):
             return True
     return False
 
 dists = {'b':Map,'r':RoundRobinMap}
 
     
     

IPython/parallel/tests/test_lbview.py

@@ -1,167 +1,178 b''
 # -*- coding: utf-8 -*-
 """test LoadBalancedView 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
 #-------------------------------------------------------------------------------
 
 import sys
 import time
 
 import zmq
 from nose import SkipTest
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 
 from IPython.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, crash, wait, skip_without
 
 def setup():
     add_engines(3)
 
 class TestLoadBalancedView(ClusterTestCase):
 
     def setUp(self):
         ClusterTestCase.setUp(self)
         self.view = self.client.load_balanced_view()
 
     def test_z_crash_task(self):
         """test graceful handling of engine death (balanced)"""
         raise SkipTest("crash tests disabled, due to undesirable crash reports")
         # self.add_engines(1)
         ar = self.view.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_map(self):
         def f(x):
             return x**2
         data = range(16)
         r = self.view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
 
     def test_map_unordered(self):
         def f(x):
             return x**2
         def slow_f(x):
             import time
             time.sleep(0.05*x)
             return x**2
         data = range(16,0,-1)
         reference = map(f, data)
         
         amr = self.view.map_async(slow_f, data, ordered=False)
         self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
         # check individual elements, retrieved as they come
         # list comprehension uses __iter__
         astheycame = [ r for r in amr ]
         # Ensure that at least one result came out of order:
         self.assertNotEquals(astheycame, reference, "should not have preserved order")
         self.assertEquals(sorted(astheycame, reverse=True), reference, "result corrupted")
 
     def test_map_ordered(self):
         def f(x):
             return x**2
         def slow_f(x):
             import time
             time.sleep(0.05*x)
             return x**2
         data = range(16,0,-1)
         reference = map(f, data)
         
         amr = self.view.map_async(slow_f, data)
         self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
         # check individual elements, retrieved as they come
         # list(amr) uses __iter__
         astheycame = list(amr)
         # Ensure that results came in order
         self.assertEquals(astheycame, reference)
         self.assertEquals(amr.result, reference)
+
+    def test_map_iterable(self):
+        """test map on iterables (balanced)"""
+        view = self.view
+        # 101 is prime, so it won't be evenly distributed
+        arr = range(101)
+        # so that it will be an iterator, even in Python 3
+        it = iter(arr)
+        r = view.map_sync(lambda x:x, arr)
+        self.assertEquals(r, list(arr))
+
     
     def test_abort(self):
         view = self.view
         ar = self.client[:].apply_async(time.sleep, .5)
         ar = self.client[:].apply_async(time.sleep, .5)
         time.sleep(0.2)
         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_retries(self):
         add_engines(3)
         view = self.view
         view.timeout = 1 # prevent hang if this doesn't behave
         def fail():
             assert False
         for r in range(len(self.client)-1):
             with view.temp_flags(retries=r):
                 self.assertRaisesRemote(AssertionError, view.apply_sync, fail)
 
         with view.temp_flags(retries=len(self.client), timeout=0.25):
             self.assertRaisesRemote(error.TaskTimeout, view.apply_sync, fail)
 
     def test_invalid_dependency(self):
         view = self.view
         with view.temp_flags(after='12345'):
             self.assertRaisesRemote(error.InvalidDependency, view.apply_sync, lambda : 1)
 
     def test_impossible_dependency(self):
         if len(self.client) < 2:
             add_engines(2)
         view = self.client.load_balanced_view()
         ar1 = view.apply_async(lambda : 1)
         ar1.get()
         e1 = ar1.engine_id
         e2 = e1
         while e2 == e1:
             ar2 = view.apply_async(lambda : 1)
             ar2.get()
             e2 = ar2.engine_id
 
         with view.temp_flags(follow=[ar1, ar2]):
             self.assertRaisesRemote(error.ImpossibleDependency, view.apply_sync, lambda : 1)
 
 
     def test_follow(self):
         ar = self.view.apply_async(lambda : 1)
         ar.get()
         ars = []
         first_id = ar.engine_id
 
         self.view.follow = ar
         for i in range(5):
             ars.append(self.view.apply_async(lambda : 1))
         self.view.wait(ars)
         for ar in ars:
             self.assertEquals(ar.engine_id, first_id)
 
     def test_after(self):
         view = self.view
         ar = view.apply_async(time.sleep, 0.5)
         with view.temp_flags(after=ar):
             ar2 = view.apply_async(lambda : 1)
 
         ar.wait()
         ar2.wait()
         self.assertTrue(ar2.started >= ar.completed, "%s not >= %s"%(ar.started, ar.completed))

IPython/parallel/tests/test_view.py

@@ -1,451 +1,460 b''
 # -*- coding: utf-8 -*-
 """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
 #-------------------------------------------------------------------------------
 
 import sys
 import time
 from tempfile import mktemp
 from StringIO import StringIO
 
 import zmq
 from nose import SkipTest
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult, AsyncMapResult
 from IPython.parallel import DirectView
 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)
 
 class TestView(ClusterTestCase):
     
     def test_z_crash_mux(self):
         """test graceful handling of engine death (direct)"""
         raise SkipTest("crash tests disabled, due to undesirable crash reports")
         # 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=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.assertEquals(d, data)
         self.client[:].push({'data':data})
         d = self.client[:].pull('data', block=True)
         self.assertEquals(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.assertEquals(r, nengines*[data])
         self.client[:].push(dict(a=10,b=20))
         r = self.client[:].pull(('a','b'), block=True)
         self.assertEquals(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.assertEquals(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.assertEquals((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.assertEquals(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.assertEquals(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)
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEquals(ahr.get(), ar.get())
         ar2 = v2.get_result(ar.msg_ids)
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.spin()
         c.close()
     
     def test_run_newline(self):
         """test that run appends newline to files"""
         tmpfile = mktemp()
         with open(tmpfile, 'w') as f:
             f.write("""def g():
                 return 5
                 """)
         v = self.client[-1]
         v.run(tmpfile, block=True)
         self.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(b, 123)
 
 
     def test_scatter_gather(self):
         view = self.client[:]
         seq1 = range(16)
         view.scatter('a', seq1)
         seq2 = view.gather('a', block=True)
         self.assertEquals(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, assert_array_almost_equal
         view = self.client[:]
         a = numpy.arange(64)
         view.scatter('a', a)
         b = view.gather('a', block=True)
         assert_array_equal(b, a)
     
     def test_map(self):
         view = self.client[:]
         def f(x):
             return x**2
         data = range(16)
         r = view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
     
+    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, arr)
+        self.assertEquals(r, list(arr))
+    
     def test_scatterGatherNonblocking(self):
         data = range(16)
         view = self.client[:]
         view.scatter('a', data, block=False)
         ar = view.gather('a', block=False)
         self.assertEquals(ar.get(), data)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy_nonblocking(self):
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_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.assertEquals(len(ar.get()), len(self.client))
         for c in view['c']:
             self.assertEquals(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_temp_flags(self):
         view = self.client[-1]
         view.block=True
         with view.temp_flags(block=False):
             self.assertFalse(view.block)
         self.assertTrue(view.block)
     
     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.assertEquals(view.apply_sync(findall, '\w+', 'hello world'), 'hello world'.split())
     
     # parallel magic tests
     
     def test_magic_px_blocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=True
 
         ip.magic_px('a=5')
         self.assertEquals(v['a'], 5)
         ip.magic_px('a=10')
         self.assertEquals(v['a'], 10)
         sio = StringIO()
         savestdout = sys.stdout
         sys.stdout = sio
         # just 'print a' worst ~99% of the time, but this ensures that
         # the stdout message has arrived when the result is finished:
         ip.magic_px('import sys,time;print a; sys.stdout.flush();time.sleep(0.2)')
         sys.stdout = savestdout
         buf = sio.getvalue()
         self.assertTrue('[stdout:' in buf, buf)
         self.assertTrue(buf.rstrip().endswith('10'))
         self.assertRaisesRemote(ZeroDivisionError, ip.magic_px, '1/0')
 
     def test_magic_px_nonblocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=False
 
         ip.magic_px('a=5')
         self.assertEquals(v['a'], 5)
         ip.magic_px('a=10')
         self.assertEquals(v['a'], 10)
         sio = StringIO()
         savestdout = sys.stdout
         sys.stdout = sio
         ip.magic_px('print a')
         sys.stdout = savestdout
         buf = sio.getvalue()
         self.assertFalse('[stdout:%i]'%v.targets in buf)
         ip.magic_px('1/0')
         ar = v.get_result(-1)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
     
     def test_magic_autopx_blocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=True
 
         sio = StringIO()
         savestdout = sys.stdout
         sys.stdout = sio
         ip.magic_autopx()
         ip.run_cell('\n'.join(('a=5','b=10','c=0')))
         ip.run_cell('print b')
         ip.run_cell("b/c")
         ip.run_code(compile('b*=2', '', 'single'))
         ip.magic_autopx()
         sys.stdout = savestdout
         output = sio.getvalue().strip()
         self.assertTrue(output.startswith('%autopx enabled'))
         self.assertTrue(output.endswith('%autopx disabled'))
         self.assertTrue('RemoteError: ZeroDivisionError' in output)
         ar = v.get_result(-2)
         self.assertEquals(v['a'], 5)
         self.assertEquals(v['b'], 20)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
 
     def test_magic_autopx_nonblocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=False
 
         sio = StringIO()
         savestdout = sys.stdout
         sys.stdout = sio
         ip.magic_autopx()
         ip.run_cell('\n'.join(('a=5','b=10','c=0')))
         ip.run_cell('print b')
         ip.run_cell("b/c")
         ip.run_code(compile('b*=2', '', 'single'))
         ip.magic_autopx()
         sys.stdout = savestdout
         output = sio.getvalue().strip()
         self.assertTrue(output.startswith('%autopx enabled'))
         self.assertTrue(output.endswith('%autopx disabled'))
         self.assertFalse('ZeroDivisionError' in output)
         ar = v.get_result(-2)
         self.assertEquals(v['a'], 5)
         self.assertEquals(v['b'], 20)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
     
     def test_magic_result(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v['a'] = 111
         ra = v['a']
         
         ar = ip.magic_result()
         self.assertEquals(ar.msg_ids, [v.history[-1]])
         self.assertEquals(ar.get(), 111)
         ar = ip.magic_result('-2')
         self.assertEquals(ar.msg_ids, [v.history[-2]])
     
     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.assertEquals(v['a'], u'é')
         
     def test_unicode_apply_result(self):
         """test unicode apply results"""
         v = self.client[-1]
         r = v.apply_sync(lambda : u'é')
         self.assertEquals(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 isinstance(a, unicode), "%r is 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
-        
-
+