Remove most uses of ParametricTestCase

IPython/parallel/tests/test_magics.py

@@ -1,380 +1,379 b''
 # -*- coding: utf-8 -*-
 """Test Parallel magics
 
 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 re
 import sys
 import time
 
 import zmq
 from nose import SkipTest
 
 from IPython.testing import decorators as dec
-from IPython.testing.ipunittest import ParametricTestCase
 from IPython.utils.io import capture_output
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult
 from IPython.parallel.util import interactive
 
 from IPython.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, generate_output
 
 def setup():
     add_engines(3, total=True)
 
-class TestParallelMagics(ClusterTestCase, ParametricTestCase):
+class TestParallelMagics(ClusterTestCase):
     
     def test_px_blocking(self):
         ip = get_ipython()
         v = self.client[-1:]
         v.activate()
         v.block=True
 
         ip.magic('px a=5')
         self.assertEqual(v['a'], [5])
         ip.magic('px a=10')
         self.assertEqual(v['a'], [10])
         # just 'print a' works ~99% of the time, but this ensures that
         # the stdout message has arrived when the result is finished:
         with capture_output() as io:
             ip.magic(
                 'px import sys,time;print(a);sys.stdout.flush();time.sleep(0.2)'
             )
         self.assertIn('[stdout:', io.stdout)
         self.assertNotIn('\n\n', io.stdout)
         assert io.stdout.rstrip().endswith('10')
         self.assertRaisesRemote(ZeroDivisionError, ip.magic, 'px 1/0')
     
     def _check_generated_stderr(self, stderr, n):
         expected = [
             r'\[stderr:\d+\]',
             '^stderr$',
             '^stderr2$',
         ] * n
         
         self.assertNotIn('\n\n', stderr)
         lines = stderr.splitlines()
         self.assertEqual(len(lines), len(expected), stderr)
         for line,expect in zip(lines, expected):
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
                 assert re.search(ex, line) is not None, "Expected %r in %r" % (ex, line)
         
     def test_cellpx_block_args(self):
         """%%px --[no]block flags work"""
         ip = get_ipython()
         v = self.client[-1:]
         v.activate()
         v.block=False
         
         for block in (True, False):
             v.block = block
             ip.magic("pxconfig --verbose")
             with capture_output(display=False) as io:
                 ip.run_cell_magic("px", "", "1")
             if block:
                 assert io.stdout.startswith("Parallel"), io.stdout
             else:
                 assert io.stdout.startswith("Async"), io.stdout
             
             with capture_output(display=False) as io:
                 ip.run_cell_magic("px", "--block", "1")
             assert io.stdout.startswith("Parallel"), io.stdout
 
             with capture_output(display=False) as io:
                 ip.run_cell_magic("px", "--noblock", "1")
             assert io.stdout.startswith("Async"), io.stdout
     
     def test_cellpx_groupby_engine(self):
         """%%px --group-outputs=engine"""
         ip = get_ipython()
         v = self.client[:]
         v.block = True
         v.activate()
         
         v['generate_output'] = generate_output
         
         with capture_output(display=False) as io:
             ip.run_cell_magic('px', '--group-outputs=engine', 'generate_output()')
         
         self.assertNotIn('\n\n', io.stdout)
         lines = io.stdout.splitlines()
         expected = [
             r'\[stdout:\d+\]',
             'stdout',
             'stdout2',
             r'\[output:\d+\]',
             r'IPython\.core\.display\.HTML',
             r'IPython\.core\.display\.Math',
             r'Out\[\d+:\d+\]:.*IPython\.core\.display\.Math',
             ] * len(v)
         
         self.assertEqual(len(lines), len(expected), io.stdout)
         for line,expect in zip(lines, expected):
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
                 assert re.search(ex, line) is not None, "Expected %r in %r" % (ex, line)
         
         self._check_generated_stderr(io.stderr, len(v))
 
 
     def test_cellpx_groupby_order(self):
         """%%px --group-outputs=order"""
         ip = get_ipython()
         v = self.client[:]
         v.block = True
         v.activate()
         
         v['generate_output'] = generate_output
         
         with capture_output(display=False) as io:
             ip.run_cell_magic('px', '--group-outputs=order', 'generate_output()')
         
         self.assertNotIn('\n\n', io.stdout)
         lines = io.stdout.splitlines()
         expected = []
         expected.extend([
             r'\[stdout:\d+\]',
             'stdout',
             'stdout2',
         ] * len(v))
         expected.extend([
             r'\[output:\d+\]',
             'IPython.core.display.HTML',
         ] * len(v))
         expected.extend([
             r'\[output:\d+\]',
             'IPython.core.display.Math',
         ] * len(v))
         expected.extend([
             r'Out\[\d+:\d+\]:.*IPython\.core\.display\.Math'
         ] * len(v))
         
         self.assertEqual(len(lines), len(expected), io.stdout)
         for line,expect in zip(lines, expected):
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
                 assert re.search(ex, line) is not None, "Expected %r in %r" % (ex, line)
         
         self._check_generated_stderr(io.stderr, len(v))
 
     def test_cellpx_groupby_type(self):
         """%%px --group-outputs=type"""
         ip = get_ipython()
         v = self.client[:]
         v.block = True
         v.activate()
         
         v['generate_output'] = generate_output
         
         with capture_output(display=False) as io:
             ip.run_cell_magic('px', '--group-outputs=type', 'generate_output()')
         
         self.assertNotIn('\n\n', io.stdout)
         lines = io.stdout.splitlines()
         
         expected = []
         expected.extend([
             r'\[stdout:\d+\]',
             'stdout',
             'stdout2',
         ] * len(v))
         expected.extend([
             r'\[output:\d+\]',
             r'IPython\.core\.display\.HTML',
             r'IPython\.core\.display\.Math',
         ] * len(v))
         expected.extend([
             (r'Out\[\d+:\d+\]', r'IPython\.core\.display\.Math')
         ] * len(v))
         
         self.assertEqual(len(lines), len(expected), io.stdout)
         for line,expect in zip(lines, expected):
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
                 assert re.search(ex, line) is not None, "Expected %r in %r" % (ex, line)
         
         self._check_generated_stderr(io.stderr, len(v))
 
 
     def test_px_nonblocking(self):
         ip = get_ipython()
         v = self.client[-1:]
         v.activate()
         v.block=False
 
         ip.magic('px a=5')
         self.assertEqual(v['a'], [5])
         ip.magic('px a=10')
         self.assertEqual(v['a'], [10])
         ip.magic('pxconfig --verbose')
         with capture_output() as io:
             ar = ip.magic('px print (a)')
         self.assertIsInstance(ar, AsyncResult)
         self.assertIn('Async', io.stdout)
         self.assertNotIn('[stdout:', io.stdout)
         self.assertNotIn('\n\n', io.stdout)
         
         ar = ip.magic('px 1/0')
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
     
     def test_autopx_blocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=True
 
         with capture_output(display=False) as io:
             ip.magic('autopx')
             ip.run_cell('\n'.join(('a=5','b=12345','c=0')))
             ip.run_cell('b*=2')
             ip.run_cell('print (b)')
             ip.run_cell('b')
             ip.run_cell("b/c")
             ip.magic('autopx')
         
         output = io.stdout
         
         assert output.startswith('%autopx enabled'), output
         assert output.rstrip().endswith('%autopx disabled'), output
         self.assertIn('ZeroDivisionError', output)
         self.assertIn('\nOut[', output)
         self.assertIn(': 24690', output)
         ar = v.get_result(-1)
         self.assertEqual(v['a'], 5)
         self.assertEqual(v['b'], 24690)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
 
     def test_autopx_nonblocking(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         v.block=False
 
         with capture_output() as io:
             ip.magic('autopx')
             ip.run_cell('\n'.join(('a=5','b=10','c=0')))
             ip.run_cell('print (b)')
             ip.run_cell('import time; time.sleep(0.1)')
             ip.run_cell("b/c")
             ip.run_cell('b*=2')
             ip.magic('autopx')
         
         output = io.stdout.rstrip()
         
         assert output.startswith('%autopx enabled'), output
         assert output.endswith('%autopx disabled'), output
         self.assertNotIn('ZeroDivisionError', output)
         ar = v.get_result(-2)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
         # prevent TaskAborted on pulls, due to ZeroDivisionError
         time.sleep(0.5)
         self.assertEqual(v['a'], 5)
         # b*=2 will not fire, due to abort
         self.assertEqual(v['b'], 10)
     
     def test_result(self):
         ip = get_ipython()
         v = self.client[-1]
         v.activate()
         data = dict(a=111,b=222)
         v.push(data, block=True)
 
         for name in ('a', 'b'):
             ip.magic('px ' + name)
             with capture_output(display=False) as io:
                 ip.magic('pxresult')
             self.assertIn(str(data[name]), io.stdout)
         
     @dec.skipif_not_matplotlib
     def test_px_pylab(self):
         """%pylab works on engines"""
         ip = get_ipython()
         v = self.client[-1]
         v.block = True
         v.activate()
         
         with capture_output() as io:
             ip.magic("px %pylab inline")
         
         self.assertIn("Populating the interactive namespace from numpy and matplotlib", io.stdout)
         
         with capture_output(display=False) as io:
             ip.magic("px plot(rand(100))")
         self.assertIn('Out[', io.stdout)
         self.assertIn('matplotlib.lines', io.stdout)
     
     def test_pxconfig(self):
         ip = get_ipython()
         rc = self.client
         v = rc.activate(-1, '_tst')
         self.assertEqual(v.targets, rc.ids[-1])
         ip.magic("%pxconfig_tst -t :")
         self.assertEqual(v.targets, rc.ids)
         ip.magic("%pxconfig_tst -t ::2")
         self.assertEqual(v.targets, rc.ids[::2])
         ip.magic("%pxconfig_tst -t 1::2")
         self.assertEqual(v.targets, rc.ids[1::2])
         ip.magic("%pxconfig_tst -t 1")
         self.assertEqual(v.targets, 1)
         ip.magic("%pxconfig_tst --block")
         self.assertEqual(v.block, True)
         ip.magic("%pxconfig_tst --noblock")
         self.assertEqual(v.block, False)
     
     def test_cellpx_targets(self):
         """%%px --targets doesn't change defaults"""
         ip = get_ipython()
         rc = self.client
         view = rc.activate(rc.ids)
         self.assertEqual(view.targets, rc.ids)
         ip.magic('pxconfig --verbose')
         for cell in ("pass", "1/0"):
             with capture_output(display=False) as io:
                 try:
                     ip.run_cell_magic("px", "--targets all", cell)
                 except pmod.RemoteError:
                     pass
             self.assertIn('engine(s): all', io.stdout)
             self.assertEqual(view.targets, rc.ids)
 
 
     def test_cellpx_block(self):
         """%%px --block doesn't change default"""
         ip = get_ipython()
         rc = self.client
         view = rc.activate(rc.ids)
         view.block = False
         self.assertEqual(view.targets, rc.ids)
         ip.magic('pxconfig --verbose')
         for cell in ("pass", "1/0"):
             with capture_output(display=False) as io:
                 try:
                     ip.run_cell_magic("px", "--block", cell)
                 except pmod.RemoteError:
                     pass
             self.assertNotIn('Async', io.stdout)
             self.assertEqual(view.block, False)
 
 

