test_interactiveshell.py
903 lines
| 31.2 KiB
| text/x-python
|
PythonLexer
|
r5094 | # -*- coding: utf-8 -*- | ||
|
r3300 | """Tests for the key interactiveshell module. | ||
Historically the main classes in interactiveshell have been under-tested. This | ||||
module should grow as many single-method tests as possible to trap many of the | ||||
recurring bugs we seem to encounter with high-level interaction. | ||||
""" | ||||
|
r16570 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r8220 | import ast | ||
|
r5094 | import os | ||
|
r12774 | import signal | ||
|
r5094 | import shutil | ||
|
r6997 | import sys | ||
|
r5094 | import tempfile | ||
|
r3300 | import unittest | ||
|
r15597 | try: | ||
from unittest import mock | ||||
except ImportError: | ||||
import mock | ||||
|
r5094 | from os.path import join | ||
|
r3822 | |||
|
r6997 | import nose.tools as nt | ||
|
r17799 | from IPython.core.error import InputRejected | ||
|
r13525 | from IPython.core.inputtransformer import InputTransformer | ||
|
r17016 | from IPython.testing.decorators import ( | ||
skipif, skip_win32, onlyif_unicode_paths, onlyif_cmds_exist, | ||||
) | ||||
|
r8085 | from IPython.testing import tools as tt | ||
|
r3822 | from IPython.utils import io | ||
|
r17016 | from IPython.utils.process import find_cmd | ||
|
r13447 | from IPython.utils import py3compat | ||
|
r13366 | from IPython.utils.py3compat import unicode_type, PY3 | ||
if PY3: | ||||
from io import StringIO | ||||
else: | ||||
from StringIO import StringIO | ||||
|
r3300 | |||
#----------------------------------------------------------------------------- | ||||
|
r6997 | # Globals | ||
#----------------------------------------------------------------------------- | ||||
# This is used by every single test, no point repeating it ad nauseam | ||||
ip = get_ipython() | ||||
#----------------------------------------------------------------------------- | ||||
|
r3300 | # Tests | ||
#----------------------------------------------------------------------------- | ||||
|
r21316 | class DerivedInterrupt(KeyboardInterrupt): | ||
pass | ||||
|
r3300 | class InteractiveShellTestCase(unittest.TestCase): | ||
def test_naked_string_cells(self): | ||||
"""Test that cells with only naked strings are fully executed""" | ||||
# First, single-line inputs | ||||
ip.run_cell('"a"\n') | ||||
|
r7874 | self.assertEqual(ip.user_ns['_'], 'a') | ||
|
r3300 | # And also multi-line cells | ||
ip.run_cell('"""a\nb"""\n') | ||||
|
r7874 | self.assertEqual(ip.user_ns['_'], 'a\nb') | ||
|
r3499 | |||
|
r3441 | def test_run_empty_cell(self): | ||
"""Just make sure we don't get a horrible error with a blank | ||||
cell of input. Yes, I did overlook that.""" | ||||
|
r3706 | old_xc = ip.execution_count | ||
|
r19630 | res = ip.run_cell('') | ||
|
r7874 | self.assertEqual(ip.execution_count, old_xc) | ||
|
r19630 | self.assertEqual(res.execution_count, None) | ||
|
r3467 | |||
|
r3499 | def test_run_cell_multiline(self): | ||
|
r3467 | """Multi-block, multi-line cells must execute correctly. | ||
""" | ||||
src = '\n'.join(["x=1", | ||||
"y=2", | ||||
"if 1:", | ||||
" x += 1", | ||||
" y += 1",]) | ||||
|
r19630 | res = ip.run_cell(src) | ||
|
r7874 | self.assertEqual(ip.user_ns['x'], 2) | ||
self.assertEqual(ip.user_ns['y'], 3) | ||||
|
r19630 | self.assertEqual(res.success, True) | ||
self.assertEqual(res.result, None) | ||||
|
r3499 | |||
def test_multiline_string_cells(self): | ||||
|
r3500 | "Code sprinkled with multiline strings should execute (GH-306)" | ||
|
r3499 | ip.run_cell('tmp=0') | ||
|
r7874 | self.assertEqual(ip.user_ns['tmp'], 0) | ||
|
r19630 | res = ip.run_cell('tmp=1;"""a\nb"""\n') | ||
|
r7874 | self.assertEqual(ip.user_ns['tmp'], 1) | ||
|
r19630 | self.assertEqual(res.success, True) | ||
self.assertEqual(res.result, "a\nb") | ||||
|
r3500 | |||
def test_dont_cache_with_semicolon(self): | ||||
"Ending a line with semicolon should not cache the returned object (GH-307)" | ||||
oldlen = len(ip.user_ns['Out']) | ||||
|
r15886 | for cell in ['1;', '1;1;']: | ||
|
r19630 | res = ip.run_cell(cell, store_history=True) | ||
|
r15883 | newlen = len(ip.user_ns['Out']) | ||
self.assertEqual(oldlen, newlen) | ||||
|
r19630 | self.assertIsNone(res.result) | ||
|
r15883 | i = 0 | ||
|
r3500 | #also test the default caching behavior | ||
|
r15886 | for cell in ['1', '1;1']: | ||
|
r15883 | ip.run_cell(cell, store_history=True) | ||
newlen = len(ip.user_ns['Out']) | ||||
i += 1 | ||||
self.assertEqual(oldlen+i, newlen) | ||||
|
r3503 | |||
|
r19630 | def test_syntax_error(self): | ||
res = ip.run_cell("raise = 3") | ||||
self.assertIsInstance(res.error_before_exec, SyntaxError) | ||||
|
r3501 | def test_In_variable(self): | ||
"Verify that In variable grows with user input (GH-284)" | ||||
oldlen = len(ip.user_ns['In']) | ||||
|
r4995 | ip.run_cell('1;', store_history=True) | ||
|
r3501 | newlen = len(ip.user_ns['In']) | ||
|
r7874 | self.assertEqual(oldlen+1, newlen) | ||
self.assertEqual(ip.user_ns['In'][-1],'1;') | ||||
|
r3707 | |||
def test_magic_names_in_string(self): | ||||
|
r3709 | ip.run_cell('a = """\n%exit\n"""') | ||
|
r7874 | self.assertEqual(ip.user_ns['a'], '\n%exit\n') | ||
|
r3821 | |||
|
r3908 | def test_trailing_newline(self): | ||
"""test that running !(command) does not raise a SyntaxError""" | ||||
ip.run_cell('!(true)\n', False) | ||||
ip.run_cell('!(true)\n\n\n', False) | ||||
|
r4281 | |||
def test_gh_597(self): | ||||
|
r4375 | """Pretty-printing lists of objects with non-ascii reprs may cause | ||
problems.""" | ||||
|
r4281 | class Spam(object): | ||
def __repr__(self): | ||||
return "\xe9"*50 | ||||
import IPython.core.formatters | ||||
f = IPython.core.formatters.PlainTextFormatter() | ||||
f([Spam(),Spam()]) | ||||
|
r4795 | |||
|
r5456 | |||
|
r4795 | def test_future_flags(self): | ||
"""Check that future flags are used for parsing code (gh-777)""" | ||||
ip.run_cell('from __future__ import print_function') | ||||
try: | ||||
ip.run_cell('prfunc_return_val = print(1,2, sep=" ")') | ||||
assert 'prfunc_return_val' in ip.user_ns | ||||
finally: | ||||
# Reset compiler flags so we don't mess up other tests. | ||||
ip.compile.reset_compiler_flags() | ||||
|
r4811 | |||
def test_future_unicode(self): | ||||
"""Check that unicode_literals is imported from __future__ (gh #786)""" | ||||
try: | ||||
|
r4820 | ip.run_cell(u'byte_str = "a"') | ||
|
r4821 | assert isinstance(ip.user_ns['byte_str'], str) # string literals are byte strings by default | ||
|
r4811 | ip.run_cell('from __future__ import unicode_literals') | ||
|
r4820 | ip.run_cell(u'unicode_str = "a"') | ||
|
r13353 | assert isinstance(ip.user_ns['unicode_str'], unicode_type) # strings literals are now unicode | ||
|
r4811 | finally: | ||
# Reset compiler flags so we don't mess up other tests. | ||||
ip.compile.reset_compiler_flags() | ||||
|
r5456 | |||
def test_can_pickle(self): | ||||
"Can we pickle objects defined interactively (GH-29)" | ||||
ip = get_ipython() | ||||
ip.reset() | ||||
ip.run_cell(("class Mylist(list):\n" | ||||
" def __init__(self,x=[]):\n" | ||||
" list.__init__(self,x)")) | ||||
ip.run_cell("w=Mylist([1,2,3])") | ||||
|
r13354 | from pickle import dumps | ||
|
r5456 | |||
# We need to swap in our main module - this is only necessary | ||||
# inside the test framework, because IPython puts the interactive module | ||||
# in place (but the test framework undoes this). | ||||
_main = sys.modules['__main__'] | ||||
sys.modules['__main__'] = ip.user_module | ||||
try: | ||||
res = dumps(ip.user_ns["w"]) | ||||
finally: | ||||
sys.modules['__main__'] = _main | ||||
self.assertTrue(isinstance(res, bytes)) | ||||
def test_global_ns(self): | ||||
"Code in functions must be able to access variables outside them." | ||||
ip = get_ipython() | ||||
ip.run_cell("a = 10") | ||||
|
r5465 | ip.run_cell(("def f(x):\n" | ||
|
r5456 | " return x + a")) | ||
ip.run_cell("b = f(12)") | ||||
self.assertEqual(ip.user_ns["b"], 22) | ||||
|
r4991 | |||
def test_bad_custom_tb(self): | ||||
"""Check that InteractiveShell is protected from bad custom exception handlers""" | ||||
from IPython.utils import io | ||||
save_stderr = io.stderr | ||||
try: | ||||
# capture stderr | ||||
io.stderr = StringIO() | ||||
|
r4999 | ip.set_custom_exc((IOError,), lambda etype,value,tb: 1/0) | ||
|
r7874 | self.assertEqual(ip.custom_exceptions, (IOError,)) | ||
|
r4991 | ip.run_cell(u'raise IOError("foo")') | ||
|
r7874 | self.assertEqual(ip.custom_exceptions, ()) | ||
|
r4991 | self.assertTrue("Custom TB Handler failed" in io.stderr.getvalue()) | ||
finally: | ||||
io.stderr = save_stderr | ||||
|
r4999 | def test_bad_custom_tb_return(self): | ||
"""Check that InteractiveShell is protected from bad return types in custom exception handlers""" | ||||
from IPython.utils import io | ||||
save_stderr = io.stderr | ||||
try: | ||||
# capture stderr | ||||
io.stderr = StringIO() | ||||
ip.set_custom_exc((NameError,),lambda etype,value,tb, tb_offset=None: 1) | ||||
|
r7874 | self.assertEqual(ip.custom_exceptions, (NameError,)) | ||
|
r4999 | ip.run_cell(u'a=abracadabra') | ||
|
r7874 | self.assertEqual(ip.custom_exceptions, ()) | ||
|
r4999 | self.assertTrue("Custom TB Handler failed" in io.stderr.getvalue()) | ||
finally: | ||||
io.stderr = save_stderr | ||||
|
r5068 | def test_drop_by_id(self): | ||
myvars = {"a":object(), "b":object(), "c": object()} | ||||
ip.push(myvars, interactive=False) | ||||
for name in myvars: | ||||
assert name in ip.user_ns, name | ||||
assert name in ip.user_ns_hidden, name | ||||
ip.user_ns['b'] = 12 | ||||
ip.drop_by_id(myvars) | ||||
for name in ["a", "c"]: | ||||
assert name not in ip.user_ns, name | ||||
assert name not in ip.user_ns_hidden, name | ||||
assert ip.user_ns['b'] == 12 | ||||
ip.reset() | ||||
|
r5094 | |||
|
r5357 | def test_var_expand(self): | ||
|
r5366 | ip.user_ns['f'] = u'Ca\xf1o' | ||
self.assertEqual(ip.var_expand(u'echo $f'), u'echo Ca\xf1o') | ||||
|
r6124 | self.assertEqual(ip.var_expand(u'echo {f}'), u'echo Ca\xf1o') | ||
self.assertEqual(ip.var_expand(u'echo {f[:-1]}'), u'echo Ca\xf1') | ||||
self.assertEqual(ip.var_expand(u'echo {1*2}'), u'echo 2') | ||||
|
r5366 | |||
ip.user_ns['f'] = b'Ca\xc3\xb1o' | ||||
|
r5357 | # This should not raise any exception: | ||
ip.var_expand(u'echo $f') | ||||
|
r6124 | |||
|
r7331 | def test_var_expand_local(self): | ||
|
r7333 | """Test local variable expansion in !system and %magic calls""" | ||
# !system | ||||
|
r7331 | ip.run_cell('def test():\n' | ||
' lvar = "ttt"\n' | ||||
' ret = !echo {lvar}\n' | ||||
' return ret[0]\n') | ||||
res = ip.user_ns['test']() | ||||
nt.assert_in('ttt', res) | ||||
|
r7333 | |||
# %magic | ||||
ip.run_cell('def makemacro():\n' | ||||
' macroname = "macro_var_expand_locals"\n' | ||||
' %macro {macroname} codestr\n') | ||||
ip.user_ns['codestr'] = "str(12)" | ||||
ip.run_cell('makemacro()') | ||||
nt.assert_in('macro_var_expand_locals', ip.user_ns) | ||||
|
r7331 | |||
|
r8224 | def test_var_expand_self(self): | ||
"""Test variable expansion with the name 'self', which was failing. | ||||
See https://github.com/ipython/ipython/issues/1878#issuecomment-7698218 | ||||
""" | ||||
ip.run_cell('class cTest:\n' | ||||
' classvar="see me"\n' | ||||
' def test(self):\n' | ||||
' res = !echo Variable: {self.classvar}\n' | ||||
|
r8225 | ' return res[0]\n') | ||
|
r8224 | nt.assert_in('see me', ip.user_ns['cTest']().test()) | ||
|
r6124 | def test_bad_var_expand(self): | ||
"""var_expand on invalid formats shouldn't raise""" | ||||
# SyntaxError | ||||
self.assertEqual(ip.var_expand(u"{'a':5}"), u"{'a':5}") | ||||
# NameError | ||||
self.assertEqual(ip.var_expand(u"{asdf}"), u"{asdf}") | ||||
# ZeroDivisionError | ||||
self.assertEqual(ip.var_expand(u"{1/0}"), u"{1/0}") | ||||
|
r6802 | |||
|
r15597 | def test_silent_postexec(self): | ||
|
r15607 | """run_cell(silent=True) doesn't invoke pre/post_run_cell callbacks""" | ||
|
r15597 | pre_explicit = mock.Mock() | ||
pre_always = mock.Mock() | ||||
post_explicit = mock.Mock() | ||||
post_always = mock.Mock() | ||||
|
r6802 | |||
|
r15607 | ip.events.register('pre_run_cell', pre_explicit) | ||
|
r15605 | ip.events.register('pre_execute', pre_always) | ||
|
r15607 | ip.events.register('post_run_cell', post_explicit) | ||
|
r15605 | ip.events.register('post_execute', post_always) | ||
|
r15597 | |||
try: | ||||
ip.run_cell("1", silent=True) | ||||
assert pre_always.called | ||||
assert not pre_explicit.called | ||||
assert post_always.called | ||||
assert not post_explicit.called | ||||
# double-check that non-silent exec did what we expected | ||||
# silent to avoid | ||||
ip.run_cell("1") | ||||
assert pre_explicit.called | ||||
assert post_explicit.called | ||||
finally: | ||||
# remove post-exec | ||||
|
r18547 | ip.events.unregister('pre_run_cell', pre_explicit) | ||
ip.events.unregister('pre_execute', pre_always) | ||||
ip.events.unregister('post_run_cell', post_explicit) | ||||
ip.events.unregister('post_execute', post_always) | ||||
|
r6802 | |||
def test_silent_noadvance(self): | ||||
"""run_cell(silent=True) doesn't advance execution_count""" | ||||
ec = ip.execution_count | ||||
# silent should force store_history=False | ||||
ip.run_cell("1", store_history=True, silent=True) | ||||
|
r7874 | self.assertEqual(ec, ip.execution_count) | ||
|
r6802 | # double-check that non-silent exec did what we expected | ||
# silent to avoid | ||||
ip.run_cell("1", store_history=True) | ||||
|
r7874 | self.assertEqual(ec+1, ip.execution_count) | ||
|
r6802 | |||
def test_silent_nodisplayhook(self): | ||||
"""run_cell(silent=True) doesn't trigger displayhook""" | ||||
d = dict(called=False) | ||||
trap = ip.display_trap | ||||
save_hook = trap.hook | ||||
def failing_hook(*args, **kwargs): | ||||
d['called'] = True | ||||
try: | ||||
trap.hook = failing_hook | ||||
|
r19630 | res = ip.run_cell("1", silent=True) | ||
|
r6802 | self.assertFalse(d['called']) | ||
|
r19630 | self.assertIsNone(res.result) | ||
|
r6802 | # double-check that non-silent exec did what we expected | ||
# silent to avoid | ||||
ip.run_cell("1") | ||||
self.assertTrue(d['called']) | ||||
finally: | ||||
trap.hook = save_hook | ||||
|
r5357 | |||
|
r6643 | @skipif(sys.version_info[0] >= 3, "softspace removed in py3") | ||
def test_print_softspace(self): | ||||
"""Verify that softspace is handled correctly when executing multiple | ||||
statements. | ||||
In [1]: print 1; print 2 | ||||
1 | ||||
2 | ||||
In [2]: print 1,; print 2 | ||||
1 2 | ||||
""" | ||||
|
r6997 | |||
def test_ofind_line_magic(self): | ||||
from IPython.core.magic import register_line_magic | ||||
@register_line_magic | ||||
def lmagic(line): | ||||
"A line magic" | ||||
# Get info on line magic | ||||
lfind = ip._ofind('lmagic') | ||||
info = dict(found=True, isalias=False, ismagic=True, | ||||
namespace = 'IPython internal', obj= lmagic.__wrapped__, | ||||
parent = None) | ||||
nt.assert_equal(lfind, info) | ||||
def test_ofind_cell_magic(self): | ||||
from IPython.core.magic import register_cell_magic | ||||
@register_cell_magic | ||||
def cmagic(line, cell): | ||||
"A cell magic" | ||||
# Get info on cell magic | ||||
find = ip._ofind('cmagic') | ||||
info = dict(found=True, isalias=False, ismagic=True, | ||||
namespace = 'IPython internal', obj= cmagic.__wrapped__, | ||||
parent = None) | ||||
nt.assert_equal(find, info) | ||||
|
r17024 | |||
def test_ofind_property_with_error(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
raise NotImplementedError() | ||||
a = A() | ||||
found = ip._ofind('a.foo', [('locals', locals())]) | ||||
info = dict(found=True, isalias=False, ismagic=False, | ||||
namespace='locals', obj=A.foo, parent=a) | ||||
nt.assert_equal(found, info) | ||||
def test_ofind_multiple_attribute_lookups(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
raise NotImplementedError() | ||||
a = A() | ||||
a.a = A() | ||||
a.a.a = A() | ||||
found = ip._ofind('a.a.a.foo', [('locals', locals())]) | ||||
info = dict(found=True, isalias=False, ismagic=False, | ||||
namespace='locals', obj=A.foo, parent=a.a.a) | ||||
nt.assert_equal(found, info) | ||||
def test_ofind_slotted_attributes(self): | ||||
class A(object): | ||||
__slots__ = ['foo'] | ||||
def __init__(self): | ||||
self.foo = 'bar' | ||||
a = A() | ||||
found = ip._ofind('a.foo', [('locals', locals())]) | ||||
info = dict(found=True, isalias=False, ismagic=False, | ||||
namespace='locals', obj=a.foo, parent=a) | ||||
nt.assert_equal(found, info) | ||||
found = ip._ofind('a.bar', [('locals', locals())]) | ||||
info = dict(found=False, isalias=False, ismagic=False, | ||||
namespace=None, obj=None, parent=a) | ||||
nt.assert_equal(found, info) | ||||
def test_ofind_prefers_property_to_instance_level_attribute(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
return 'bar' | ||||
a = A() | ||||
a.__dict__['foo'] = 'baz' | ||||
nt.assert_equal(a.foo, 'bar') | ||||
found = ip._ofind('a.foo', [('locals', locals())]) | ||||
nt.assert_is(found['obj'], A.foo) | ||||
|
r7110 | def test_custom_exception(self): | ||
called = [] | ||||
def my_handler(shell, etype, value, tb, tb_offset=None): | ||||
called.append(etype) | ||||
shell.showtraceback((etype, value, tb), tb_offset=tb_offset) | ||||
ip.set_custom_exc((ValueError,), my_handler) | ||||
try: | ||||
|
r19630 | res = ip.run_cell("raise ValueError('test')") | ||
|
r7110 | # Check that this was called, and only once. | ||
self.assertEqual(called, [ValueError]) | ||||
|
r19630 | # Check that the error is on the result object | ||
self.assertIsInstance(res.error_in_exec, ValueError) | ||||
|
r7110 | finally: | ||
# Reset the custom exception hook | ||||
ip.set_custom_exc((), None) | ||||
|
r9139 | |||
@skipif(sys.version_info[0] >= 3, "no differences with __future__ in py3") | ||||
def test_future_environment(self): | ||||
"Can we run code with & without the shell's __future__ imports?" | ||||
ip.run_cell("from __future__ import division") | ||||
ip.run_cell("a = 1/2", shell_futures=True) | ||||
self.assertEqual(ip.user_ns['a'], 0.5) | ||||
ip.run_cell("b = 1/2", shell_futures=False) | ||||
self.assertEqual(ip.user_ns['b'], 0) | ||||
ip.compile.reset_compiler_flags() | ||||
# This shouldn't leak to the shell's compiler | ||||
ip.run_cell("from __future__ import division \nc=1/2", shell_futures=False) | ||||
self.assertEqual(ip.user_ns['c'], 0.5) | ||||
ip.run_cell("d = 1/2", shell_futures=True) | ||||
self.assertEqual(ip.user_ns['d'], 0) | ||||
|
r6997 | |||
|
r17337 | def test_mktempfile(self): | ||
filename = ip.mktempfile() | ||||
# Check that we can open the file again on Windows | ||||
with open(filename, 'w') as f: | ||||
f.write('abc') | ||||
filename = ip.mktempfile(data='blah') | ||||
with open(filename, 'r') as f: | ||||
self.assertEqual(f.read(), 'blah') | ||||
|
r5357 | |||
|
r17592 | def test_new_main_mod(self): | ||
# Smoketest to check that this accepts a unicode module name | ||||
name = u'jiefmw' | ||||
mod = ip.new_main_mod(u'%s.py' % name, name) | ||||
self.assertEqual(mod.__name__, name) | ||||
|
r19114 | def test_get_exception_only(self): | ||
try: | ||||
raise KeyboardInterrupt | ||||
except KeyboardInterrupt: | ||||
msg = ip.get_exception_only() | ||||
self.assertEqual(msg, 'KeyboardInterrupt\n') | ||||
try: | ||||
raise DerivedInterrupt("foo") | ||||
except KeyboardInterrupt: | ||||
msg = ip.get_exception_only() | ||||
|
r19119 | if sys.version_info[0] <= 2: | ||
self.assertEqual(msg, 'DerivedInterrupt: foo\n') | ||||
else: | ||||
self.assertEqual(msg, 'IPython.core.tests.test_interactiveshell.DerivedInterrupt: foo\n') | ||||
|
r19114 | |||
|
r5094 | class TestSafeExecfileNonAsciiPath(unittest.TestCase): | ||
|
r12168 | @onlyif_unicode_paths | ||
|
r5094 | def setUp(self): | ||
self.BASETESTDIR = tempfile.mkdtemp() | ||||
self.TESTDIR = join(self.BASETESTDIR, u"åäö") | ||||
os.mkdir(self.TESTDIR) | ||||
with open(join(self.TESTDIR, u"åäötestscript.py"), "w") as sfile: | ||||
sfile.write("pass\n") | ||||
|
r13447 | self.oldpath = py3compat.getcwd() | ||
|
r5094 | os.chdir(self.TESTDIR) | ||
self.fname = u"åäötestscript.py" | ||||
def tearDown(self): | ||||
os.chdir(self.oldpath) | ||||
shutil.rmtree(self.BASETESTDIR) | ||||
|
r12168 | @onlyif_unicode_paths | ||
|
r5094 | def test_1(self): | ||
"""Test safe_execfile with non-ascii path | ||||
""" | ||||
|
r6997 | ip.safe_execfile(self.fname, {}, raise_exceptions=True) | ||
|
r5314 | |||
|
r12774 | class ExitCodeChecks(tt.TempFileMixin): | ||
def test_exit_code_ok(self): | ||||
self.system('exit 0') | ||||
self.assertEqual(ip.user_ns['_exit_code'], 0) | ||||
def test_exit_code_error(self): | ||||
self.system('exit 1') | ||||
self.assertEqual(ip.user_ns['_exit_code'], 1) | ||||
|
r17016 | |||
|
r12774 | @skipif(not hasattr(signal, 'SIGALRM')) | ||
def test_exit_code_signal(self): | ||||
self.mktmp("import signal, time\n" | ||||
"signal.setitimer(signal.ITIMER_REAL, 0.1)\n" | ||||
"time.sleep(1)\n") | ||||
self.system("%s %s" % (sys.executable, self.fname)) | ||||
self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGALRM) | ||||
|
r17016 | |||
@onlyif_cmds_exist("csh") | ||||
def test_exit_code_signal_csh(self): | ||||
SHELL = os.environ.get('SHELL', None) | ||||
os.environ['SHELL'] = find_cmd("csh") | ||||
try: | ||||
self.test_exit_code_signal() | ||||
finally: | ||||
if SHELL is not None: | ||||
os.environ['SHELL'] = SHELL | ||||
else: | ||||
del os.environ['SHELL'] | ||||
|
r12774 | |||
class TestSystemRaw(unittest.TestCase, ExitCodeChecks): | ||||
system = ip.system_raw | ||||
|
r5314 | |||
|
r12168 | @onlyif_unicode_paths | ||
|
r5314 | def test_1(self): | ||
"""Test system_raw with non-ascii cmd | ||||
""" | ||||
|
r12774 | cmd = u'''python -c "'åäö'" ''' | ||
|
r6997 | ip.system_raw(cmd) | ||
|
r5493 | |||
|
r18765 | @mock.patch('subprocess.call', side_effect=KeyboardInterrupt) | ||
@mock.patch('os.system', side_effect=KeyboardInterrupt) | ||||
def test_control_c(self, *mocks): | ||||
|
r18732 | try: | ||
self.system("sleep 1 # wont happen") | ||||
except KeyboardInterrupt: | ||||
self.fail("system call should intercept " | ||||
"keyboard interrupt from subprocess.call") | ||||
self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGINT) | ||||
|
r12774 | # TODO: Exit codes are currently ignored on Windows. | ||
class TestSystemPipedExitCode(unittest.TestCase, ExitCodeChecks): | ||||
system = ip.system_piped | ||||
|
r12767 | @skip_win32 | ||
|
r12774 | def test_exit_code_ok(self): | ||
ExitCodeChecks.test_exit_code_ok(self) | ||||
@skip_win32 | ||||
def test_exit_code_error(self): | ||||
ExitCodeChecks.test_exit_code_error(self) | ||||
@skip_win32 | ||||
def test_exit_code_signal(self): | ||||
ExitCodeChecks.test_exit_code_signal(self) | ||||
|
r5493 | |||
|
r8085 | class TestModules(unittest.TestCase, tt.TempFileMixin): | ||
def test_extraneous_loads(self): | ||||
"""Test we're not loading modules on startup that we shouldn't. | ||||
""" | ||||
self.mktmp("import sys\n" | ||||
"print('numpy' in sys.modules)\n" | ||||
|
r21326 | "print('ipyparallel' in sys.modules)\n" | ||
|
r21337 | "print('ipykernel' in sys.modules)\n" | ||
|
r8085 | ) | ||
out = "False\nFalse\nFalse\n" | ||||
tt.ipexec_validate(self.fname, out) | ||||
|
r8220 | class Negator(ast.NodeTransformer): | ||
"""Negates all number literals in an AST.""" | ||||
def visit_Num(self, node): | ||||
node.n = -node.n | ||||
return node | ||||
class TestAstTransform(unittest.TestCase): | ||||
def setUp(self): | ||||
self.negator = Negator() | ||||
ip.ast_transformers.append(self.negator) | ||||
def tearDown(self): | ||||
ip.ast_transformers.remove(self.negator) | ||||
def test_run_cell(self): | ||||
with tt.AssertPrints('-34'): | ||||
ip.run_cell('print (12 + 22)') | ||||
# A named reference to a number shouldn't be transformed. | ||||
ip.user_ns['n'] = 55 | ||||
with tt.AssertNotPrints('-55'): | ||||
ip.run_cell('print (n)') | ||||
|
r8485 | |||
def test_timeit(self): | ||||
called = set() | ||||
def f(x): | ||||
called.add(x) | ||||
ip.push({'f':f}) | ||||
with tt.AssertPrints("best of "): | ||||
ip.run_line_magic("timeit", "-n1 f(1)") | ||||
self.assertEqual(called, set([-1])) | ||||
called.clear() | ||||
with tt.AssertPrints("best of "): | ||||
ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)") | ||||
self.assertEqual(called, set([-2, -3])) | ||||
|
r8486 | |||
def test_time(self): | ||||
called = [] | ||||
def f(x): | ||||
called.append(x) | ||||
ip.push({'f':f}) | ||||
# Test with an expression | ||||
|
r9901 | with tt.AssertPrints("Wall time: "): | ||
|
r8486 | ip.run_line_magic("time", "f(5+9)") | ||
self.assertEqual(called, [-14]) | ||||
called[:] = [] | ||||
# Test with a statement (different code path) | ||||
|
r9901 | with tt.AssertPrints("Wall time: "): | ||
|
r8486 | ip.run_line_magic("time", "a = f(-3 + -2)") | ||
self.assertEqual(called, [5]) | ||||
|
r8488 | |||
def test_macro(self): | ||||
ip.push({'a':10}) | ||||
# The AST transformation makes this do a+=-1 | ||||
ip.define_macro("amacro", "a+=1\nprint(a)") | ||||
with tt.AssertPrints("9"): | ||||
ip.run_cell("amacro") | ||||
with tt.AssertPrints("8"): | ||||
ip.run_cell("amacro") | ||||
|
r5493 | |||
|
r8479 | class IntegerWrapper(ast.NodeTransformer): | ||
"""Wraps all integers in a call to Integer()""" | ||||
def visit_Num(self, node): | ||||
if isinstance(node.n, int): | ||||
return ast.Call(func=ast.Name(id='Integer', ctx=ast.Load()), | ||||
args=[node], keywords=[]) | ||||
|
r8508 | return node | ||
|
r8479 | |||
class TestAstTransform2(unittest.TestCase): | ||||
def setUp(self): | ||||
self.intwrapper = IntegerWrapper() | ||||
ip.ast_transformers.append(self.intwrapper) | ||||
self.calls = [] | ||||
def Integer(*args): | ||||
self.calls.append(args) | ||||
|
r8485 | return args | ||
|
r8479 | ip.push({"Integer": Integer}) | ||
def tearDown(self): | ||||
ip.ast_transformers.remove(self.intwrapper) | ||||
del ip.user_ns['Integer'] | ||||
def test_run_cell(self): | ||||
ip.run_cell("n = 2") | ||||
self.assertEqual(self.calls, [(2,)]) | ||||
|
r8508 | |||
# This shouldn't throw an error | ||||
ip.run_cell("o = 2.0") | ||||
self.assertEqual(ip.user_ns['o'], 2.0) | ||||
|
r8485 | |||
def test_timeit(self): | ||||
called = set() | ||||
def f(x): | ||||
called.add(x) | ||||
ip.push({'f':f}) | ||||
with tt.AssertPrints("best of "): | ||||
ip.run_line_magic("timeit", "-n1 f(1)") | ||||
self.assertEqual(called, set([(1,)])) | ||||
called.clear() | ||||
with tt.AssertPrints("best of "): | ||||
ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)") | ||||
self.assertEqual(called, set([(2,), (3,)])) | ||||
|
r8479 | |||
|
r8221 | class ErrorTransformer(ast.NodeTransformer): | ||
"""Throws an error when it sees a number.""" | ||||
|
r17797 | def visit_Num(self, node): | ||
|
r8221 | raise ValueError("test") | ||
class TestAstTransformError(unittest.TestCase): | ||||
def test_unregistering(self): | ||||
err_transformer = ErrorTransformer() | ||||
ip.ast_transformers.append(err_transformer) | ||||
with tt.AssertPrints("unregister", channel='stderr'): | ||||
ip.run_cell("1 + 2") | ||||
# This should have been removed. | ||||
nt.assert_not_in(err_transformer, ip.ast_transformers) | ||||
|
r5493 | |||
|
r17798 | |||
class StringRejector(ast.NodeTransformer): | ||||
"""Throws an InputRejected when it sees a string literal. | ||||
Used to verify that NodeTransformers can signal that a piece of code should | ||||
not be executed by throwing an InputRejected. | ||||
""" | ||||
def visit_Str(self, node): | ||||
raise InputRejected("test") | ||||
class TestAstTransformInputRejection(unittest.TestCase): | ||||
def setUp(self): | ||||
self.transformer = StringRejector() | ||||
ip.ast_transformers.append(self.transformer) | ||||
def tearDown(self): | ||||
ip.ast_transformers.remove(self.transformer) | ||||
def test_input_rejection(self): | ||||
"""Check that NodeTransformers can reject input.""" | ||||
expect_exception_tb = tt.AssertPrints("InputRejected: test") | ||||
expect_no_cell_output = tt.AssertNotPrints("'unsafe'", suppress=False) | ||||
# Run the same check twice to verify that the transformer is not | ||||
# disabled after raising. | ||||
with expect_exception_tb, expect_no_cell_output: | ||||
ip.run_cell("'unsafe'") | ||||
with expect_exception_tb, expect_no_cell_output: | ||||
|
r19630 | res = ip.run_cell("'unsafe'") | ||
self.assertIsInstance(res.error_before_exec, InputRejected) | ||||
|
r17798 | |||
|
r5493 | def test__IPYTHON__(): | ||
# This shouldn't raise a NameError, that's all | ||||
__IPYTHON__ | ||||
|
r10636 | |||
class DummyRepr(object): | ||||
def __repr__(self): | ||||
return "DummyRepr" | ||||
def _repr_html_(self): | ||||
return "<b>dummy</b>" | ||||
def _repr_javascript_(self): | ||||
return "console.log('hi');", {'key': 'value'} | ||||
def test_user_variables(): | ||||
# enable all formatters | ||||
ip.display_formatter.active_types = ip.display_formatter.format_types | ||||
ip.user_ns['dummy'] = d = DummyRepr() | ||||
|
r10682 | keys = set(['dummy', 'doesnotexist']) | ||
|
r16570 | r = ip.user_expressions({ key:key for key in keys}) | ||
|
r10636 | |||
|
r10682 | nt.assert_equal(keys, set(r.keys())) | ||
|
r10636 | dummy = r['dummy'] | ||
|
r10682 | nt.assert_equal(set(['status', 'data', 'metadata']), set(dummy.keys())) | ||
|
r10636 | nt.assert_equal(dummy['status'], 'ok') | ||
data = dummy['data'] | ||||
metadata = dummy['metadata'] | ||||
nt.assert_equal(data.get('text/html'), d._repr_html_()) | ||||
js, jsmd = d._repr_javascript_() | ||||
nt.assert_equal(data.get('application/javascript'), js) | ||||
nt.assert_equal(metadata.get('application/javascript'), jsmd) | ||||
dne = r['doesnotexist'] | ||||
nt.assert_equal(dne['status'], 'error') | ||||
|
r16570 | nt.assert_equal(dne['ename'], 'NameError') | ||
|
r10636 | |||
# back to text only | ||||
ip.display_formatter.active_types = ['text/plain'] | ||||
def test_user_expression(): | ||||
# enable all formatters | ||||
ip.display_formatter.active_types = ip.display_formatter.format_types | ||||
query = { | ||||
'a' : '1 + 2', | ||||
'b' : '1/0', | ||||
} | ||||
r = ip.user_expressions(query) | ||||
import pprint | ||||
pprint.pprint(r) | ||||
|
r14839 | nt.assert_equal(set(r.keys()), set(query.keys())) | ||
|
r10636 | a = r['a'] | ||
|
r10682 | nt.assert_equal(set(['status', 'data', 'metadata']), set(a.keys())) | ||
|
r10636 | nt.assert_equal(a['status'], 'ok') | ||
data = a['data'] | ||||
metadata = a['metadata'] | ||||
nt.assert_equal(data.get('text/plain'), '3') | ||||
b = r['b'] | ||||
nt.assert_equal(b['status'], 'error') | ||||
nt.assert_equal(b['ename'], 'ZeroDivisionError') | ||||
# back to text only | ||||
ip.display_formatter.active_types = ['text/plain'] | ||||
|
r13525 | class TestSyntaxErrorTransformer(unittest.TestCase): | ||
"""Check that SyntaxError raised by an input transformer is handled by run_cell()""" | ||||
class SyntaxErrorTransformer(InputTransformer): | ||||
def push(self, line): | ||||
pos = line.find('syntaxerror') | ||||
if pos >= 0: | ||||
e = SyntaxError('input contains "syntaxerror"') | ||||
e.text = line | ||||
e.offset = pos + 1 | ||||
raise e | ||||
return line | ||||
def reset(self): | ||||
pass | ||||
def setUp(self): | ||||
self.transformer = TestSyntaxErrorTransformer.SyntaxErrorTransformer() | ||||
ip.input_splitter.python_line_transforms.append(self.transformer) | ||||
ip.input_transformer_manager.python_line_transforms.append(self.transformer) | ||||
def tearDown(self): | ||||
ip.input_splitter.python_line_transforms.remove(self.transformer) | ||||
ip.input_transformer_manager.python_line_transforms.remove(self.transformer) | ||||
def test_syntaxerror_input_transformer(self): | ||||
with tt.AssertPrints('1234'): | ||||
ip.run_cell('1234') | ||||
with tt.AssertPrints('SyntaxError: invalid syntax'): | ||||
ip.run_cell('1 2 3') # plain python syntax error | ||||
with tt.AssertPrints('SyntaxError: input contains "syntaxerror"'): | ||||
ip.run_cell('2345 # syntaxerror') # input transformer syntax error | ||||
with tt.AssertPrints('3456'): | ||||
ip.run_cell('3456') | ||||
|
r10636 | |||
|
r18548 | def test_warning_suppression(): | ||
ip.run_cell("import warnings") | ||||
try: | ||||
with tt.AssertPrints("UserWarning: asdf", channel="stderr"): | ||||
ip.run_cell("warnings.warn('asdf')") | ||||
# Here's the real test -- if we run that again, we should get the | ||||
# warning again. Traditionally, each warning was only issued once per | ||||
# IPython session (approximately), even if the user typed in new and | ||||
# different code that should have also triggered the warning, leading | ||||
# to much confusion. | ||||
with tt.AssertPrints("UserWarning: asdf", channel="stderr"): | ||||
ip.run_cell("warnings.warn('asdf')") | ||||
finally: | ||||
ip.run_cell("del warnings") | ||||