Merge pull request #11382 from felixzhuologist/master

Fix  #11312: nonlocal does not raise SyntaxError

IPython/core/async_helpers.py

@@ -1,157 +1,165 b''
 """
 Async helper function that are invalid syntax on Python 3.5 and below.
 
 This code is best effort, and may have edge cases not behaving as expected. In
 particular it contain a number of heuristics to detect whether code is
 effectively async and need to run in an event loop or not.
 
 Some constructs (like top-level `return`, or `yield`) are taken care of
 explicitly to actually raise a SyntaxError and stay as close as possible to
 Python semantics.
 """
 
 
 import ast
 import sys
 from textwrap import dedent, indent
 
 
 class _AsyncIORunner:
 
     def __call__(self, coro):
         """
         Handler for asyncio autoawait
         """
         import asyncio
 
         return asyncio.get_event_loop().run_until_complete(coro)
 
     def __str__(self):
         return 'asyncio'
 
 _asyncio_runner = _AsyncIORunner()
 
 
 def _curio_runner(coroutine):
     """
     handler for curio autoawait
     """
     import curio
 
     return curio.run(coroutine)
 
 
 def _trio_runner(async_fn):
     import trio
 
     async def loc(coro):
         """
         We need the dummy no-op async def to protect from
         trio's internal. See https://github.com/python-trio/trio/issues/89
         """
         return await coro
 
     return trio.run(loc, async_fn)
 
 
 def _pseudo_sync_runner(coro):
     """
     A runner that does not really allow async execution, and just advance the coroutine.
 
     See discussion in https://github.com/python-trio/trio/issues/608,
 
     Credit to Nathaniel Smith
 
     """
     try:
         coro.send(None)
     except StopIteration as exc:
         return exc.value
     else:
         # TODO: do not raise but return an execution result with the right info.
         raise RuntimeError(
             "{coro_name!r} needs a real async loop".format(coro_name=coro.__name__)
         )
 
 
 def _asyncify(code: str) -> str:
     """wrap code in async def definition.
 
     And setup a bit of context to run it later.
     """
     res = dedent(
         """
     async def __wrapper__():
         try:
     {usercode}
         finally:
             locals()
     """
     ).format(usercode=indent(code, " " * 8))
     return res
 
 
 class _AsyncSyntaxErrorVisitor(ast.NodeVisitor):
     """
     Find syntax errors that would be an error in an async repl, but because
     the implementation involves wrapping the repl in an async function, it
     is erroneously allowed (e.g. yield or return at the top level)
     """
+    def __init__(self):
+        self.depth = 0
+        super().__init__()
 
     def generic_visit(self, node):
         func_types = (ast.FunctionDef, ast.AsyncFunctionDef)
-        invalid_types = (ast.Return, ast.Yield, ast.YieldFrom)
-
-        if isinstance(node, func_types):
-            return  # Don't recurse into functions
-        elif isinstance(node, invalid_types):
+        invalid_types_by_depth = {
+            0: (ast.Return, ast.Yield, ast.YieldFrom),
+            1: (ast.Nonlocal,)
+        }
+
+        should_traverse = self.depth < max(invalid_types_by_depth.keys())
+        if isinstance(node, func_types) and should_traverse:
+            self.depth += 1
+            super().generic_visit(node)
+        elif isinstance(node, invalid_types_by_depth[self.depth]):
             raise SyntaxError()
         else:
             super().generic_visit(node)
 
 
 def _async_parse_cell(cell: str) -> ast.AST:
     """
     This is a compatibility shim for pre-3.7 when async outside of a function
     is a syntax error at the parse stage.
 
     It will return an abstract syntax tree parsed as if async and await outside
     of a function were not a syntax error.
     """
     if sys.version_info < (3, 7):
         # Prior to 3.7 you need to asyncify before parse
         wrapped_parse_tree = ast.parse(_asyncify(cell))
         return wrapped_parse_tree.body[0].body[0]
     else:
         return ast.parse(cell)
 
 
 def _should_be_async(cell: str) -> bool:
     """Detect if a block of code need to be wrapped in an `async def`
 
     Attempt to parse the block of code, it it compile we're fine.
     Otherwise we  wrap if and try to compile.
 
     If it works, assume it should be async. Otherwise Return False.
 
     Not handled yet: If the block of code has a return statement as  the top
     level, it will be seen as async. This is a know limitation.
     """
 
     try:
         # we can't limit ourself to ast.parse, as it __accepts__ to parse on
         # 3.7+, but just does not _compile_
         compile(cell, "<>", "exec")
         return False
     except SyntaxError:
         try:
             parse_tree = _async_parse_cell(cell)
 
             # Raise a SyntaxError if there are top-level return or yields
             v = _AsyncSyntaxErrorVisitor()
             v.visit(parse_tree)
 
         except SyntaxError:
             return False
         return True
     return False

IPython/core/tests/test_async_helpers.py

@@ -1,277 +1,306 b''
 """
 Test for async helpers. 
 
 Should only trigger on python 3.5+ or will have syntax errors.
 """
 
 import sys
 from itertools import chain, repeat
 import nose.tools as nt
 from textwrap import dedent, indent
 from unittest import TestCase
 from IPython.testing.decorators import skip_without
 
 ip = get_ipython()
 iprc = lambda x: ip.run_cell(dedent(x)).raise_error()
 iprc_nr = lambda x: ip.run_cell(dedent(x))
 
 if sys.version_info > (3, 5):
     from IPython.core.async_helpers import _should_be_async
 
     class AsyncTest(TestCase):
         def test_should_be_async(self):
             nt.assert_false(_should_be_async("False"))
             nt.assert_true(_should_be_async("await bar()"))
             nt.assert_true(_should_be_async("x = await bar()"))
             nt.assert_false(
                 _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))')
 
             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))')
             async_safe = (True,
                           sys.version_info >= (3, 6),
                           False)
             vals = tuple(zip(vals, async_safe))
 
             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, async_safe in vals:
                         val_should_fail = (should_fail or
                                            (async_func and not async_safe))
 
                         test_id = (context_name, test_name, val)
                         cell = nested_case.format(val=val)
 
                         with self.subTest(test_id):
                             if val_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)
             """
             )
 
         @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 nt.assert_raises(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 nt.assert_raises(RuntimeError):
                 res.raise_error()
 
 
         def tearDown(self):
             ip.loop_runner = "asyncio"