##// END OF EJS Templates
better flush iopub with AsyncResults...
better flush iopub with AsyncResults requesting metadata (e.g. ar.data or ar.stdout) will result in flushing iopub if the outputs are incomplete, so separate wait(0) need not be called. This also applies the workaround discussed in #2215

File last commit:

r7875:2549896b
r8140:c0973b13
Show More
test_io.py
86 lines | 2.7 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
#-----------------------------------------------------------------------------
from __future__ import print_function
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, 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(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(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)
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')