<commit_before>"""
Test for async helpers. 

Should only trigger on python 3.5+ or will have syntax errors.
"""

import sys
import nose.tools as nt
from textwrap import dedent
from unittest import TestCase

ip = get_ipython()
iprc = 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 test_execute(self):
            iprc(
                """
            import asyncio
            await asyncio.sleep(0.001)
            """
            )

        def test_autoawait(self):
            ip.run_cell("%autoawait False")
            ip.run_cell("%autoawait True")
            iprc(
                """
                from asyncio import sleep
                await.sleep(0.1)
            """
            )

        def test_autoawait_curio(self):
            ip.run_cell("%autoawait curio")

        def test_autoawait_trio(self):
            ip.run_cell("%autoawait trio")

        def tearDown(self):
            ip.loop_runner = "asyncio"
<commit_msg>Add tests for SyntaxError with top-level return<commit_after>"""
Test for async helpers. 

Should only trigger on python 3.5+ or will have syntax errors.
"""

import sys
import nose.tools as nt
from textwrap import dedent, indent
from unittest import TestCase

ip = get_ipython()
iprc = lambda x: ip.run_cell(dedent(x))
iprc_err = lambda x: iprc(x).raise_error()

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_err(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_err(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', dedent("""
            def f():""")))

            func_contexts.append(('method', dedent("""
            class MyClass:
                def __init__(self):
            """)))

            func_contexts.append(('async-func', dedent("""
            async def f():""")))

            func_contexts.append(('closure', 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
            vals = ('return', 'yield')

            success_tests = self._get_top_level_cases()
            failure_tests = self._get_ry_syntax_errors()

            for context_name, context in func_contexts:
                # These tests should now successfully run
                for test_name, test_case in success_tests:
                    nested_case = nest_case(context, test_case)

                    for val in vals:
                        with self.subTest((test_name, context_name, val)):
                            iprc_err(nested_case.format(val=val))

                # These tests should still raise a SyntaxError
                for test_name, test_case in failure_tests:
                    nested_case = nest_case(context, test_case)

                    for val in vals:
                        with self.subTest((test_name, context_name, val)):
                            with self.assertRaises(SyntaxError):
                                iprc_err(nested_case.format(val=val))


        def test_execute(self):
            iprc(
                """
            import asyncio
            await asyncio.sleep(0.001)
            """
            )

        def test_autoawait(self):
            ip.run_cell("%autoawait False")
            ip.run_cell("%autoawait True")
            iprc(
                """
                from asyncio import sleep
                await.sleep(0.1)
            """
            )

        def test_autoawait_curio(self):
            ip.run_cell("%autoawait curio")

        def test_autoawait_trio(self):
            ip.run_cell("%autoawait trio")

        def tearDown(self):
            ip.loop_runner = "asyncio"

{'fraction_non_alphanumeric': 0.09675952245594087, 'fraction_numerical': 0.0036384309266628764, 'mean_word_length': 1.2640926640926642, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '512779', 'n_tokens_mistral': 2400, 'n_tokens_neox': 2256, 'n_words': 560}