IPython/parallel/tests/test_view.py

@@ -1,814 +1,813 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 base64
 import sys
 import platform
 import time
 from collections import namedtuple
 from tempfile import mktemp
 from StringIO import StringIO
 
 import zmq
 from nose import SkipTest
 from nose.plugins.attrib import attr
 
 from IPython.testing import decorators as dec
-from IPython.testing.ipunittest import ParametricTestCase
 from IPython.utils.io import capture_output
 
 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, total=True)
 
 point = namedtuple("point", "x y")
 
-class TestView(ClusterTestCase, ParametricTestCase):
+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=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"""
         tmpfile = mktemp()
         with open(tmpfile, 'w') as f:
             f.write("""def g():
                 return 5
                 """)
         v = self.client[-1]
         v.run(tmpfile, 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 = 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, assert_array_almost_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 = 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, assert_array_almost_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_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(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 = range(16)
         r = view.map_sync(f, data)
         self.assertEqual(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, 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 = 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, 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.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 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
     
     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'] = 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 = 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]]
-        yield self.assertEqual(len(e0), 1)
+        self.assertEqual(len(e0), 1)
         v = self.client[:]
-        yield self.assertEqual(len(v), len(self.client.ids))
+        self.assertEqual(len(v), len(self.client.ids))
         v = self.client.direct_view('all')
-        yield self.assertEqual(len(v), len(self.client.ids))
+        self.assertEqual(len(v), len(self.client.ids))
         v = self.client[:2]
-        yield self.assertEqual(len(v), 2)
+        self.assertEqual(len(v), 2)
         v = self.client[:1]
-        yield self.assertEqual(len(v), 1)
+        self.assertEqual(len(v), 1)
         v = self.client.load_balanced_view()
-        yield self.assertEqual(len(v), len(self.client.ids))
+        self.assertEqual(len(v), len(self.client.ids))
         # parametric tests seem to require manual closing?
         self.client.close()
 
     
     # 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(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(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))
 

IPython/utils/tests/test_io.py

@@ -1,86 +1,86 b''
 # encoding: utf-8
 """Tests for io.py"""
 
 #-----------------------------------------------------------------------------
 #  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 print_function
 
 import sys
 
 from StringIO import StringIO
 from subprocess import Popen, PIPE
+import unittest
 
 import nose.tools as nt
 
-from IPython.testing.ipunittest import ParametricTestCase
 from IPython.utils.io import Tee, capture_output
 from IPython.utils.py3compat import doctest_refactor_print
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 
 def test_tee_simple():
     "Very simple check with stdout only"
     chan = StringIO()
     text = 'Hello'
     tee = Tee(chan, channel='stdout')
     print(text, file=chan)
     nt.assert_equal(chan.getvalue(), text+"\n")
 
 
-class TeeTestCase(ParametricTestCase):
+class TeeTestCase(unittest.TestCase):
 
     def tchan(self, channel, check='close'):
         trap = StringIO()
         chan = StringIO()
         text = 'Hello'
         
         std_ori = getattr(sys, channel)
         setattr(sys, channel, trap)
 
         tee = Tee(chan, channel=channel)
         print(text, end='', file=chan)
         setattr(sys, channel, std_ori)
         trap_val = trap.getvalue()
         nt.assert_equal(chan.getvalue(), text)
         if check=='close':
             tee.close()
         else:
             del tee
 
     def test(self):
         for chan in ['stdout', 'stderr']:
             for check in ['close', 'del']:
-                yield self.tchan(chan, check)
+                self.tchan(chan, check)
 
 def test_io_init():
     """Test that io.stdin/out/err exist at startup"""
     for name in ('stdin', 'stdout', 'stderr'):
         cmd = doctest_refactor_print("from IPython.utils import io;print io.%s.__class__"%name)
         p = Popen([sys.executable, '-c', cmd],
                     stdout=PIPE)
         p.wait()
         classname = p.stdout.read().strip().decode('ascii')
         # __class__ is a reference to the class object in Python 3, so we can't
         # just test for string equality.
         assert 'IPython.utils.io.IOStream' in classname, classname
 
 def test_capture_output():
     """capture_output() context works"""
     
     with capture_output() as io:
         print('hi, stdout')
         print('hi, stderr', file=sys.stderr)
     
     nt.assert_equal(io.stdout, 'hi, stdout\n')
     nt.assert_equal(io.stderr, 'hi, stderr\n')