1  # -*- coding: utf-8 -*-
   2  """Tests for the key interactiveshell module.
   3  
   4  Historically the main classes in interactiveshell have been under-tested.  This
   5  module should grow as many single-method tests as possible to trap many of the
   6  recurring bugs we seem to encounter with high-level interaction.
   7  """
   8  
   9  # Copyright (c) IPython Development Team.
  10  # Distributed under the terms of the Modified BSD License.
  11  
  12  import asyncio
  13  import ast
  14  import os
  15  import signal
  16  import shutil
  17  import sys
  18  import tempfile
  19  import unittest
  20  import pytest
  21  from unittest import mock
  22  
  23  from os.path import join
  24  
  25  from IPython.core.error import InputRejected
  26  from IPython.core.inputtransformer import InputTransformer
  27  from IPython.core import interactiveshell
  28  from IPython.core.oinspect import OInfo
  29  from IPython.testing.decorators import (
  30      skipif,
  31      skip_win32,
  32      onlyif_unicode_paths,
  33      onlyif_cmds_exist,
  34      skip_if_not_osx,
  35  )
  36  from IPython.testing import tools as tt
  37  from IPython.utils.process import find_cmd
  38  
  39  #-----------------------------------------------------------------------------
  40  # Globals
  41  #-----------------------------------------------------------------------------
  42  # This is used by every single test, no point repeating it ad nauseam
  43  
  44  #-----------------------------------------------------------------------------
  45  # Tests
  46  #-----------------------------------------------------------------------------
  47  
  48  class DerivedInterrupt(KeyboardInterrupt):
  49      pass
  50  
  51  class InteractiveShellTestCase(unittest.TestCase):
  52      def test_naked_string_cells(self):
  53          """Test that cells with only naked strings are fully executed"""
  54          # First, single-line inputs
  55          ip.run_cell('"a"\n')
  56          self.assertEqual(ip.user_ns['_'], 'a')
  57          # And also multi-line cells
  58          ip.run_cell('"""a\nb"""\n')
  59          self.assertEqual(ip.user_ns['_'], 'a\nb')
  60  
  61      def test_run_empty_cell(self):
  62          """Just make sure we don't get a horrible error with a blank
  63          cell of input. Yes, I did overlook that."""
  64          old_xc = ip.execution_count
  65          res = ip.run_cell('')
  66          self.assertEqual(ip.execution_count, old_xc)
  67          self.assertEqual(res.execution_count, None)
  68  
  69      def test_run_cell_multiline(self):
  70          """Multi-block, multi-line cells must execute correctly.
  71          """
  72          src = '\n'.join(["x=1",
  73                           "y=2",
  74                           "if 1:",
  75                           "    x += 1",
  76                           "    y += 1",])
  77          res = ip.run_cell(src)
  78          self.assertEqual(ip.user_ns['x'], 2)
  79          self.assertEqual(ip.user_ns['y'], 3)
  80          self.assertEqual(res.success, True)
  81          self.assertEqual(res.result, None)
  82  
  83      def test_multiline_string_cells(self):
  84          "Code sprinkled with multiline strings should execute (GH-306)"
  85          ip.run_cell('tmp=0')
  86          self.assertEqual(ip.user_ns['tmp'], 0)
  87          res = ip.run_cell('tmp=1;"""a\nb"""\n')
  88          self.assertEqual(ip.user_ns['tmp'], 1)
  89          self.assertEqual(res.success, True)
  90          self.assertEqual(res.result, "a\nb")
  91  
  92      def test_dont_cache_with_semicolon(self):
  93          "Ending a line with semicolon should not cache the returned object (GH-307)"
  94          oldlen = len(ip.user_ns['Out'])
  95          for cell in ['1;', '1;1;']:
  96              res = ip.run_cell(cell, store_history=True)
  97              newlen = len(ip.user_ns['Out'])
  98              self.assertEqual(oldlen, newlen)
  99              self.assertIsNone(res.result)
 100          i = 0
 101          #also test the default caching behavior
 102          for cell in ['1', '1;1']:
 103              ip.run_cell(cell, store_history=True)
 104              newlen = len(ip.user_ns['Out'])
 105              i += 1
 106              self.assertEqual(oldlen+i, newlen)
 107  
 108      def test_syntax_error(self):
 109          res = ip.run_cell("raise = 3")
 110          self.assertIsInstance(res.error_before_exec, SyntaxError)
 111  
 112      def test_open_standard_input_stream(self):
 113          res = ip.run_cell("open(0)")
 114          self.assertIsInstance(res.error_in_exec, ValueError)
 115  
 116      def test_open_standard_output_stream(self):
 117          res = ip.run_cell("open(1)")
 118          self.assertIsInstance(res.error_in_exec, ValueError)
 119  
 120      def test_open_standard_error_stream(self):
 121          res = ip.run_cell("open(2)")
 122          self.assertIsInstance(res.error_in_exec, ValueError)
 123  
 124      def test_In_variable(self):
 125          "Verify that In variable grows with user input (GH-284)"
 126          oldlen = len(ip.user_ns['In'])
 127          ip.run_cell('1;', store_history=True)
 128          newlen = len(ip.user_ns['In'])
 129          self.assertEqual(oldlen+1, newlen)
 130          self.assertEqual(ip.user_ns['In'][-1],'1;')
 131  
 132      def test_magic_names_in_string(self):
 133          ip.run_cell('a = """\n%exit\n"""')
 134          self.assertEqual(ip.user_ns['a'], '\n%exit\n')
 135  
 136      def test_trailing_newline(self):
 137          """test that running !(command) does not raise a SyntaxError"""
 138          ip.run_cell('!(true)\n', False)
 139          ip.run_cell('!(true)\n\n\n', False)
 140  
 141      def test_gh_597(self):
 142          """Pretty-printing lists of objects with non-ascii reprs may cause
 143          problems."""
 144          class Spam(object):
 145              def __repr__(self):
 146                  return "\xe9"*50
 147          import IPython.core.formatters
 148          f = IPython.core.formatters.PlainTextFormatter()
 149          f([Spam(), Spam()])
 150  
 151      def test_future_flags(self):
 152          """Check that future flags are used for parsing code (gh-777)"""
 153          ip.run_cell('from __future__ import barry_as_FLUFL')
 154          try:
 155              ip.run_cell('prfunc_return_val = 1 <> 2')
 156              assert 'prfunc_return_val' in ip.user_ns
 157          finally:
 158              # Reset compiler flags so we don't mess up other tests.
 159              ip.compile.reset_compiler_flags()
 160  
 161      def test_can_pickle(self):
 162          "Can we pickle objects defined interactively (GH-29)"
 163          ip = get_ipython()
 164          ip.reset()
 165          ip.run_cell(("class Mylist(list):\n"
 166                       "    def __init__(self,x=[]):\n"
 167                       "        list.__init__(self,x)"))
 168          ip.run_cell("w=Mylist([1,2,3])")
 169  
 170          from pickle import dumps
 171  
 172          # We need to swap in our main module - this is only necessary
 173          # inside the test framework, because IPython puts the interactive module
 174          # in place (but the test framework undoes this).
 175          _main = sys.modules['__main__']
 176          sys.modules['__main__'] = ip.user_module
 177          try:
 178              res = dumps(ip.user_ns["w"])
 179          finally:
 180              sys.modules['__main__'] = _main
 181          self.assertTrue(isinstance(res, bytes))
 182  
 183      def test_global_ns(self):
 184          "Code in functions must be able to access variables outside them."
 185          ip = get_ipython()
 186          ip.run_cell("a = 10")
 187          ip.run_cell(("def f(x):\n"
 188                       "    return x + a"))
 189          ip.run_cell("b = f(12)")
 190          self.assertEqual(ip.user_ns["b"], 22)
 191  
 192      def test_bad_custom_tb(self):
 193          """Check that InteractiveShell is protected from bad custom exception handlers"""
 194          ip.set_custom_exc((IOError,), lambda etype,value,tb: 1/0)
 195          self.assertEqual(ip.custom_exceptions, (IOError,))
 196          with tt.AssertPrints("Custom TB Handler failed", channel='stderr'):
 197              ip.run_cell(u'raise IOError("foo")')
 198          self.assertEqual(ip.custom_exceptions, ())
 199  
 200      def test_bad_custom_tb_return(self):
 201          """Check that InteractiveShell is protected from bad return types in custom exception handlers"""
 202          ip.set_custom_exc((NameError,),lambda etype,value,tb, tb_offset=None: 1)
 203          self.assertEqual(ip.custom_exceptions, (NameError,))
 204          with tt.AssertPrints("Custom TB Handler failed", channel='stderr'):
 205              ip.run_cell(u'a=abracadabra')
 206          self.assertEqual(ip.custom_exceptions, ())
 207  
 208      def test_drop_by_id(self):
 209          myvars = {"a":object(), "b":object(), "c": object()}
 210          ip.push(myvars, interactive=False)
 211          for name in myvars:
 212              assert name in ip.user_ns, name
 213              assert name in ip.user_ns_hidden, name
 214          ip.user_ns['b'] = 12
 215          ip.drop_by_id(myvars)
 216          for name in ["a", "c"]:
 217              assert name not in ip.user_ns, name
 218              assert name not in ip.user_ns_hidden, name
 219          assert ip.user_ns['b'] == 12
 220          ip.reset()
 221  
 222      def test_var_expand(self):
 223          ip.user_ns['f'] = u'Ca\xf1o'
 224          self.assertEqual(ip.var_expand(u'echo $f'), u'echo Ca\xf1o')
 225          self.assertEqual(ip.var_expand(u'echo {f}'), u'echo Ca\xf1o')
 226          self.assertEqual(ip.var_expand(u'echo {f[:-1]}'), u'echo Ca\xf1')
 227          self.assertEqual(ip.var_expand(u'echo {1*2}'), u'echo 2')
 228  
 229          self.assertEqual(ip.var_expand(u"grep x | awk '{print $1}'"), u"grep x | awk '{print $1}'")
 230  
 231          ip.user_ns['f'] = b'Ca\xc3\xb1o'
 232          # This should not raise any exception:
 233          ip.var_expand(u'echo $f')
 234  
 235      def test_var_expand_local(self):
 236          """Test local variable expansion in !system and %magic calls"""
 237          # !system
 238          ip.run_cell(
 239              "def test():\n"
 240              '    lvar = "ttt"\n'
 241              "    ret = !echo {lvar}\n"
 242              "    return ret[0]\n"
 243          )
 244          res = ip.user_ns["test"]()
 245          self.assertIn("ttt", res)
 246  
 247          # %magic
 248          ip.run_cell(
 249              "def makemacro():\n"
 250              '    macroname = "macro_var_expand_locals"\n'
 251              "    %macro {macroname} codestr\n"
 252          )
 253          ip.user_ns["codestr"] = "str(12)"
 254          ip.run_cell("makemacro()")
 255          self.assertIn("macro_var_expand_locals", ip.user_ns)
 256  
 257      def test_var_expand_self(self):
 258          """Test variable expansion with the name 'self', which was failing.
 259  
 260          See https://github.com/ipython/ipython/issues/1878#issuecomment-7698218
 261          """
 262          ip.run_cell(
 263              "class cTest:\n"
 264              '  classvar="see me"\n'
 265              "  def test(self):\n"
 266              "    res = !echo Variable: {self.classvar}\n"
 267              "    return res[0]\n"
 268          )
 269          self.assertIn("see me", ip.user_ns["cTest"]().test())
 270  
 271      def test_bad_var_expand(self):
 272          """var_expand on invalid formats shouldn't raise"""
 273          # SyntaxError
 274          self.assertEqual(ip.var_expand(u"{'a':5}"), u"{'a':5}")
 275          # NameError
 276          self.assertEqual(ip.var_expand(u"{asdf}"), u"{asdf}")
 277          # ZeroDivisionError
 278          self.assertEqual(ip.var_expand(u"{1/0}"), u"{1/0}")
 279  
 280      def test_silent_postexec(self):
 281          """run_cell(silent=True) doesn't invoke pre/post_run_cell callbacks"""
 282          pre_explicit = mock.Mock()
 283          pre_always = mock.Mock()
 284          post_explicit = mock.Mock()
 285          post_always = mock.Mock()
 286          all_mocks = [pre_explicit, pre_always, post_explicit, post_always]
 287  
 288          ip.events.register('pre_run_cell', pre_explicit)
 289          ip.events.register('pre_execute', pre_always)
 290          ip.events.register('post_run_cell', post_explicit)
 291          ip.events.register('post_execute', post_always)
 292  
 293          try:
 294              ip.run_cell("1", silent=True)
 295              assert pre_always.called
 296              assert not pre_explicit.called
 297              assert post_always.called
 298              assert not post_explicit.called
 299              # double-check that non-silent exec did what we expected
 300              # silent to avoid
 301              ip.run_cell("1")
 302              assert pre_explicit.called
 303              assert post_explicit.called
 304              info, = pre_explicit.call_args[0]
 305              result, = post_explicit.call_args[0]
 306              self.assertEqual(info, result.info)
 307              # check that post hooks are always called
 308              [m.reset_mock() for m in all_mocks]
 309              ip.run_cell("syntax error")
 310              assert pre_always.called
 311              assert pre_explicit.called
 312              assert post_always.called
 313              assert post_explicit.called
 314              info, = pre_explicit.call_args[0]
 315              result, = post_explicit.call_args[0]
 316              self.assertEqual(info, result.info)
 317          finally:
 318              # remove post-exec
 319              ip.events.unregister('pre_run_cell', pre_explicit)
 320              ip.events.unregister('pre_execute', pre_always)
 321              ip.events.unregister('post_run_cell', post_explicit)
 322              ip.events.unregister('post_execute', post_always)
 323  
 324      def test_silent_noadvance(self):
 325          """run_cell(silent=True) doesn't advance execution_count"""
 326          ec = ip.execution_count
 327          # silent should force store_history=False
 328          ip.run_cell("1", store_history=True, silent=True)
 329  
 330          self.assertEqual(ec, ip.execution_count)
 331          # double-check that non-silent exec did what we expected
 332          # silent to avoid
 333          ip.run_cell("1", store_history=True)
 334          self.assertEqual(ec+1, ip.execution_count)
 335  
 336      def test_silent_nodisplayhook(self):
 337          """run_cell(silent=True) doesn't trigger displayhook"""
 338          d = dict(called=False)
 339  
 340          trap = ip.display_trap
 341          save_hook = trap.hook
 342  
 343          def failing_hook(*args, **kwargs):
 344              d['called'] = True
 345  
 346          try:
 347              trap.hook = failing_hook
 348              res = ip.run_cell("1", silent=True)
 349              self.assertFalse(d['called'])
 350              self.assertIsNone(res.result)
 351              # double-check that non-silent exec did what we expected
 352              # silent to avoid
 353              ip.run_cell("1")
 354              self.assertTrue(d['called'])
 355          finally:
 356              trap.hook = save_hook
 357  
 358      def test_ofind_line_magic(self):
 359          from IPython.core.magic import register_line_magic
 360  
 361          @register_line_magic
 362          def lmagic(line):
 363              "A line magic"
 364  
 365          # Get info on line magic
 366          lfind = ip._ofind("lmagic")
 367          info = OInfo(
 368              found=True,
 369              isalias=False,
 370              ismagic=True,
 371              namespace="IPython internal",
 372              obj=lmagic,
 373              parent=None,
 374          )
 375          self.assertEqual(lfind, info)
 376  
 377      def test_ofind_cell_magic(self):
 378          from IPython.core.magic import register_cell_magic
 379  
 380          @register_cell_magic
 381          def cmagic(line, cell):
 382              "A cell magic"
 383  
 384          # Get info on cell magic
 385          find = ip._ofind("cmagic")
 386          info = OInfo(
 387              found=True,
 388              isalias=False,
 389              ismagic=True,
 390              namespace="IPython internal",
 391              obj=cmagic,
 392              parent=None,
 393          )
 394          self.assertEqual(find, info)
 395  
 396      def test_ofind_property_with_error(self):
 397          class A(object):
 398              @property
 399              def foo(self):
 400                  raise NotImplementedError()  # pragma: no cover
 401  
 402          a = A()
 403  
 404          found = ip._ofind("a.foo", [("locals", locals())])
 405          info = OInfo(
 406              found=True,
 407              isalias=False,
 408              ismagic=False,
 409              namespace="locals",
 410              obj=A.foo,
 411              parent=a,
 412          )
 413          self.assertEqual(found, info)
 414  
 415      def test_ofind_multiple_attribute_lookups(self):
 416          class A(object):
 417              @property
 418              def foo(self):
 419                  raise NotImplementedError()  # pragma: no cover
 420  
 421          a = A()
 422          a.a = A()
 423          a.a.a = A()
 424  
 425          found = ip._ofind("a.a.a.foo", [("locals", locals())])
 426          info = OInfo(
 427              found=True,
 428              isalias=False,
 429              ismagic=False,
 430              namespace="locals",
 431              obj=A.foo,
 432              parent=a.a.a,
 433          )
 434          self.assertEqual(found, info)
 435  
 436      def test_ofind_slotted_attributes(self):
 437          class A(object):
 438              __slots__ = ['foo']
 439              def __init__(self):
 440                  self.foo = 'bar'
 441  
 442          a = A()
 443          found = ip._ofind("a.foo", [("locals", locals())])
 444          info = OInfo(
 445              found=True,
 446              isalias=False,
 447              ismagic=False,
 448              namespace="locals",
 449              obj=a.foo,
 450              parent=a,
 451          )
 452          self.assertEqual(found, info)
 453  
 454          found = ip._ofind("a.bar", [("locals", locals())])
 455          expected = OInfo(
 456              found=False,
 457              isalias=False,
 458              ismagic=False,
 459              namespace=None,
 460              obj=None,
 461              parent=a,
 462          )
 463          assert found == expected
 464  
 465      def test_ofind_prefers_property_to_instance_level_attribute(self):
 466          class A(object):
 467              @property
 468              def foo(self):
 469                  return 'bar'
 470          a = A()
 471          a.__dict__["foo"] = "baz"
 472          self.assertEqual(a.foo, "bar")
 473          found = ip._ofind("a.foo", [("locals", locals())])
 474          self.assertIs(found.obj, A.foo)
 475  
 476      def test_custom_syntaxerror_exception(self):
 477          called = []
 478          def my_handler(shell, etype, value, tb, tb_offset=None):
 479              called.append(etype)
 480              shell.showtraceback((etype, value, tb), tb_offset=tb_offset)
 481  
 482          ip.set_custom_exc((SyntaxError,), my_handler)
 483          try:
 484              ip.run_cell("1f")
 485              # Check that this was called, and only once.
 486              self.assertEqual(called, [SyntaxError])
 487          finally:
 488              # Reset the custom exception hook
 489              ip.set_custom_exc((), None)
 490  
 491      def test_custom_exception(self):
 492          called = []
 493          def my_handler(shell, etype, value, tb, tb_offset=None):
 494              called.append(etype)
 495              shell.showtraceback((etype, value, tb), tb_offset=tb_offset)
 496  
 497          ip.set_custom_exc((ValueError,), my_handler)
 498          try:
 499              res = ip.run_cell("raise ValueError('test')")
 500              # Check that this was called, and only once.
 501              self.assertEqual(called, [ValueError])
 502              # Check that the error is on the result object
 503              self.assertIsInstance(res.error_in_exec, ValueError)
 504          finally:
 505              # Reset the custom exception hook
 506              ip.set_custom_exc((), None)
 507  
 508      @mock.patch("builtins.print")
 509      def test_showtraceback_with_surrogates(self, mocked_print):
 510          values = []
 511  
 512          def mock_print_func(value, sep=" ", end="\n", file=sys.stdout, flush=False):
 513              values.append(value)
 514              if value == chr(0xD8FF):
 515                  raise UnicodeEncodeError("utf-8", chr(0xD8FF), 0, 1, "")
 516  
 517          # mock builtins.print
 518          mocked_print.side_effect = mock_print_func
 519  
 520          # ip._showtraceback() is replaced in globalipapp.py.
 521          # Call original method to test.
 522          interactiveshell.InteractiveShell._showtraceback(ip, None, None, chr(0xD8FF))
 523  
 524          self.assertEqual(mocked_print.call_count, 2)
 525          self.assertEqual(values, [chr(0xD8FF), "\\ud8ff"])
 526  
 527      def test_mktempfile(self):
 528          filename = ip.mktempfile()
 529          # Check that we can open the file again on Windows
 530          with open(filename, "w", encoding="utf-8") as f:
 531              f.write("abc")
 532  
 533          filename = ip.mktempfile(data="blah")
 534          with open(filename, "r", encoding="utf-8") as f:
 535              self.assertEqual(f.read(), "blah")
 536  
 537      def test_new_main_mod(self):
 538          # Smoketest to check that this accepts a unicode module name
 539          name = u'jiefmw'
 540          mod = ip.new_main_mod(u'%s.py' % name, name)
 541          self.assertEqual(mod.__name__, name)
 542  
 543      def test_get_exception_only(self):
 544          try:
 545              raise KeyboardInterrupt
 546          except KeyboardInterrupt:
 547              msg = ip.get_exception_only()
 548          self.assertEqual(msg, 'KeyboardInterrupt\n')
 549  
 550          try:
 551              raise DerivedInterrupt("foo")
 552          except KeyboardInterrupt:
 553              msg = ip.get_exception_only()
 554          self.assertEqual(msg, 'IPython.core.tests.test_interactiveshell.DerivedInterrupt: foo\n')
 555  
 556      def test_inspect_text(self):
 557          ip.run_cell('a = 5')
 558          text = ip.object_inspect_text('a')
 559          self.assertIsInstance(text, str)
 560  
 561      def test_last_execution_result(self):
 562          """ Check that last execution result gets set correctly (GH-10702) """
 563          result = ip.run_cell('a = 5; a')
 564          self.assertTrue(ip.last_execution_succeeded)
 565          self.assertEqual(ip.last_execution_result.result, 5)
 566  
 567          result = ip.run_cell('a = x_invalid_id_x')
 568          self.assertFalse(ip.last_execution_succeeded)
 569          self.assertFalse(ip.last_execution_result.success)
 570          self.assertIsInstance(ip.last_execution_result.error_in_exec, NameError)
 571  
 572      def test_reset_aliasing(self):
 573          """ Check that standard posix aliases work after %reset. """
 574          if os.name != 'posix':
 575              return
 576  
 577          ip.reset()
 578          for cmd in ('clear', 'more', 'less', 'man'):
 579              res = ip.run_cell('%' + cmd)
 580              self.assertEqual(res.success, True)
 581  
 582  
 583  @pytest.mark.skipif(
 584      sys.implementation.name == "pypy"
 585      and ((7, 3, 13) < sys.implementation.version < (7, 3, 16)),
 586      reason="Unicode issues with scandir on PyPy, see https://github.com/pypy/pypy/issues/4860",
 587  )
 588  class TestSafeExecfileNonAsciiPath(unittest.TestCase):
 589      @onlyif_unicode_paths
 590      def setUp(self):
 591          self.BASETESTDIR = tempfile.mkdtemp()
 592          self.TESTDIR = join(self.BASETESTDIR, u"åäö")
 593          os.mkdir(self.TESTDIR)
 594          with open(
 595              join(self.TESTDIR, "åäötestscript.py"), "w", encoding="utf-8"
 596          ) as sfile:
 597              sfile.write("pass\n")
 598          self.oldpath = os.getcwd()
 599          os.chdir(self.TESTDIR)
 600          self.fname = u"åäötestscript.py"
 601  
 602      def tearDown(self):
 603          os.chdir(self.oldpath)
 604          shutil.rmtree(self.BASETESTDIR)
 605  
 606      @onlyif_unicode_paths
 607      def test_1(self):
 608          """Test safe_execfile with non-ascii path
 609          """
 610          ip.safe_execfile(self.fname, {}, raise_exceptions=True)
 611  
 612  class ExitCodeChecks(tt.TempFileMixin):
 613  
 614      def setUp(self):
 615          self.system = ip.system_raw
 616  
 617      def test_exit_code_ok(self):
 618          self.system('exit 0')
 619          self.assertEqual(ip.user_ns['_exit_code'], 0)
 620  
 621      def test_exit_code_error(self):
 622          self.system('exit 1')
 623          self.assertEqual(ip.user_ns['_exit_code'], 1)
 624  
 625      @skipif(not hasattr(signal, 'SIGALRM'))
 626      def test_exit_code_signal(self):
 627          self.mktmp("import signal, time\n"
 628                     "signal.setitimer(signal.ITIMER_REAL, 0.1)\n"
 629                     "time.sleep(1)\n")
 630          self.system("%s %s" % (sys.executable, self.fname))
 631          self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGALRM)
 632  
 633      @onlyif_cmds_exist("csh")
 634      def test_exit_code_signal_csh(self):  # pragma: no cover
 635          SHELL = os.environ.get("SHELL", None)
 636          os.environ["SHELL"] = find_cmd("csh")
 637          try:
 638              self.test_exit_code_signal()
 639          finally:
 640              if SHELL is not None:
 641                  os.environ['SHELL'] = SHELL
 642              else:
 643                  del os.environ['SHELL']
 644  
 645  
 646  class TestSystemRaw(ExitCodeChecks):
 647  
 648      def setUp(self):
 649          super().setUp()
 650          self.system = ip.system_raw
 651  
 652      @onlyif_unicode_paths
 653      def test_1(self):
 654          """Test system_raw with non-ascii cmd
 655          """
 656          cmd = u'''python -c "'åäö'"   '''
 657          ip.system_raw(cmd)
 658  
 659      @mock.patch('subprocess.call', side_effect=KeyboardInterrupt)
 660      @mock.patch('os.system', side_effect=KeyboardInterrupt)
 661      def test_control_c(self, *mocks):
 662          try:
 663              self.system("sleep 1 # won't happen")
 664          except KeyboardInterrupt:  # pragma: no cove
 665              self.fail(
 666                  "system call should intercept "
 667                  "keyboard interrupt from subprocess.call"
 668              )
 669          self.assertEqual(ip.user_ns["_exit_code"], -signal.SIGINT)
 670  
 671  
 672  @pytest.mark.parametrize("magic_cmd", ["pip", "conda", "cd"])
 673  def test_magic_warnings(magic_cmd):
 674      if sys.platform == "win32":
 675          to_mock = "os.system"
 676          expected_arg, expected_kwargs = magic_cmd, dict()
 677      else:
 678          to_mock = "subprocess.call"
 679          expected_arg, expected_kwargs = magic_cmd, dict(
 680              shell=True, executable=os.environ.get("SHELL", None)
 681          )
 682  
 683      with mock.patch(to_mock, return_value=0) as mock_sub:
 684          with pytest.warns(Warning, match=r"You executed the system command"):
 685              ip.system_raw(magic_cmd)
 686          mock_sub.assert_called_once_with(expected_arg, **expected_kwargs)
 687  
 688  
 689  # TODO: Exit codes are currently ignored on Windows.
 690  class TestSystemPipedExitCode(ExitCodeChecks):
 691  
 692      def setUp(self):
 693          super().setUp()
 694          self.system = ip.system_piped
 695  
 696      @skip_win32
 697      def test_exit_code_ok(self):
 698          ExitCodeChecks.test_exit_code_ok(self)
 699  
 700      @skip_win32
 701      def test_exit_code_error(self):
 702          ExitCodeChecks.test_exit_code_error(self)
 703  
 704      @skip_win32
 705      def test_exit_code_signal(self):
 706          ExitCodeChecks.test_exit_code_signal(self)
 707  
 708  class TestModules(tt.TempFileMixin):
 709      def test_extraneous_loads(self):
 710          """Test we're not loading modules on startup that we shouldn't.
 711          """
 712          self.mktmp("import sys\n"
 713                     "print('numpy' in sys.modules)\n"
 714                     "print('ipyparallel' in sys.modules)\n"
 715                     "print('ipykernel' in sys.modules)\n"
 716                     )
 717          out = "False\nFalse\nFalse\n"
 718          tt.ipexec_validate(self.fname, out)
 719  
 720  class Negator(ast.NodeTransformer):
 721      """Negates all number literals in an AST."""
 722  
 723      def visit_Num(self, node):
 724          node.value = -node.value
 725          return node
 726  
 727      def visit_Constant(self, node):
 728          if isinstance(node.value, int):
 729              return self.visit_Num(node)
 730          return node
 731  
 732  class TestAstTransform(unittest.TestCase):
 733      def setUp(self):
 734          self.negator = Negator()
 735          ip.ast_transformers.append(self.negator)
 736  
 737      def tearDown(self):
 738          ip.ast_transformers.remove(self.negator)
 739  
 740      def test_non_int_const(self):
 741          with tt.AssertPrints("hello"):
 742              ip.run_cell('print("hello")')
 743  
 744      def test_run_cell(self):
 745          with tt.AssertPrints("-34"):
 746              ip.run_cell("print(12 + 22)")
 747  
 748          # A named reference to a number shouldn't be transformed.
 749          ip.user_ns["n"] = 55
 750          with tt.AssertNotPrints("-55"):
 751              ip.run_cell("print(n)")
 752  
 753      def test_timeit(self):
 754          called = set()
 755          def f(x):
 756              called.add(x)
 757          ip.push({'f':f})
 758  
 759          with tt.AssertPrints("std. dev. of"):
 760              ip.run_line_magic("timeit", "-n1 f(1)")
 761          self.assertEqual(called, {-1})
 762          called.clear()
 763  
 764          with tt.AssertPrints("std. dev. of"):
 765              ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
 766          self.assertEqual(called, {-2, -3})
 767  
 768      def test_time(self):
 769          called = []
 770          def f(x):
 771              called.append(x)
 772          ip.push({'f':f})
 773  
 774          # Test with an expression
 775          with tt.AssertPrints("Wall time: "):
 776              ip.run_line_magic("time", "f(5+9)")
 777          self.assertEqual(called, [-14])
 778          called[:] = []
 779  
 780          # Test with a statement (different code path)
 781          with tt.AssertPrints("Wall time: "):
 782              ip.run_line_magic("time", "a = f(-3 + -2)")
 783          self.assertEqual(called, [5])
 784  
 785      def test_macro(self):
 786          ip.push({'a':10})
 787          # The AST transformation makes this do a+=-1
 788          ip.define_macro("amacro", "a+=1\nprint(a)")
 789  
 790          with tt.AssertPrints("9"):
 791              ip.run_cell("amacro")
 792          with tt.AssertPrints("8"):
 793              ip.run_cell("amacro")
 794  
 795  class TestMiscTransform(unittest.TestCase):
 796  
 797  
 798      def test_transform_only_once(self):
 799          cleanup = 0
 800          line_t = 0
 801          def count_cleanup(lines):
 802              nonlocal cleanup
 803              cleanup += 1
 804              return lines
 805  
 806          def count_line_t(lines):
 807              nonlocal line_t
 808              line_t += 1
 809              return lines
 810  
 811          ip.input_transformer_manager.cleanup_transforms.append(count_cleanup)
 812          ip.input_transformer_manager.line_transforms.append(count_line_t)
 813  
 814          ip.run_cell('1')
 815  
 816          assert cleanup == 1
 817          assert line_t == 1
 818  
 819  class IntegerWrapper(ast.NodeTransformer):
 820      """Wraps all integers in a call to Integer()"""
 821  
 822      # for Python 3.7 and earlier
 823  
 824      # for Python 3.7 and earlier
 825      def visit_Num(self, node):
 826          if isinstance(node.n, int):
 827              return ast.Call(func=ast.Name(id='Integer', ctx=ast.Load()),
 828                              args=[node], keywords=[])
 829          return node
 830  
 831      # For Python 3.8+
 832      def visit_Constant(self, node):
 833          if isinstance(node.value, int):
 834              return self.visit_Num(node)
 835          return node
 836  
 837  
 838  class TestAstTransform2(unittest.TestCase):
 839      def setUp(self):
 840          self.intwrapper = IntegerWrapper()
 841          ip.ast_transformers.append(self.intwrapper)
 842  
 843          self.calls = []
 844          def Integer(*args):
 845              self.calls.append(args)
 846              return args
 847          ip.push({"Integer": Integer})
 848  
 849      def tearDown(self):
 850          ip.ast_transformers.remove(self.intwrapper)
 851          del ip.user_ns['Integer']
 852  
 853      def test_run_cell(self):
 854          ip.run_cell("n = 2")
 855          self.assertEqual(self.calls, [(2,)])
 856  
 857          # This shouldn't throw an error
 858          ip.run_cell("o = 2.0")
 859          self.assertEqual(ip.user_ns['o'], 2.0)
 860  
 861      def test_run_cell_non_int(self):
 862          ip.run_cell("n = 'a'")
 863          assert self.calls == []
 864  
 865      def test_timeit(self):
 866          called = set()
 867          def f(x):
 868              called.add(x)
 869          ip.push({'f':f})
 870  
 871          with tt.AssertPrints("std. dev. of"):
 872              ip.run_line_magic("timeit", "-n1 f(1)")
 873          self.assertEqual(called, {(1,)})
 874          called.clear()
 875  
 876          with tt.AssertPrints("std. dev. of"):
 877              ip.run_cell_magic("timeit", "-n1 f(2)", "f(3)")
 878          self.assertEqual(called, {(2,), (3,)})
 879  
 880  class ErrorTransformer(ast.NodeTransformer):
 881      """Throws an error when it sees a number."""
 882  
 883      def visit_Constant(self, node):
 884          if isinstance(node.value, int):
 885              raise ValueError("test")
 886          return node
 887  
 888  
 889  class TestAstTransformError(unittest.TestCase):
 890      def test_unregistering(self):
 891          err_transformer = ErrorTransformer()
 892          ip.ast_transformers.append(err_transformer)
 893  
 894          with self.assertWarnsRegex(UserWarning, "It will be unregistered"):
 895              ip.run_cell("1 + 2")
 896  
 897          # This should have been removed.
 898          self.assertNotIn(err_transformer, ip.ast_transformers)
 899  
 900  
 901  class StringRejector(ast.NodeTransformer):
 902      """Throws an InputRejected when it sees a string literal.
 903  
 904      Used to verify that NodeTransformers can signal that a piece of code should
 905      not be executed by throwing an InputRejected.
 906      """
 907  
 908      def visit_Constant(self, node):
 909          if isinstance(node.value, str):
 910              raise InputRejected("test")
 911          return node
 912  
 913  
 914  class TestAstTransformInputRejection(unittest.TestCase):
 915  
 916      def setUp(self):
 917          self.transformer = StringRejector()
 918          ip.ast_transformers.append(self.transformer)
 919  
 920      def tearDown(self):
 921          ip.ast_transformers.remove(self.transformer)
 922  
 923      def test_input_rejection(self):
 924          """Check that NodeTransformers can reject input."""
 925  
 926          expect_exception_tb = tt.AssertPrints("InputRejected: test")
 927          expect_no_cell_output = tt.AssertNotPrints("'unsafe'", suppress=False)
 928  
 929          # Run the same check twice to verify that the transformer is not
 930          # disabled after raising.
 931          with expect_exception_tb, expect_no_cell_output:
 932              ip.run_cell("'unsafe'")
 933  
 934          with expect_exception_tb, expect_no_cell_output:
 935              res = ip.run_cell("'unsafe'")
 936  
 937          self.assertIsInstance(res.error_before_exec, InputRejected)
 938  
 939  def test__IPYTHON__():
 940      # This shouldn't raise a NameError, that's all
 941      __IPYTHON__
 942  
 943  
 944  class DummyRepr(object):
 945      def __repr__(self):
 946          return "DummyRepr"
 947  
 948      def _repr_html_(self):
 949          return "<b>dummy</b>"
 950  
 951      def _repr_javascript_(self):
 952          return "console.log('hi');", {'key': 'value'}
 953  
 954  
 955  def test_user_variables():
 956      # enable all formatters
 957      ip.display_formatter.active_types = ip.display_formatter.format_types
 958  
 959      ip.user_ns['dummy'] = d = DummyRepr()
 960      keys = {'dummy', 'doesnotexist'}
 961      r = ip.user_expressions({ key:key for key in keys})
 962  
 963      assert keys == set(r.keys())
 964      dummy = r["dummy"]
 965      assert {"status", "data", "metadata"} == set(dummy.keys())
 966      assert dummy["status"] == "ok"
 967      data = dummy["data"]
 968      metadata = dummy["metadata"]
 969      assert data.get("text/html") == d._repr_html_()
 970      js, jsmd = d._repr_javascript_()
 971      assert data.get("application/javascript") == js
 972      assert metadata.get("application/javascript") == jsmd
 973  
 974      dne = r["doesnotexist"]
 975      assert dne["status"] == "error"
 976      assert dne["ename"] == "NameError"
 977  
 978      # back to text only
 979      ip.display_formatter.active_types = ['text/plain']
 980  
 981  def test_user_expression():
 982      # enable all formatters
 983      ip.display_formatter.active_types = ip.display_formatter.format_types
 984      query = {
 985          'a' : '1 + 2',
 986          'b' : '1/0',
 987      }
 988      r = ip.user_expressions(query)
 989      import pprint
 990      pprint.pprint(r)
 991      assert set(r.keys()) == set(query.keys())
 992      a = r["a"]
 993      assert {"status", "data", "metadata"} == set(a.keys())
 994      assert a["status"] == "ok"
 995      data = a["data"]
 996      metadata = a["metadata"]
 997      assert data.get("text/plain") == "3"
 998  
 999      b = r["b"]
