##// END OF EJS Templates
Merge pull request #1548 from minrk/ar_sugar...
Merge pull request #1548 from minrk/ar_sugar Add sugar methods/properties to AsyncResult that are generically useful: * `ar.wall_time` = received - submitted * `ar.serial_time` = sum of serial computation time * `ar.elapsed` = time since submission (wall_time if done) * `ar.progress` = (int) number of sub-tasks that have completed * `len(ar)` = # of tasks * `ar.wait_interactive()`: prints progress These are simple methods derived from the metadata/timestamps already created. But I've been persuaded by @wesm's practice of including simple methods that do useful (and/or cool) things. This also required/revealed some minor fixes/cleanup to clear_output in some cases: * dedent base `core.displaypub.clear_output`, so it's actually defined in the class * clear_output publishes `'\r\b'`, so it will clear terminal-like frontends that don't understand full clear_output behavior. * `core.display.clear_output()` still works, even outside an IPython session. Added a new notebook that shows how to use these new methods and how to do simple animations/progress bars using `clear_output()`. Added `Client.spin_thread(interval)` / `stop_spin_thread()` for running spin in a background thread, to keep zmq queue clear. This can be used to ensure that timing information is as accurate as possible (at the cost of having a background thread active).

File last commit:

r5390:c82649ea
r6485:e0b43119 merge
Show More
test_io.py
75 lines | 2.4 KiB | text/x-python | PythonLexer
# 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
#-----------------------------------------------------------------------------
import sys
from StringIO import StringIO
from subprocess import Popen, PIPE
import nose.tools as nt
from IPython.testing import decorators as dec
from IPython.utils.io import Tee
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 >> chan, text
nt.assert_equal(chan.getvalue(), text+"\n")
class TeeTestCase(dec.ParametricTestCase):
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 >> chan, text,
setattr(sys, channel, std_ori)
trap_val = trap.getvalue()
nt.assert_equals(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)
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