##// END OF EJS Templates
Backport PR #4118: ZMQ heartbeat channel: catch EINTR exceptions and continue....
Backport PR #4118: ZMQ heartbeat channel: catch EINTR exceptions and continue. The zmq heartbeat channel exits when it encounters an "Interrupted system call", thereby causing the qtconsole to report that it is not responding, despite the console working perfectly fine. Observed in a large Qt application embedding the ipython qtconsole in KDE desktop. Fixes #2310 Inspired by #501

File last commit:

r11134:83e4105a
r12317:bf63ce77
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.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):
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')