test_kernel.py
255 lines
| 7.8 KiB
| text/x-python
|
PythonLexer
|
r12335 | """test the IPython Kernel""" | ||
#------------------------------------------------------------------------------- | ||||
# Copyright (C) 2013 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 os | ||||
import shutil | ||||
import sys | ||||
import tempfile | ||||
from contextlib import contextmanager | ||||
from subprocess import PIPE | ||||
import nose.tools as nt | ||||
from IPython.kernel import KernelManager | ||||
from IPython.kernel.tests.test_message_spec import execute, flush_channels | ||||
|
r12421 | from IPython.testing import decorators as dec, tools as tt | ||
|
r12340 | from IPython.utils import path, py3compat | ||
|
r12335 | |||
#------------------------------------------------------------------------------- | ||||
# Tests | ||||
#------------------------------------------------------------------------------- | ||||
IPYTHONDIR = None | ||||
save_env = None | ||||
save_get_ipython_dir = None | ||||
|
r12340 | STARTUP_TIMEOUT = 60 | ||
TIMEOUT = 15 | ||||
|
r12335 | def setup(): | ||
"""setup temporary IPYTHONDIR for tests""" | ||||
global IPYTHONDIR | ||||
global save_env | ||||
global save_get_ipython_dir | ||||
IPYTHONDIR = tempfile.mkdtemp() | ||||
save_env = os.environ.copy() | ||||
os.environ["IPYTHONDIR"] = IPYTHONDIR | ||||
save_get_ipython_dir = path.get_ipython_dir | ||||
path.get_ipython_dir = lambda : IPYTHONDIR | ||||
def teardown(): | ||||
path.get_ipython_dir = save_get_ipython_dir | ||||
os.environ = save_env | ||||
try: | ||||
shutil.rmtree(IPYTHONDIR) | ||||
except (OSError, IOError): | ||||
# no such file | ||||
pass | ||||
@contextmanager | ||||
def new_kernel(): | ||||
"""start a kernel in a subprocess, and wait for it to be ready | ||||
Returns | ||||
------- | ||||
kernel_manager: connected KernelManager instance | ||||
""" | ||||
KM = KernelManager() | ||||
KM.start_kernel(stdout=PIPE, stderr=PIPE) | ||||
KC = KM.client() | ||||
KC.start_channels() | ||||
# wait for kernel to be ready | ||||
KC.shell_channel.execute("import sys") | ||||
|
r12340 | KC.shell_channel.get_msg(block=True, timeout=STARTUP_TIMEOUT) | ||
|
r12335 | flush_channels(KC) | ||
try: | ||||
yield KC | ||||
finally: | ||||
KC.stop_channels() | ||||
KM.shutdown_kernel() | ||||
def assemble_output(iopub): | ||||
"""assemble stdout/err from an execution""" | ||||
stdout = '' | ||||
stderr = '' | ||||
while True: | ||||
msg = iopub.get_msg(block=True, timeout=1) | ||||
msg_type = msg['msg_type'] | ||||
content = msg['content'] | ||||
if msg_type == 'status' and content['execution_state'] == 'idle': | ||||
# idle message signals end of output | ||||
break | ||||
elif msg['msg_type'] == 'stream': | ||||
if content['name'] == 'stdout': | ||||
stdout = stdout + content['data'] | ||||
elif content['name'] == 'stderr': | ||||
stderr = stderr + content['data'] | ||||
else: | ||||
raise KeyError("bad stream: %r" % content['name']) | ||||
else: | ||||
# other output, ignored | ||||
pass | ||||
return stdout, stderr | ||||
def _check_mp_mode(kc, expected=False, stream="stdout"): | ||||
execute(kc=kc, code="import sys") | ||||
flush_channels(kc) | ||||
msg_id, content = execute(kc=kc, code="print (sys.%s._check_mp_mode())" % stream) | ||||
stdout, stderr = assemble_output(kc.iopub_channel) | ||||
nt.assert_equal(eval(stdout.strip()), expected) | ||||
|
r12340 | # printing tests | ||
|
r12335 | def test_simple_print(): | ||
"""simple print statement in kernel""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
msg_id, content = execute(kc=kc, code="print ('hi')") | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout, 'hi\n') | ||||
nt.assert_equal(stderr, '') | ||||
_check_mp_mode(kc, expected=False) | ||||
@dec.knownfailureif(sys.platform == 'win32', "subprocess prints fail on Windows") | ||||
def test_subprocess_print(): | ||||
"""printing from forked mp.Process""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
_check_mp_mode(kc, expected=False) | ||||
flush_channels(kc) | ||||
np = 5 | ||||
code = '\n'.join([ | ||||
"from __future__ import print_function", | ||||
"import multiprocessing as mp", | ||||
"pool = [mp.Process(target=print, args=('hello', i,)) for i in range(%i)]" % np, | ||||
"for p in pool: p.start()", | ||||
"for p in pool: p.join()" | ||||
]) | ||||
expected = '\n'.join([ | ||||
"hello %s" % i for i in range(np) | ||||
]) + '\n' | ||||
msg_id, content = execute(kc=kc, code=code) | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout.count("hello"), np, stdout) | ||||
for n in range(np): | ||||
nt.assert_equal(stdout.count(str(n)), 1, stdout) | ||||
nt.assert_equal(stderr, '') | ||||
_check_mp_mode(kc, expected=False) | ||||
_check_mp_mode(kc, expected=False, stream="stderr") | ||||
def test_subprocess_noprint(): | ||||
"""mp.Process without print doesn't trigger iostream mp_mode""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
np = 5 | ||||
code = '\n'.join([ | ||||
"import multiprocessing as mp", | ||||
"pool = [mp.Process(target=range, args=(i,)) for i in range(%i)]" % np, | ||||
"for p in pool: p.start()", | ||||
"for p in pool: p.join()" | ||||
]) | ||||
msg_id, content = execute(kc=kc, code=code) | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout, '') | ||||
nt.assert_equal(stderr, '') | ||||
_check_mp_mode(kc, expected=False) | ||||
_check_mp_mode(kc, expected=False, stream="stderr") | ||||
@dec.knownfailureif(sys.platform == 'win32', "subprocess prints fail on Windows") | ||||
def test_subprocess_error(): | ||||
"""error in mp.Process doesn't crash""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
code = '\n'.join([ | ||||
"import multiprocessing as mp", | ||||
"p = mp.Process(target=int, args=('hi',))", | ||||
"p.start()", | ||||
"p.join()", | ||||
]) | ||||
msg_id, content = execute(kc=kc, code=code) | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout, '') | ||||
nt.assert_true("ValueError" in stderr, stderr) | ||||
_check_mp_mode(kc, expected=False) | ||||
_check_mp_mode(kc, expected=False, stream="stderr") | ||||
|
r12340 | |||
# raw_input tests | ||||
def test_raw_input(): | ||||
"""test [raw_]input""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
input_f = "input" if py3compat.PY3 else "raw_input" | ||||
theprompt = "prompt> " | ||||
code = 'print({input_f}("{theprompt}"))'.format(**locals()) | ||||
msg_id = kc.execute(code, allow_stdin=True) | ||||
msg = kc.get_stdin_msg(block=True, timeout=TIMEOUT) | ||||
nt.assert_equal(msg['header']['msg_type'], u'input_request') | ||||
content = msg['content'] | ||||
nt.assert_equal(content['prompt'], theprompt) | ||||
text = "some text" | ||||
kc.input(text) | ||||
reply = kc.get_shell_msg(block=True, timeout=TIMEOUT) | ||||
nt.assert_equal(reply['content']['status'], 'ok') | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout, text + "\n") | ||||
@dec.skipif(py3compat.PY3) | ||||
def test_eval_input(): | ||||
"""test input() on Python 2""" | ||||
with new_kernel() as kc: | ||||
iopub = kc.iopub_channel | ||||
input_f = "input" if py3compat.PY3 else "raw_input" | ||||
theprompt = "prompt> " | ||||
code = 'print(input("{theprompt}"))'.format(**locals()) | ||||
msg_id = kc.execute(code, allow_stdin=True) | ||||
msg = kc.get_stdin_msg(block=True, timeout=TIMEOUT) | ||||
nt.assert_equal(msg['header']['msg_type'], u'input_request') | ||||
content = msg['content'] | ||||
nt.assert_equal(content['prompt'], theprompt) | ||||
kc.input("1+1") | ||||
reply = kc.get_shell_msg(block=True, timeout=TIMEOUT) | ||||
nt.assert_equal(reply['content']['status'], 'ok') | ||||
stdout, stderr = assemble_output(iopub) | ||||
nt.assert_equal(stdout, "2\n") | ||||
|
r12421 | |||
def test_help_output(): | ||||
"""ipython kernel --help-all works""" | ||||
tt.help_all_output_test('kernel') | ||||