1  """
   2  Test for async helpers.
   3  
   4  Should only trigger on python 3.5+ or will have syntax errors.
   5  """
   6  from itertools import chain, repeat
   7  from textwrap import dedent, indent
   8  from typing import TYPE_CHECKING
   9  from unittest import TestCase
  10  
  11  import pytest
  12  
  13  from IPython.core.async_helpers import _should_be_async
  14  from IPython.testing.decorators import skip_without
  15  
  16  if TYPE_CHECKING:
  17      from IPython import get_ipython
  18  
  19      ip = get_ipython()
  20  
  21  
  22  def iprc(x):
  23      return ip.run_cell(dedent(x)).raise_error()
  24  
  25  
  26  def iprc_nr(x):
  27      return ip.run_cell(dedent(x))
  28  
  29  
  30  class AsyncTest(TestCase):
  31      def test_should_be_async(self):
  32          self.assertFalse(_should_be_async("False"))
  33          self.assertTrue(_should_be_async("await bar()"))
  34          self.assertTrue(_should_be_async("x = await bar()"))
  35          self.assertFalse(
  36              _should_be_async(
  37                  dedent(
  38                      """
  39              async def awaitable():
  40                  pass
  41          """
  42                  )
  43              )
  44          )
  45  
  46      def _get_top_level_cases(self):
  47          # These are test cases that should be valid in a function
  48          # but invalid outside of a function.
  49          test_cases = []
  50          test_cases.append(('basic', "{val}"))
  51  
  52          # Note, in all conditional cases, I use True instead of
  53          # False so that the peephole optimizer won't optimize away
  54          # the return, so CPython will see this as a syntax error:
  55          #
  56          # while True:
  57          #    break
  58          #    return
  59          #
  60          # But not this:
  61          #
  62          # while False:
  63          #    return
  64          #
  65          # See https://bugs.python.org/issue1875
  66  
  67          test_cases.append(('if', dedent("""
  68          if True:
  69              {val}
  70          """)))
  71  
  72          test_cases.append(('while', dedent("""
  73          while True:
  74              {val}
  75              break
  76          """)))
  77  
  78          test_cases.append(('try', dedent("""
  79          try:
  80              {val}
  81          except:
  82              pass
  83          """)))
  84  
  85          test_cases.append(('except', dedent("""
  86          try:
  87              pass
  88          except:
  89              {val}
  90          """)))
  91  
  92          test_cases.append(('finally', dedent("""
  93          try:
  94              pass
  95          except:
  96              pass
  97          finally:
  98              {val}
  99          """)))
 100  
 101          test_cases.append(('for', dedent("""
 102          for _ in range(4):
 103              {val}
 104          """)))
 105  
 106  
 107          test_cases.append(('nested', dedent("""
 108          if True:
 109              while True:
 110                  {val}
 111                  break
 112          """)))
 113  
 114          test_cases.append(('deep-nested', dedent("""
 115          if True:
 116              while True:
 117                  break
 118                  for x in range(3):
 119                      if True:
 120                          while True:
 121                              for x in range(3):
 122                                  {val}
 123          """)))
 124  
 125          return test_cases
 126  
 127      def _get_ry_syntax_errors(self):
 128          # This is a mix of tests that should be a syntax error if
 129          # return or yield whether or not they are in a function
 130  
 131          test_cases = []
 132  
 133          test_cases.append(('class', dedent("""
 134          class V:
 135              {val}
 136          """)))
 137  
 138          test_cases.append(('nested-class', dedent("""
 139          class V:
 140              class C:
 141                  {val}
 142          """)))
 143  
 144          return test_cases
 145  
 146  
 147      def test_top_level_return_error(self):
 148          tl_err_test_cases = self._get_top_level_cases()
 149          tl_err_test_cases.extend(self._get_ry_syntax_errors())
 150  
 151          vals = ('return', 'yield', 'yield from (_ for _ in range(3))',
 152                  dedent('''
 153                      def f():
 154                          pass
 155                      return
 156                      '''),
 157                  )
 158  
 159          for test_name, test_case in tl_err_test_cases:
 160              # This example should work if 'pass' is used as the value
 161              with self.subTest((test_name, 'pass')):
 162                  iprc(test_case.format(val='pass'))
 163  
 164              # It should fail with all the values
 165              for val in vals:
 166                  with self.subTest((test_name, val)):
 167                      msg = "Syntax error not raised for %s, %s" % (test_name, val)
 168                      with self.assertRaises(SyntaxError, msg=msg):
 169                          iprc(test_case.format(val=val))
 170  
 171      def test_in_func_no_error(self):
 172          # Test that the implementation of top-level return/yield
 173          # detection isn't *too* aggressive, and works inside a function
 174          func_contexts = []
 175  
 176          func_contexts.append(('func', False, dedent("""
 177          def f():""")))
 178  
 179          func_contexts.append(('method', False, dedent("""
 180          class MyClass:
 181              def __init__(self):
 182          """)))
 183  
 184          func_contexts.append(('async-func', True,  dedent("""
 185          async def f():""")))
 186  
 187          func_contexts.append(('async-method', True,  dedent("""
 188          class MyClass:
 189              async def f(self):""")))
 190  
 191          func_contexts.append(('closure', False, dedent("""
 192          def f():
 193              def g():
 194          """)))
 195  
 196          def nest_case(context, case):
 197              # Detect indentation
 198              lines = context.strip().splitlines()
 199              prefix_len = 0
 200              for c in lines[-1]:
 201                  if c != ' ':
 202                      break
 203                  prefix_len += 1
 204  
 205              indented_case = indent(case, ' ' * (prefix_len + 4))
 206              return context + '\n' + indented_case
 207  
 208          # Gather and run the tests
 209  
 210          # yield is allowed in async functions, starting in Python 3.6,
 211          # and yield from is not allowed in any version
 212          vals = ('return', 'yield', 'yield from (_ for _ in range(3))')
 213  
 214          success_tests = zip(self._get_top_level_cases(), repeat(False))
 215          failure_tests = zip(self._get_ry_syntax_errors(), repeat(True))
 216  
 217          tests = chain(success_tests, failure_tests)
 218  
 219          for context_name, async_func, context in func_contexts:
 220              for (test_name, test_case), should_fail in tests:
 221                  nested_case = nest_case(context, test_case)
 222  
 223                  for val in vals:
 224                      test_id = (context_name, test_name, val)
 225                      cell = nested_case.format(val=val)
 226  
 227                      with self.subTest(test_id):
 228                          if should_fail:
 229                              msg = ("SyntaxError not raised for %s" %
 230                                      str(test_id))
 231                              with self.assertRaises(SyntaxError, msg=msg):
 232                                  iprc(cell)
 233  
 234                                  print(cell)
 235                          else:
 236                              iprc(cell)
 237  
 238      def test_nonlocal(self):
 239          # fails if outer scope is not a function scope or if var not defined
 240          with self.assertRaises(SyntaxError):
 241              iprc("nonlocal x")
 242              iprc("""
 243              x = 1
 244              def f():
 245                  nonlocal x
 246                  x = 10000
 247                  yield x
 248              """)
 249              iprc("""
 250              def f():
 251                  def g():
 252                      nonlocal x
 253                      x = 10000
 254                      yield x
 255              """)
 256  
 257          # works if outer scope is a function scope and var exists
 258          iprc("""
 259          def f():
 260              x = 20
 261              def g():
 262                  nonlocal x
 263                  x = 10000
 264                  yield x
 265          """)
 266  
 267  
 268      def test_execute(self):
 269          iprc("""
 270          import asyncio
 271          await asyncio.sleep(0.001)
 272          """
 273          )
 274  
 275      def test_autoawait(self):
 276          iprc("%autoawait False")
 277          iprc("%autoawait True")
 278          iprc("""
 279          from asyncio import sleep
 280          await sleep(0.1)
 281          """
 282          )
 283  
 284      def test_memory_error(self):
 285          """
 286          The pgen parser in 3.8 or before use to raise MemoryError on too many
 287          nested parens anymore"""
 288  
 289          iprc("(" * 200 + ")" * 200)
 290  
 291      @pytest.mark.xfail(reason="fail on curio 1.6 and before on Python 3.12")
 292      @pytest.mark.skip(
 293          reason="skip_without(curio) fails on 3.12 for now even with other skip so must uncond skip"
 294      )
 295      # @skip_without("curio")
 296      def test_autoawait_curio(self):
 297          iprc("%autoawait curio")
 298  
 299      @skip_without('trio')
 300      def test_autoawait_trio(self):
 301          iprc("%autoawait trio")
 302  
 303      @skip_without('trio')
 304      def test_autoawait_trio_wrong_sleep(self):
 305          iprc("%autoawait trio")
 306          res = iprc_nr("""
 307          import asyncio
 308          await asyncio.sleep(0)
 309          """)
 310          with self.assertRaises(TypeError):
 311              res.raise_error()
 312  
 313      @skip_without('trio')
 314      def test_autoawait_asyncio_wrong_sleep(self):
 315          iprc("%autoawait asyncio")
 316          res = iprc_nr("""
 317          import trio
 318          await trio.sleep(0)
 319          """)
 320          with self.assertRaises(RuntimeError):
 321              res.raise_error()
 322  
 323  
 324      def tearDown(self):
 325          ip.loop_runner = "asyncio"
 326