1000      assert b["status"] == "error"
1001      assert b["ename"] == "ZeroDivisionError"
1002  
1003      # back to text only
1004      ip.display_formatter.active_types = ['text/plain']
1005  
1006  
1007  class TestSyntaxErrorTransformer(unittest.TestCase):
1008      """Check that SyntaxError raised by an input transformer is handled by run_cell()"""
1009  
1010      @staticmethod
1011      def transformer(lines):
1012          for line in lines:
1013              pos = line.find('syntaxerror')
1014              if pos >= 0:
1015                  e = SyntaxError('input contains "syntaxerror"')
1016                  e.text = line
1017                  e.offset = pos + 1
1018                  raise e
1019          return lines
1020  
1021      def setUp(self):
1022          ip.input_transformers_post.append(self.transformer)
1023  
1024      def tearDown(self):
1025          ip.input_transformers_post.remove(self.transformer)
1026  
1027      def test_syntaxerror_input_transformer(self):
1028          with tt.AssertPrints('1234'):
1029              ip.run_cell('1234')
1030          with tt.AssertPrints('SyntaxError: invalid syntax'):
1031              ip.run_cell('1 2 3')   # plain python syntax error
1032          with tt.AssertPrints('SyntaxError: input contains "syntaxerror"'):
1033              ip.run_cell('2345  # syntaxerror')  # input transformer syntax error
1034          with tt.AssertPrints('3456'):
1035              ip.run_cell('3456')
1036  
1037  
1038  class TestWarningSuppression(unittest.TestCase):
1039      def test_warning_suppression(self):
1040          ip.run_cell("import warnings")
1041          try:
1042              with self.assertWarnsRegex(UserWarning, "asdf"):
1043                  ip.run_cell("warnings.warn('asdf')")
1044              # Here's the real test -- if we run that again, we should get the
1045              # warning again. Traditionally, each warning was only issued once per
1046              # IPython session (approximately), even if the user typed in new and
1047              # different code that should have also triggered the warning, leading
1048              # to much confusion.
1049              with self.assertWarnsRegex(UserWarning, "asdf"):
1050                  ip.run_cell("warnings.warn('asdf')")
1051          finally:
1052              ip.run_cell("del warnings")
1053  
1054  
1055      def test_deprecation_warning(self):
1056          ip.run_cell("""
1057  import warnings
1058  def wrn():
1059      warnings.warn(
1060          "I AM  A WARNING",
1061          DeprecationWarning
1062      )
1063          """)
1064          try:
1065              with self.assertWarnsRegex(DeprecationWarning, "I AM  A WARNING"):
1066                  ip.run_cell("wrn()")
1067          finally:
1068              ip.run_cell("del warnings")
1069              ip.run_cell("del wrn")
1070  
1071  
1072  class TestImportNoDeprecate(tt.TempFileMixin):
1073  
1074      def setUp(self):
1075          """Make a valid python temp file."""
1076          self.mktmp("""
1077  import warnings
1078  def wrn():
1079      warnings.warn(
1080          "I AM  A WARNING",
1081          DeprecationWarning
1082      )
1083  """)
1084          super().setUp()
1085  
1086      def test_no_dep(self):
1087          """
1088          No deprecation warning should be raised from imported functions
1089          """
1090          ip.run_cell("from {} import wrn".format(self.fname))
1091  
1092          with tt.AssertNotPrints("I AM  A WARNING"):
1093              ip.run_cell("wrn()")
1094          ip.run_cell("del wrn")
1095  
1096  
1097  def test_custom_exc_count():
1098      hook = mock.Mock(return_value=None)
1099      ip.set_custom_exc((SyntaxError,), hook)
1100      before = ip.execution_count
1101      ip.run_cell("def foo()", store_history=True)
1102      # restore default excepthook
1103      ip.set_custom_exc((), None)
1104      assert hook.call_count == 1
1105      assert ip.execution_count == before + 1
1106  
1107  
1108  def test_run_cell_async():
1109      ip.run_cell("import asyncio")
1110      coro = ip.run_cell_async("await asyncio.sleep(0.01)\n5")
1111      assert asyncio.iscoroutine(coro)
1112      loop = asyncio.new_event_loop()
1113      result = loop.run_until_complete(coro)
1114      assert isinstance(result, interactiveshell.ExecutionResult)
1115      assert result.result == 5
1116  
1117  
1118  def test_run_cell_await():
1119      ip.run_cell("import asyncio")
1120      result = ip.run_cell("await asyncio.sleep(0.01); 10")
1121      assert ip.user_ns["_"] == 10
1122  
1123  
1124  def test_run_cell_asyncio_run():
1125      ip.run_cell("import asyncio")
1126      result = ip.run_cell("await asyncio.sleep(0.01); 1")
1127      assert ip.user_ns["_"] == 1
1128      result = ip.run_cell("asyncio.run(asyncio.sleep(0.01)); 2")
1129      assert ip.user_ns["_"] == 2
1130      result = ip.run_cell("await asyncio.sleep(0.01); 3")
1131      assert ip.user_ns["_"] == 3
1132  
1133  
1134  def test_should_run_async():
1135      assert not ip.should_run_async("a = 5", transformed_cell="a = 5")
1136      assert ip.should_run_async("await x", transformed_cell="await x")
1137      assert ip.should_run_async(
1138          "import asyncio; await asyncio.sleep(1)",
1139          transformed_cell="import asyncio; await asyncio.sleep(1)",
1140      )
1141  
1142  
1143  def test_set_custom_completer():
1144      num_completers = len(ip.Completer.matchers)
1145  
1146      def foo(*args, **kwargs):
1147          return "I'm a completer!"
1148  
1149      ip.set_custom_completer(foo, 0)
1150  
1151      # check that we've really added a new completer
1152      assert len(ip.Completer.matchers) == num_completers + 1
1153  
1154      # check that the first completer is the function we defined
1155      assert ip.Completer.matchers[0]() == "I'm a completer!"
1156  
1157      # clean up
1158      ip.Completer.custom_matchers.pop()
1159  
1160  
1161  class TestShowTracebackAttack(unittest.TestCase):
1162      """Test that the interactive shell is resilient against the client attack of
1163      manipulating the showtracebacks method. These attacks shouldn't result in an
1164      unhandled exception in the kernel."""
1165  
1166      def setUp(self):
1167          self.orig_showtraceback = interactiveshell.InteractiveShell.showtraceback
1168  
1169      def tearDown(self):
1170          interactiveshell.InteractiveShell.showtraceback = self.orig_showtraceback
1171  
1172      def test_set_show_tracebacks_none(self):
1173          """Test the case of the client setting showtracebacks to None"""
1174  
1175          result = ip.run_cell(
1176              """
1177              import IPython.core.interactiveshell
1178              IPython.core.interactiveshell.InteractiveShell.showtraceback = None
1179  
1180              assert False, "This should not raise an exception"
1181          """
1182          )
1183          print(result)
1184  
1185          assert result.result is None
1186          assert isinstance(result.error_in_exec, TypeError)
1187          assert str(result.error_in_exec) == "'NoneType' object is not callable"
1188  
1189      def test_set_show_tracebacks_noop(self):
1190          """Test the case of the client setting showtracebacks to a no op lambda"""
1191  
1192          result = ip.run_cell(
1193              """
1194              import IPython.core.interactiveshell
1195              IPython.core.interactiveshell.InteractiveShell.showtraceback = lambda *args, **kwargs: None
1196  
1197              assert False, "This should not raise an exception"
1198          """
1199          )
1200          print(result)
1201  
1202          assert result.result is None
1203          assert isinstance(result.error_in_exec, AssertionError)
1204          assert str(result.error_in_exec) == "This should not raise an exception"
1205  
1206  
1207  @skip_if_not_osx
1208  def test_enable_gui_osx():
1209      simple_prompt = ip.simple_prompt
1210      ip.simple_prompt = False
1211  
1212      ip.enable_gui("osx")
1213      assert ip.active_eventloop == "osx"
1214      ip.enable_gui()
1215  
1216      # The following line fails for IPython <= 8.25.0
1217      ip.enable_gui("macosx")
1218      assert ip.active_eventloop == "osx"
1219      ip.enable_gui()
1220  
1221      ip.simple_prompt = simple_prompt
1222