test_interactiveshell.py
1221 lines
| 39.1 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. | ||||
|
r24538 | import asyncio | ||
|
r8220 | import ast | ||
|
r5094 | import os | ||
|
r12774 | import signal | ||
|
r5094 | import shutil | ||
|
r6997 | import sys | ||
|
r5094 | import tempfile | ||
|
r3300 | import unittest | ||
|
r28098 | import pytest | ||
|
r23064 | from unittest import mock | ||
|
r5094 | from os.path import join | ||
|
r3822 | |||
|
r17799 | from IPython.core.error import InputRejected | ||
|
r13525 | from IPython.core.inputtransformer import InputTransformer | ||
|
r24538 | from IPython.core import interactiveshell | ||
|
r28165 | from IPython.core.oinspect import OInfo | ||
|
r17016 | from IPython.testing.decorators import ( | ||
|
r28785 | skipif, | ||
skip_win32, | ||||
onlyif_unicode_paths, | ||||
onlyif_cmds_exist, | ||||
skip_if_not_osx, | ||||
|
r17016 | ) | ||
|
r8085 | from IPython.testing import tools as tt | ||
|
r17016 | from IPython.utils.process import find_cmd | ||
|
r3300 | |||
#----------------------------------------------------------------------------- | ||||
|
r6997 | # Globals | ||
#----------------------------------------------------------------------------- | ||||
# This is used by every single test, no point repeating it ad nauseam | ||||
#----------------------------------------------------------------------------- | ||||
|
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) | ||||
|
r27850 | def test_open_standard_input_stream(self): | ||
res = ip.run_cell("open(0)") | ||||
self.assertIsInstance(res.error_in_exec, ValueError) | ||||
def test_open_standard_output_stream(self): | ||||
res = ip.run_cell("open(1)") | ||||
self.assertIsInstance(res.error_in_exec, ValueError) | ||||
def test_open_standard_error_stream(self): | ||||
res = ip.run_cell("open(2)") | ||||
self.assertIsInstance(res.error_in_exec, ValueError) | ||||
|
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;') | ||||
|
r28785 | |||
|
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') | ||
|
r28785 | |||
|
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) | ||||
|
r28785 | |||
|
r4281 | def test_gh_597(self): | ||
|
r4375 | """Pretty-printing lists of objects with non-ascii reprs may cause | ||
problems.""" | ||||
|
r4281 | class Spam(object): | ||
|
r25115 | def __repr__(self): | ||
return "\xe9"*50 | ||||
|
r4281 | import IPython.core.formatters | ||
f = IPython.core.formatters.PlainTextFormatter() | ||||
|
r28785 | f([Spam(), Spam()]) | ||
|
r5456 | |||
|
r4795 | def test_future_flags(self): | ||
"""Check that future flags are used for parsing code (gh-777)""" | ||||
|
r22975 | ip.run_cell('from __future__ import barry_as_FLUFL') | ||
|
r4795 | try: | ||
|
r22975 | ip.run_cell('prfunc_return_val = 1 <> 2') | ||
|
r4795 | 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 | |||
|
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])") | ||||
|
r28785 | |||
|
r13354 | from pickle import dumps | ||
|
r28785 | |||
|
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)) | ||||
|
r28785 | |||
|
r5456 | 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""" | ||||
|
r22192 | ip.set_custom_exc((IOError,), lambda etype,value,tb: 1/0) | ||
self.assertEqual(ip.custom_exceptions, (IOError,)) | ||||
with tt.AssertPrints("Custom TB Handler failed", channel='stderr'): | ||||
|
r4991 | ip.run_cell(u'raise IOError("foo")') | ||
|
r22192 | self.assertEqual(ip.custom_exceptions, ()) | ||
|
r4991 | |||
|
r4999 | def test_bad_custom_tb_return(self): | ||
"""Check that InteractiveShell is protected from bad return types in custom exception handlers""" | ||||
|
r22192 | ip.set_custom_exc((NameError,),lambda etype,value,tb, tb_offset=None: 1) | ||
self.assertEqual(ip.custom_exceptions, (NameError,)) | ||||
with tt.AssertPrints("Custom TB Handler failed", channel='stderr'): | ||||
|
r4999 | ip.run_cell(u'a=abracadabra') | ||
|
r22192 | self.assertEqual(ip.custom_exceptions, ()) | ||
|
r4999 | |||
|
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') | ||||
|
r28785 | |||
|
r23620 | self.assertEqual(ip.var_expand(u"grep x | awk '{print $1}'"), u"grep x | awk '{print $1}'") | ||
|
r5366 | |||
ip.user_ns['f'] = b'Ca\xc3\xb1o' | ||||
|
r5357 | # This should not raise any exception: | ||
ip.var_expand(u'echo $f') | ||||
|
r28785 | |||
|
r7331 | def test_var_expand_local(self): | ||
|
r7333 | """Test local variable expansion in !system and %magic calls""" | ||
# !system | ||||
|
r26901 | ip.run_cell( | ||
"def test():\n" | ||||
' lvar = "ttt"\n' | ||||
" ret = !echo {lvar}\n" | ||||
" return ret[0]\n" | ||||
) | ||||
res = ip.user_ns["test"]() | ||||
self.assertIn("ttt", res) | ||||
|
r7333 | # %magic | ||
|
r26901 | 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()") | ||||
self.assertIn("macro_var_expand_locals", ip.user_ns) | ||||
|
r8224 | def test_var_expand_self(self): | ||
"""Test variable expansion with the name 'self', which was failing. | ||||
|
r28785 | |||
|
r8224 | See https://github.com/ipython/ipython/issues/1878#issuecomment-7698218 | ||
""" | ||||
|
r26901 | ip.run_cell( | ||
"class cTest:\n" | ||||
' classvar="see me"\n' | ||||
" def test(self):\n" | ||||
" res = !echo Variable: {self.classvar}\n" | ||||
" return res[0]\n" | ||||
) | ||||
self.assertIn("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}") | ||||
|
r28785 | |||
|
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() | ||||
|
r23982 | all_mocks = [pre_explicit, pre_always, post_explicit, post_always] | ||
|
r28785 | |||
|
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) | ||
|
r28785 | |||
|
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 | ||||
|
r23997 | info, = pre_explicit.call_args[0] | ||
result, = post_explicit.call_args[0] | ||||
self.assertEqual(info, result.info) | ||||
|
r23982 | # check that post hooks are always called | ||
|
r23909 | [m.reset_mock() for m in all_mocks] | ||
ip.run_cell("syntax error") | ||||
assert pre_always.called | ||||
assert pre_explicit.called | ||||
|
r23982 | assert post_always.called | ||
assert post_explicit.called | ||||
|
r23997 | info, = pre_explicit.call_args[0] | ||
result, = post_explicit.call_args[0] | ||||
self.assertEqual(info, result.info) | ||||
|
r15597 | 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) | ||||
|
r28785 | |||
|
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) | ||||
|
r28785 | |||
|
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) | ||
|
r28785 | |||
|
r6802 | def test_silent_nodisplayhook(self): | ||
"""run_cell(silent=True) doesn't trigger displayhook""" | ||||
d = dict(called=False) | ||||
|
r28785 | |||
|
r6802 | trap = ip.display_trap | ||
save_hook = trap.hook | ||||
|
r28785 | |||
|
r6802 | def failing_hook(*args, **kwargs): | ||
d['called'] = True | ||||
|
r28785 | |||
|
r6802 | 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 | |||
|
r6997 | def test_ofind_line_magic(self): | ||
from IPython.core.magic import register_line_magic | ||||
|
r28785 | |||
|
r6997 | @register_line_magic | ||
def lmagic(line): | ||||
"A line magic" | ||||
# Get info on line magic | ||||
|
r27102 | lfind = ip._ofind("lmagic") | ||
|
r28165 | info = OInfo( | ||
|
r27102 | found=True, | ||
isalias=False, | ||||
ismagic=True, | ||||
namespace="IPython internal", | ||||
obj=lmagic, | ||||
parent=None, | ||||
) | ||||
|
r26901 | self.assertEqual(lfind, info) | ||
|
r28785 | |||
|
r6997 | def test_ofind_cell_magic(self): | ||
from IPython.core.magic import register_cell_magic | ||||
|
r28785 | |||
|
r6997 | @register_cell_magic | ||
def cmagic(line, cell): | ||||
"A cell magic" | ||||
# Get info on cell magic | ||||
|
r27102 | find = ip._ofind("cmagic") | ||
|
r28165 | info = OInfo( | ||
|
r27102 | found=True, | ||
isalias=False, | ||||
ismagic=True, | ||||
namespace="IPython internal", | ||||
obj=cmagic, | ||||
parent=None, | ||||
) | ||||
|
r26901 | self.assertEqual(find, info) | ||
|
r17024 | |||
def test_ofind_property_with_error(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
|
r27561 | raise NotImplementedError() # pragma: no cover | ||
|
r17024 | a = A() | ||
|
r28165 | found = ip._ofind("a.foo", [("locals", locals())]) | ||
info = OInfo( | ||||
found=True, | ||||
isalias=False, | ||||
ismagic=False, | ||||
namespace="locals", | ||||
obj=A.foo, | ||||
parent=a, | ||||
) | ||||
|
r26901 | self.assertEqual(found, info) | ||
|
r17024 | |||
def test_ofind_multiple_attribute_lookups(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
|
r27561 | raise NotImplementedError() # pragma: no cover | ||
|
r17024 | |||
a = A() | ||||
a.a = A() | ||||
a.a.a = A() | ||||
|
r28165 | found = ip._ofind("a.a.a.foo", [("locals", locals())]) | ||
info = OInfo( | ||||
found=True, | ||||
isalias=False, | ||||
ismagic=False, | ||||
namespace="locals", | ||||
obj=A.foo, | ||||
parent=a.a.a, | ||||
) | ||||
|
r26901 | self.assertEqual(found, info) | ||
|
r17024 | |||
def test_ofind_slotted_attributes(self): | ||||
class A(object): | ||||
__slots__ = ['foo'] | ||||
def __init__(self): | ||||
self.foo = 'bar' | ||||
a = A() | ||||
|
r28165 | found = ip._ofind("a.foo", [("locals", locals())]) | ||
info = OInfo( | ||||
found=True, | ||||
isalias=False, | ||||
ismagic=False, | ||||
namespace="locals", | ||||
obj=a.foo, | ||||
parent=a, | ||||
) | ||||
|
r26901 | self.assertEqual(found, info) | ||
|
r17024 | |||
|
r28165 | found = ip._ofind("a.bar", [("locals", locals())]) | ||
|
r28167 | expected = OInfo( | ||
|
r28165 | found=False, | ||
isalias=False, | ||||
ismagic=False, | ||||
namespace=None, | ||||
obj=None, | ||||
parent=a, | ||||
) | ||||
|
r28167 | assert found == expected | ||
|
r17024 | |||
def test_ofind_prefers_property_to_instance_level_attribute(self): | ||||
class A(object): | ||||
@property | ||||
def foo(self): | ||||
return 'bar' | ||||
a = A() | ||||
|
r26901 | a.__dict__["foo"] = "baz" | ||
self.assertEqual(a.foo, "bar") | ||||
found = ip._ofind("a.foo", [("locals", locals())]) | ||||
|
r28165 | self.assertIs(found.obj, A.foo) | ||
|
r17024 | |||
|
r22432 | def test_custom_syntaxerror_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((SyntaxError,), my_handler) | ||||
try: | ||||
ip.run_cell("1f") | ||||
# Check that this was called, and only once. | ||||
self.assertEqual(called, [SyntaxError]) | ||||
finally: | ||||
# Reset the custom exception hook | ||||
ip.set_custom_exc((), None) | ||||
|
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) | ||||
|
r28785 | |||
|
r7110 | 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) | ||||
|
r28785 | |||
|
r26276 | @mock.patch("builtins.print") | ||
|
r26275 | def test_showtraceback_with_surrogates(self, mocked_print): | ||
values = [] | ||||
|
r26277 | |||
|
r26276 | def mock_print_func(value, sep=" ", end="\n", file=sys.stdout, flush=False): | ||
|
r26275 | values.append(value) | ||
|
r26276 | if value == chr(0xD8FF): | ||
raise UnicodeEncodeError("utf-8", chr(0xD8FF), 0, 1, "") | ||||
|
r26275 | |||
# mock builtins.print | ||||
mocked_print.side_effect = mock_print_func | ||||
|
r26276 | # ip._showtraceback() is replaced in globalipapp.py. | ||
|
r26275 | # Call original method to test. | ||
|
r26276 | interactiveshell.InteractiveShell._showtraceback(ip, None, None, chr(0xD8FF)) | ||
|
r26277 | |||
|
r26275 | self.assertEqual(mocked_print.call_count, 2) | ||
|
r26276 | self.assertEqual(values, [chr(0xD8FF), "\\ud8ff"]) | ||
|
r26275 | |||
|
r17337 | def test_mktempfile(self): | ||
filename = ip.mktempfile() | ||||
# Check that we can open the file again on Windows | ||||
|
r27495 | with open(filename, "w", encoding="utf-8") as f: | ||
f.write("abc") | ||||
|
r17337 | |||
|
r27495 | filename = ip.mktempfile(data="blah") | ||
with open(filename, "r", encoding="utf-8") 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() | ||||
|
r22959 | self.assertEqual(msg, 'IPython.core.tests.test_interactiveshell.DerivedInterrupt: foo\n') | ||
|
r19114 | |||
|
r22504 | def test_inspect_text(self): | ||
ip.run_cell('a = 5') | ||||
text = ip.object_inspect_text('a') | ||||
|
r23044 | self.assertIsInstance(text, str) | ||
|
r22504 | |||
|
r23803 | def test_last_execution_result(self): | ||
""" Check that last execution result gets set correctly (GH-10702) """ | ||||
result = ip.run_cell('a = 5; a') | ||||
self.assertTrue(ip.last_execution_succeeded) | ||||
self.assertEqual(ip.last_execution_result.result, 5) | ||||
result = ip.run_cell('a = x_invalid_id_x') | ||||
self.assertFalse(ip.last_execution_succeeded) | ||||
self.assertFalse(ip.last_execution_result.success) | ||||
self.assertIsInstance(ip.last_execution_result.error_in_exec, NameError) | ||||
|
r24889 | def test_reset_aliasing(self): | ||
""" Check that standard posix aliases work after %reset. """ | ||||
if os.name != 'posix': | ||||
return | ||||
ip.reset() | ||||
for cmd in ('clear', 'more', 'less', 'man'): | ||||
res = ip.run_cell('%' + cmd) | ||||
self.assertEqual(res.success, True) | ||||
|
r22504 | |||
|
r28589 | @pytest.mark.skipif( | ||
sys.implementation.name == "pypy" | ||||
and ((7, 3, 13) < sys.implementation.version < (7, 3, 16)), | ||||
reason="Unicode issues with scandir on PyPy, see https://github.com/pypy/pypy/issues/4860", | ||||
) | ||||
|
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) | ||||
|
r27495 | with open( | ||
|
r27639 | join(self.TESTDIR, "åäötestscript.py"), "w", encoding="utf-8" | ||
|
r27495 | ) as sfile: | ||
|
r5094 | sfile.write("pass\n") | ||
|
r23045 | self.oldpath = os.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): | ||
|
r25113 | |||
def setUp(self): | ||||
self.system = ip.system_raw | ||||
|
r12774 | 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) | ||||
|
r28785 | |||
|
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) | ||||
|
r28785 | |||
|
r17016 | @onlyif_cmds_exist("csh") | ||
|
r27561 | def test_exit_code_signal_csh(self): # pragma: no cover | ||
SHELL = os.environ.get("SHELL", None) | ||||
os.environ["SHELL"] = find_cmd("csh") | ||||
|
r17016 | try: | ||
self.test_exit_code_signal() | ||||
finally: | ||||
if SHELL is not None: | ||||
os.environ['SHELL'] = SHELL | ||||
else: | ||||
del os.environ['SHELL'] | ||||
|
r12774 | |||
|
r24394 | |||
|
r25113 | class TestSystemRaw(ExitCodeChecks): | ||
def setUp(self): | ||||
super().setUp() | ||||
|
r25167 | self.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") | ||||
|
r27561 | except KeyboardInterrupt: # pragma: no cove | ||
|
r26529 | self.fail( | ||
"system call should intercept " | ||||
"keyboard interrupt from subprocess.call" | ||||
) | ||||
self.assertEqual(ip.user_ns["_exit_code"], -signal.SIGINT) | ||||
|
r28098 | |||
@pytest.mark.parametrize("magic_cmd", ["pip", "conda", "cd"]) | ||||
def test_magic_warnings(magic_cmd): | ||||
if sys.platform == "win32": | ||||
to_mock = "os.system" | ||||
expected_arg, expected_kwargs = magic_cmd, dict() | ||||
else: | ||||
to_mock = "subprocess.call" | ||||
expected_arg, expected_kwargs = magic_cmd, dict( | ||||
shell=True, executable=os.environ.get("SHELL", None) | ||||
) | ||||
with mock.patch(to_mock, return_value=0) as mock_sub: | ||||
with pytest.warns(Warning, match=r"You executed the system command"): | ||||
ip.system_raw(magic_cmd) | ||||
mock_sub.assert_called_once_with(expected_arg, **expected_kwargs) | ||||
|
r18732 | |||
|
r12774 | # TODO: Exit codes are currently ignored on Windows. | ||
|
r25113 | class TestSystemPipedExitCode(ExitCodeChecks): | ||
def setUp(self): | ||||
super().setUp() | ||||
|
r25167 | self.system = ip.system_piped | ||
|
r12774 | |||
|
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 | |||
|
r25113 | class TestModules(tt.TempFileMixin): | ||
|
r8085 | 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.""" | ||||
|
r24923 | |||
|
r8220 | def visit_Num(self, node): | ||
node.n = -node.n | ||||
return node | ||||
|
r24932 | def visit_Constant(self, node): | ||
if isinstance(node.value, int): | ||||
return self.visit_Num(node) | ||||
return node | ||||
|
r24923 | |||
|
r8220 | class TestAstTransform(unittest.TestCase): | ||
def setUp(self): | ||||
self.negator = Negator() | ||||
ip.ast_transformers.append(self.negator) | ||||
|
r28785 | |||
|
r8220 | def tearDown(self): | ||
ip.ast_transformers.remove(self.negator) | ||||
|
r27561 | |||
def test_non_int_const(self): | ||||
with tt.AssertPrints("hello"): | ||||
ip.run_cell('print("hello")') | ||||
|
r8220 | def test_run_cell(self): | ||
|
r27561 | with tt.AssertPrints("-34"): | ||
ip.run_cell("print(12 + 22)") | ||||
|
r8220 | # A named reference to a number shouldn't be transformed. | ||
|
r27561 | 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}) | ||||
|
r28785 | |||
|
r23497 | with tt.AssertPrints("std. dev. of"): | ||
|
r8485 | ip.run_line_magic("timeit", "-n1 f(1)") | ||
|
r21804 | self.assertEqual(called, {-1}) | ||
|
r8485 | called.clear() | ||
|
r22919 | |||
|
r23497 | with tt.AssertPrints("std. dev. of"): | ||
|
r8485 | ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)") | ||
|
r21804 | self.assertEqual(called, {-2, -3}) | ||
|
r28785 | |||
|
r8486 | def test_time(self): | ||
called = [] | ||||
def f(x): | ||||
called.append(x) | ||||
ip.push({'f':f}) | ||||
|
r28785 | |||
|
r8486 | # Test with an expression | ||
|
r9901 | with tt.AssertPrints("Wall time: "): | ||
|
r8486 | ip.run_line_magic("time", "f(5+9)") | ||
self.assertEqual(called, [-14]) | ||||
called[:] = [] | ||||
|
r28785 | |||
|
r8486 | # 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]) | ||||
|
r28785 | |||
|
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)") | ||||
|
r28785 | |||
|
r8488 | with tt.AssertPrints("9"): | ||
ip.run_cell("amacro") | ||||
with tt.AssertPrints("8"): | ||||
ip.run_cell("amacro") | ||||
|
r5493 | |||
|
r25925 | class TestMiscTransform(unittest.TestCase): | ||
def test_transform_only_once(self): | ||||
cleanup = 0 | ||||
line_t = 0 | ||||
def count_cleanup(lines): | ||||
nonlocal cleanup | ||||
cleanup += 1 | ||||
return lines | ||||
def count_line_t(lines): | ||||
nonlocal line_t | ||||
line_t += 1 | ||||
return lines | ||||
ip.input_transformer_manager.cleanup_transforms.append(count_cleanup) | ||||
ip.input_transformer_manager.line_transforms.append(count_line_t) | ||||
ip.run_cell('1') | ||||
assert cleanup == 1 | ||||
assert line_t == 1 | ||||
|
r8479 | class IntegerWrapper(ast.NodeTransformer): | ||
"""Wraps all integers in a call to Integer()""" | ||||
|
r24932 | |||
# for Python 3.7 and earlier | ||||
# for Python 3.7 and earlier | ||||
|
r8479 | 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 | |||
|
r24932 | # For Python 3.8+ | ||
def visit_Constant(self, node): | ||||
if isinstance(node.value, int): | ||||
return self.visit_Num(node) | ||||
return node | ||||
|
r24923 | |||
|
r8479 | class TestAstTransform2(unittest.TestCase): | ||
def setUp(self): | ||||
self.intwrapper = IntegerWrapper() | ||||
ip.ast_transformers.append(self.intwrapper) | ||||
|
r28785 | |||
|
r8479 | self.calls = [] | ||
def Integer(*args): | ||||
self.calls.append(args) | ||||
|
r8485 | return args | ||
|
r8479 | ip.push({"Integer": Integer}) | ||
|
r28785 | |||
|
r8479 | def tearDown(self): | ||
ip.ast_transformers.remove(self.intwrapper) | ||||
del ip.user_ns['Integer'] | ||||
|
r28785 | |||
|
r8479 | def test_run_cell(self): | ||
ip.run_cell("n = 2") | ||||
self.assertEqual(self.calls, [(2,)]) | ||||
|
r28785 | |||
|
r8508 | # This shouldn't throw an error | ||
ip.run_cell("o = 2.0") | ||||
self.assertEqual(ip.user_ns['o'], 2.0) | ||||
|
r27561 | |||
def test_run_cell_non_int(self): | ||||
ip.run_cell("n = 'a'") | ||||
assert self.calls == [] | ||||
|
r8485 | def test_timeit(self): | ||
called = set() | ||||
def f(x): | ||||
called.add(x) | ||||
ip.push({'f':f}) | ||||
|
r22919 | |||
|
r23497 | with tt.AssertPrints("std. dev. of"): | ||
|
r8485 | ip.run_line_magic("timeit", "-n1 f(1)") | ||
|
r21804 | self.assertEqual(called, {(1,)}) | ||
|
r8485 | called.clear() | ||
|
r22919 | |||
|
r23497 | with tt.AssertPrints("std. dev. of"): | ||
|
r8485 | ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)") | ||
|
r21804 | self.assertEqual(called, {(2,), (3,)}) | ||
|
r8479 | |||
|
r8221 | class ErrorTransformer(ast.NodeTransformer): | ||
"""Throws an error when it sees a number.""" | ||||
|
r24932 | |||
def visit_Constant(self, node): | ||||
if isinstance(node.value, int): | ||||
|
r27561 | raise ValueError("test") | ||
|
r24932 | return node | ||
|
r24923 | |||
|
r8221 | class TestAstTransformError(unittest.TestCase): | ||
def test_unregistering(self): | ||||
err_transformer = ErrorTransformer() | ||||
ip.ast_transformers.append(err_transformer) | ||||
|
r28785 | |||
|
r25115 | with self.assertWarnsRegex(UserWarning, "It will be unregistered"): | ||
|
r8221 | ip.run_cell("1 + 2") | ||
|
r28785 | |||
|
r8221 | # This should have been removed. | ||
|
r26901 | self.assertNotIn(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. | ||||
""" | ||||
|
r28785 | |||
|
r24924 | def visit_Constant(self, node): | ||
if isinstance(node.value, str): | ||||
raise InputRejected("test") | ||||
|
r24925 | return node | ||
|
r24924 | |||
|
r17798 | |||
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" | ||||
|
r28785 | |||
|
r10636 | def _repr_html_(self): | ||
return "<b>dummy</b>" | ||||
|
r28785 | |||
|
r10636 | def _repr_javascript_(self): | ||
return "console.log('hi');", {'key': 'value'} | ||||
|
r28785 | |||
|
r10636 | |||
def test_user_variables(): | ||||
# enable all formatters | ||||
ip.display_formatter.active_types = ip.display_formatter.format_types | ||||
|
r28785 | |||
|
r10636 | ip.user_ns['dummy'] = d = DummyRepr() | ||
|
r21804 | keys = {'dummy', 'doesnotexist'} | ||
|
r16570 | r = ip.user_expressions({ key:key for key in keys}) | ||
|
r10636 | |||
|
r26901 | assert keys == set(r.keys()) | ||
dummy = r["dummy"] | ||||
assert {"status", "data", "metadata"} == set(dummy.keys()) | ||||
assert dummy["status"] == "ok" | ||||
data = dummy["data"] | ||||
metadata = dummy["metadata"] | ||||
assert data.get("text/html") == d._repr_html_() | ||||
|
r10636 | js, jsmd = d._repr_javascript_() | ||
|
r26901 | assert data.get("application/javascript") == js | ||
assert metadata.get("application/javascript") == jsmd | ||||
dne = r["doesnotexist"] | ||||
assert dne["status"] == "error" | ||||
assert dne["ename"] == "NameError" | ||||
|
r10636 | # back to text only | ||
ip.display_formatter.active_types = ['text/plain'] | ||||
|
r28785 | |||
|
r10636 | 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) | ||||
|
r26901 | assert set(r.keys()) == set(query.keys()) | ||
a = r["a"] | ||||
assert {"status", "data", "metadata"} == set(a.keys()) | ||||
assert a["status"] == "ok" | ||||
data = a["data"] | ||||
metadata = a["metadata"] | ||||
assert data.get("text/plain") == "3" | ||||
b = r["b"] | ||||
assert b["status"] == "error" | ||||
assert b["ename"] == "ZeroDivisionError" | ||||
|
r10636 | # 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()""" | ||||
|
r24164 | @staticmethod | ||
def transformer(lines): | ||||
for line in lines: | ||||
|
r13525 | pos = line.find('syntaxerror') | ||
if pos >= 0: | ||||
e = SyntaxError('input contains "syntaxerror"') | ||||
e.text = line | ||||
e.offset = pos + 1 | ||||
raise e | ||||
|
r24164 | return lines | ||
|
r13525 | |||
def setUp(self): | ||||
|
r24404 | ip.input_transformers_post.append(self.transformer) | ||
|
r13525 | |||
def tearDown(self): | ||||
|
r24404 | ip.input_transformers_post.remove(self.transformer) | ||
|
r13525 | |||
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') | ||||
|
r25167 | class TestWarningSuppression(unittest.TestCase): | ||
|
r25115 | def test_warning_suppression(self): | ||
ip.run_cell("import warnings") | ||||
try: | ||||
with self.assertWarnsRegex(UserWarning, "asdf"): | ||||
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 self.assertWarnsRegex(UserWarning, "asdf"): | ||||
ip.run_cell("warnings.warn('asdf')") | ||||
finally: | ||||
ip.run_cell("del warnings") | ||||
|
r21392 | |||
|
r25115 | def test_deprecation_warning(self): | ||
ip.run_cell(""" | ||||
|
r21392 | import warnings | ||
def wrn(): | ||||
warnings.warn( | ||||
"I AM A WARNING", | ||||
DeprecationWarning | ||||
) | ||||
""") | ||||
|
r25115 | try: | ||
with self.assertWarnsRegex(DeprecationWarning, "I AM A WARNING"): | ||||
ip.run_cell("wrn()") | ||||
finally: | ||||
ip.run_cell("del warnings") | ||||
ip.run_cell("del wrn") | ||||
|
r21393 | |||
class TestImportNoDeprecate(tt.TempFileMixin): | ||||
|
r25113 | def setUp(self): | ||
|
r21393 | """Make a valid python temp file.""" | ||
self.mktmp(""" | ||||
import warnings | ||||
def wrn(): | ||||
warnings.warn( | ||||
"I AM A WARNING", | ||||
DeprecationWarning | ||||
) | ||||
""") | ||||
|
r25113 | super().setUp() | ||
|
r21393 | |||
def test_no_dep(self): | ||||
""" | ||||
No deprecation warning should be raised from imported functions | ||||
""" | ||||
ip.run_cell("from {} import wrn".format(self.fname)) | ||||
with tt.AssertNotPrints("I AM A WARNING"): | ||||
ip.run_cell("wrn()") | ||||
ip.run_cell("del wrn") | ||||
|
r24135 | |||
def test_custom_exc_count(): | ||||
hook = mock.Mock(return_value=None) | ||||
ip.set_custom_exc((SyntaxError,), hook) | ||||
before = ip.execution_count | ||||
ip.run_cell("def foo()", store_history=True) | ||||
|
r24136 | # restore default excepthook | ||
ip.set_custom_exc((), None) | ||||
|
r26901 | assert hook.call_count == 1 | ||
assert ip.execution_count == before + 1 | ||||
|
r24538 | |||
def test_run_cell_async(): | ||||
ip.run_cell("import asyncio") | ||||
coro = ip.run_cell_async("await asyncio.sleep(0.01)\n5") | ||||
assert asyncio.iscoroutine(coro) | ||||
|
r27387 | loop = asyncio.new_event_loop() | ||
|
r24538 | result = loop.run_until_complete(coro) | ||
assert isinstance(result, interactiveshell.ExecutionResult) | ||||
assert result.result == 5 | ||||
|
r27192 | def test_run_cell_await(): | ||
ip.run_cell("import asyncio") | ||||
result = ip.run_cell("await asyncio.sleep(0.01); 10") | ||||
assert ip.user_ns["_"] == 10 | ||||
def test_run_cell_asyncio_run(): | ||||
ip.run_cell("import asyncio") | ||||
result = ip.run_cell("await asyncio.sleep(0.01); 1") | ||||
assert ip.user_ns["_"] == 1 | ||||
result = ip.run_cell("asyncio.run(asyncio.sleep(0.01)); 2") | ||||
assert ip.user_ns["_"] == 2 | ||||
result = ip.run_cell("await asyncio.sleep(0.01); 3") | ||||
assert ip.user_ns["_"] == 3 | ||||
|
r24538 | def test_should_run_async(): | ||
|
r28098 | assert not ip.should_run_async("a = 5", transformed_cell="a = 5") | ||
assert ip.should_run_async("await x", transformed_cell="await x") | ||||
assert ip.should_run_async( | ||||
"import asyncio; await asyncio.sleep(1)", | ||||
transformed_cell="import asyncio; await asyncio.sleep(1)", | ||||
) | ||||
|
r25490 | |||
def test_set_custom_completer(): | ||||
num_completers = len(ip.Completer.matchers) | ||||
def foo(*args, **kwargs): | ||||
return "I'm a completer!" | ||||
ip.set_custom_completer(foo, 0) | ||||
# check that we've really added a new completer | ||||
assert len(ip.Completer.matchers) == num_completers + 1 | ||||
# check that the first completer is the function we defined | ||||
assert ip.Completer.matchers[0]() == "I'm a completer!" | ||||
# clean up | ||||
ip.Completer.custom_matchers.pop() | ||||
|
r28095 | |||
|
r28096 | class TestShowTracebackAttack(unittest.TestCase): | ||
|
r28095 | """Test that the interactive shell is resilient against the client attack of | ||
manipulating the showtracebacks method. These attacks shouldn't result in an | ||||
unhandled exception in the kernel.""" | ||||
|
r28096 | def setUp(self): | ||
self.orig_showtraceback = interactiveshell.InteractiveShell.showtraceback | ||||
def tearDown(self): | ||||
interactiveshell.InteractiveShell.showtraceback = self.orig_showtraceback | ||||
|
r28095 | def test_set_show_tracebacks_none(self): | ||
"""Test the case of the client setting showtracebacks to None""" | ||||
result = ip.run_cell( | ||||
""" | ||||
import IPython.core.interactiveshell | ||||
IPython.core.interactiveshell.InteractiveShell.showtraceback = None | ||||
assert False, "This should not raise an exception" | ||||
""" | ||||
) | ||||
print(result) | ||||
assert result.result is None | ||||
assert isinstance(result.error_in_exec, TypeError) | ||||
assert str(result.error_in_exec) == "'NoneType' object is not callable" | ||||
def test_set_show_tracebacks_noop(self): | ||||
"""Test the case of the client setting showtracebacks to a no op lambda""" | ||||
result = ip.run_cell( | ||||
""" | ||||
import IPython.core.interactiveshell | ||||
IPython.core.interactiveshell.InteractiveShell.showtraceback = lambda *args, **kwargs: None | ||||
assert False, "This should not raise an exception" | ||||
""" | ||||
) | ||||
print(result) | ||||
assert result.result is None | ||||
assert isinstance(result.error_in_exec, AssertionError) | ||||
assert str(result.error_in_exec) == "This should not raise an exception" | ||||
|
r28785 | |||
@skip_if_not_osx | ||||
def test_enable_gui_osx(): | ||||
simple_prompt = ip.simple_prompt | ||||
ip.simple_prompt = False | ||||
ip.enable_gui("osx") | ||||
assert ip.active_eventloop == "osx" | ||||
ip.enable_gui() | ||||
# The following line fails for IPython <= 8.25.0 | ||||
ip.enable_gui("macosx") | ||||
assert ip.active_eventloop == "osx" | ||||
ip.enable_gui() | ||||
ip.simple_prompt = simple_prompt | ||||