|
|
"""
|
|
|
Test for async helpers.
|
|
|
|
|
|
Should only trigger on python 3.5+ or will have syntax errors.
|
|
|
"""
|
|
|
from itertools import chain, repeat
|
|
|
from textwrap import dedent, indent
|
|
|
from typing import TYPE_CHECKING
|
|
|
from unittest import TestCase
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
from IPython.core.async_helpers import _should_be_async
|
|
|
from IPython.testing.decorators import skip_without
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
|
from IPython import get_ipython
|
|
|
|
|
|
ip = get_ipython()
|
|
|
|
|
|
|
|
|
def iprc(x):
|
|
|
return ip.run_cell(dedent(x)).raise_error()
|
|
|
|
|
|
|
|
|
def iprc_nr(x):
|
|
|
return ip.run_cell(dedent(x))
|
|
|
|
|
|
|
|
|
class AsyncTest(TestCase):
|
|
|
def test_should_be_async(self):
|
|
|
self.assertFalse(_should_be_async("False"))
|
|
|
self.assertTrue(_should_be_async("await bar()"))
|
|
|
self.assertTrue(_should_be_async("x = await bar()"))
|
|
|
self.assertFalse(
|
|
|
_should_be_async(
|
|
|
dedent(
|
|
|
"""
|
|
|
async def awaitable():
|
|
|
pass
|
|
|
"""
|
|
|
)
|
|
|
)
|
|
|
)
|
|
|
|
|
|
def _get_top_level_cases(self):
|
|
|
# These are test cases that should be valid in a function
|
|
|
# but invalid outside of a function.
|
|
|
test_cases = []
|
|
|
test_cases.append(('basic', "{val}"))
|
|
|
|
|
|
# Note, in all conditional cases, I use True instead of
|
|
|
# False so that the peephole optimizer won't optimize away
|
|
|
# the return, so CPython will see this as a syntax error:
|
|
|
#
|
|
|
# while True:
|
|
|
# break
|
|
|
# return
|
|
|
#
|
|
|
# But not this:
|
|
|
#
|
|
|
# while False:
|
|
|
# return
|
|
|
#
|
|
|
# See https://bugs.python.org/issue1875
|
|
|
|
|
|
test_cases.append(('if', dedent("""
|
|
|
if True:
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('while', dedent("""
|
|
|
while True:
|
|
|
{val}
|
|
|
break
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('try', dedent("""
|
|
|
try:
|
|
|
{val}
|
|
|
except:
|
|
|
pass
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('except', dedent("""
|
|
|
try:
|
|
|
pass
|
|
|
except:
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('finally', dedent("""
|
|
|
try:
|
|
|
pass
|
|
|
except:
|
|
|
pass
|
|
|
finally:
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('for', dedent("""
|
|
|
for _ in range(4):
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
|
|
|
test_cases.append(('nested', dedent("""
|
|
|
if True:
|
|
|
while True:
|
|
|
{val}
|
|
|
break
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('deep-nested', dedent("""
|
|
|
if True:
|
|
|
while True:
|
|
|
break
|
|
|
for x in range(3):
|
|
|
if True:
|
|
|
while True:
|
|
|
for x in range(3):
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
return test_cases
|
|
|
|
|
|
def _get_ry_syntax_errors(self):
|
|
|
# This is a mix of tests that should be a syntax error if
|
|
|
# return or yield whether or not they are in a function
|
|
|
|
|
|
test_cases = []
|
|
|
|
|
|
test_cases.append(('class', dedent("""
|
|
|
class V:
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
test_cases.append(('nested-class', dedent("""
|
|
|
class V:
|
|
|
class C:
|
|
|
{val}
|
|
|
""")))
|
|
|
|
|
|
return test_cases
|
|
|
|
|
|
|
|
|
def test_top_level_return_error(self):
|
|
|
tl_err_test_cases = self._get_top_level_cases()
|
|
|
tl_err_test_cases.extend(self._get_ry_syntax_errors())
|
|
|
|
|
|
vals = ('return', 'yield', 'yield from (_ for _ in range(3))',
|
|
|
dedent('''
|
|
|
def f():
|
|
|
pass
|
|
|
return
|
|
|
'''),
|
|
|
)
|
|
|
|
|
|
for test_name, test_case in tl_err_test_cases:
|
|
|
# This example should work if 'pass' is used as the value
|
|
|
with self.subTest((test_name, 'pass')):
|
|
|
iprc(test_case.format(val='pass'))
|
|
|
|
|
|
# It should fail with all the values
|
|
|
for val in vals:
|
|
|
with self.subTest((test_name, val)):
|
|
|
msg = "Syntax error not raised for %s, %s" % (test_name, val)
|
|
|
with self.assertRaises(SyntaxError, msg=msg):
|
|
|
iprc(test_case.format(val=val))
|
|
|
|
|
|
def test_in_func_no_error(self):
|
|
|
# Test that the implementation of top-level return/yield
|
|
|
# detection isn't *too* aggressive, and works inside a function
|
|
|
func_contexts = []
|
|
|
|
|
|
func_contexts.append(('func', False, dedent("""
|
|
|
def f():""")))
|
|
|
|
|
|
func_contexts.append(('method', False, dedent("""
|
|
|
class MyClass:
|
|
|
def __init__(self):
|
|
|
""")))
|
|
|
|
|
|
func_contexts.append(('async-func', True, dedent("""
|
|
|
async def f():""")))
|
|
|
|
|
|
func_contexts.append(('async-method', True, dedent("""
|
|
|
class MyClass:
|
|
|
async def f(self):""")))
|
|
|
|
|
|
func_contexts.append(('closure', False, dedent("""
|
|
|
def f():
|
|
|
def g():
|
|
|
""")))
|
|
|
|
|
|
def nest_case(context, case):
|
|
|
# Detect indentation
|
|
|
lines = context.strip().splitlines()
|
|
|
prefix_len = 0
|
|
|
for c in lines[-1]:
|
|
|
if c != ' ':
|
|
|
break
|
|
|
prefix_len += 1
|
|
|
|
|
|
indented_case = indent(case, ' ' * (prefix_len + 4))
|
|
|
return context + '\n' + indented_case
|
|
|
|
|
|
# Gather and run the tests
|
|
|
|
|
|
# yield is allowed in async functions, starting in Python 3.6,
|
|
|
# and yield from is not allowed in any version
|
|
|
vals = ('return', 'yield', 'yield from (_ for _ in range(3))')
|
|
|
|
|
|
success_tests = zip(self._get_top_level_cases(), repeat(False))
|
|
|
failure_tests = zip(self._get_ry_syntax_errors(), repeat(True))
|
|
|
|
|
|
tests = chain(success_tests, failure_tests)
|
|
|
|
|
|
for context_name, async_func, context in func_contexts:
|
|
|
for (test_name, test_case), should_fail in tests:
|
|
|
nested_case = nest_case(context, test_case)
|
|
|
|
|
|
for val in vals:
|
|
|
test_id = (context_name, test_name, val)
|
|
|
cell = nested_case.format(val=val)
|
|
|
|
|
|
with self.subTest(test_id):
|
|
|
if should_fail:
|
|
|
msg = ("SyntaxError not raised for %s" %
|
|
|
str(test_id))
|
|
|
with self.assertRaises(SyntaxError, msg=msg):
|
|
|
iprc(cell)
|
|
|
|
|
|
print(cell)
|
|
|
else:
|
|
|
iprc(cell)
|
|
|
|
|
|
def test_nonlocal(self):
|
|
|
# fails if outer scope is not a function scope or if var not defined
|
|
|
with self.assertRaises(SyntaxError):
|
|
|
iprc("nonlocal x")
|
|
|
iprc("""
|
|
|
x = 1
|
|
|
def f():
|
|
|
nonlocal x
|
|
|
x = 10000
|
|
|
yield x
|
|
|
""")
|
|
|
iprc("""
|
|
|
def f():
|
|
|
def g():
|
|
|
nonlocal x
|
|
|
x = 10000
|
|
|
yield x
|
|
|
""")
|
|
|
|
|
|
# works if outer scope is a function scope and var exists
|
|
|
iprc("""
|
|
|
def f():
|
|
|
x = 20
|
|
|
def g():
|
|
|
nonlocal x
|
|
|
x = 10000
|
|
|
yield x
|
|
|
""")
|
|
|
|
|
|
|
|
|
def test_execute(self):
|
|
|
iprc("""
|
|
|
import asyncio
|
|
|
await asyncio.sleep(0.001)
|
|
|
"""
|
|
|
)
|
|
|
|
|
|
def test_autoawait(self):
|
|
|
iprc("%autoawait False")
|
|
|
iprc("%autoawait True")
|
|
|
iprc("""
|
|
|
from asyncio import sleep
|
|
|
await sleep(0.1)
|
|
|
"""
|
|
|
)
|
|
|
|
|
|
def test_memory_error(self):
|
|
|
"""
|
|
|
The pgen parser in 3.8 or before use to raise MemoryError on too many
|
|
|
nested parens anymore"""
|
|
|
|
|
|
iprc("(" * 200 + ")" * 200)
|
|
|
|
|
|
@pytest.mark.xfail(reason="fail on curio 1.6 and before on Python 3.12")
|
|
|
@pytest.mark.skip(
|
|
|
reason="skip_without(curio) fails on 3.12 for now even with other skip so must uncond skip"
|
|
|
)
|
|
|
# @skip_without("curio")
|
|
|
def test_autoawait_curio(self):
|
|
|
iprc("%autoawait curio")
|
|
|
|
|
|
@skip_without('trio')
|
|
|
def test_autoawait_trio(self):
|
|
|
iprc("%autoawait trio")
|
|
|
|
|
|
@skip_without('trio')
|
|
|
def test_autoawait_trio_wrong_sleep(self):
|
|
|
iprc("%autoawait trio")
|
|
|
res = iprc_nr("""
|
|
|
import asyncio
|
|
|
await asyncio.sleep(0)
|
|
|
""")
|
|
|
with self.assertRaises(TypeError):
|
|
|
res.raise_error()
|
|
|
|
|
|
@skip_without('trio')
|
|
|
def test_autoawait_asyncio_wrong_sleep(self):
|
|
|
iprc("%autoawait asyncio")
|
|
|
res = iprc_nr("""
|
|
|
import trio
|
|
|
await trio.sleep(0)
|
|
|
""")
|
|
|
with self.assertRaises(RuntimeError):
|
|
|
res.raise_error()
|
|
|
|
|
|
|
|
|
def tearDown(self):
|
|
|
ip.loop_runner = "asyncio"
|
|
|
|
