diff --git a/IPython/kernel/tests/test_kernel.py b/IPython/kernel/tests/test_kernel.py index 305dce7..253ce15 100644 --- a/IPython/kernel/tests/test_kernel.py +++ b/IPython/kernel/tests/test_kernel.py @@ -24,7 +24,7 @@ import nose.tools as nt from IPython.kernel import KernelManager from IPython.kernel.tests.test_message_spec import execute, flush_channels from IPython.testing import decorators as dec -from IPython.utils import path +from IPython.utils import path, py3compat #------------------------------------------------------------------------------- # Tests @@ -33,6 +33,9 @@ IPYTHONDIR = None save_env = None save_get_ipython_dir = None +STARTUP_TIMEOUT = 60 +TIMEOUT = 15 + def setup(): """setup temporary IPYTHONDIR for tests""" global IPYTHONDIR @@ -75,7 +78,7 @@ def new_kernel(): # wait for kernel to be ready KC.shell_channel.execute("import sys") - KC.shell_channel.get_msg(block=True, timeout=5) + KC.shell_channel.get_msg(block=True, timeout=STARTUP_TIMEOUT) flush_channels(KC) try: yield KC @@ -116,6 +119,8 @@ def _check_mp_mode(kc, expected=False, stream="stdout"): nt.assert_equal(eval(stdout.strip()), expected) +# printing tests + def test_simple_print(): """simple print statement in kernel""" with new_kernel() as kc: @@ -125,7 +130,6 @@ def test_simple_print(): nt.assert_equal(stdout, 'hi\n') nt.assert_equal(stderr, '') _check_mp_mode(kc, expected=False) - print ('hello') @dec.knownfailureif(sys.platform == 'win32', "subprocess prints fail on Windows") @@ -202,3 +206,47 @@ def test_subprocess_error(): _check_mp_mode(kc, expected=False) _check_mp_mode(kc, expected=False, stream="stderr") + +# 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") +