asynccontextmanager, AbstractAsyncContextManager,
AsyncExitStack, nullcontext, aclosing, contextmanager)
from test import support
+from test.support import run_no_yield_async_fn as _run_async_fn
import unittest
import traceback
from test.test_contextlib import TestBaseExitStack
-def _run_async_fn(async_fn, /, *args, **kwargs):
- coro = async_fn(*args, **kwargs)
- try:
- coro.send(None)
- except StopIteration as e:
- return e.value
- else:
- raise AssertionError("coroutine did not complete")
- finally:
- coro.close()
-
-
def _async_test(async_fn):
"""Decorator to turn an async function into a synchronous function"""
@functools.wraps(async_fn)
exit_stack = SyncAsyncExitStack
callback_error_internal_frames = [
('__exit__', 'return _run_async_fn(self.__aexit__, *exc_details)'),
- ('_run_async_fn', 'coro.send(None)'),
+ ('run_no_yield_async_fn', 'coro.send(None)'),
('__aexit__', 'raise exc'),
('__aexit__', 'cb_suppress = cb(*exc_details)'),
]
from test.support import cpython_only, import_helper
from test.support import MISSING_C_DOCSTRINGS, ALWAYS_EQ
+from test.support import run_no_yield_async_fn
from test.support.import_helper import DirsOnSysPath, ready_to_import
from test.support.os_helper import TESTFN, temp_cwd
from test.support.script_helper import assert_python_ok, assert_python_failure, kill_python
sys.modules.pop("inspect_actual")
def test_nested_class_definition_inside_async_function(self):
- def run(coro):
- try:
- coro.send(None)
- except StopIteration as e:
- return e.value
- else:
- raise RuntimeError("coroutine did not complete synchronously!")
- finally:
- coro.close()
+ run = run_no_yield_async_fn
- self.assertSourceEqual(run(mod2.func225()), 226, 227)
+ self.assertSourceEqual(run(mod2.func225), 226, 227)
self.assertSourceEqual(mod2.cls226, 231, 235)
- self.assertSourceEqual(run(mod2.cls226().func232()), 233, 234)
+ self.assertSourceEqual(run(mod2.cls226().func232), 233, 234)
def test_class_definition_same_name_diff_methods(self):
self.assertSourceEqual(mod2.cls296, 296, 298)
from test.support.pty_helper import run_pty, FakeInput
from unittest.mock import patch
-# gh-114275: WASI fails to run asyncio tests, similar skip than test_asyncio.
-SKIP_ASYNCIO_TESTS = (not support.has_socket_support)
+SKIP_CORO_TESTS = False
class PdbTestInput(object):
"""
def test_pdb_next_command_for_generator():
- """Testing skip unwindng stack on yield for generators for "next" command
+ """Testing skip unwinding stack on yield for generators for "next" command
>>> def test_gen():
... yield 0
finished
"""
-if not SKIP_ASYNCIO_TESTS:
+if not SKIP_CORO_TESTS:
def test_pdb_next_command_for_coroutine():
- """Testing skip unwindng stack on yield for coroutines for "next" command
+ """Testing skip unwinding stack on yield for coroutines for "next" command
- >>> import asyncio
+ >>> from test.support import run_yielding_async_fn, async_yield
>>> async def test_coro():
- ... await asyncio.sleep(0)
- ... await asyncio.sleep(0)
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
+ ... await async_yield(0)
+ ... await async_yield(0)
>>> async def test_main():
... import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
... await test_coro()
>>> def test_function():
- ... asyncio.run(test_main())
+ ... run_yielding_async_fn(test_main)
... print("finished")
>>> with PdbTestInput(['step',
-> async def test_coro():
(Pdb) step
> <doctest test.test_pdb.test_pdb_next_command_for_coroutine[1]>(2)test_coro()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) next
> <doctest test.test_pdb.test_pdb_next_command_for_coroutine[1]>(3)test_coro()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) next
> <doctest test.test_pdb.test_pdb_next_command_for_coroutine[1]>(4)test_coro()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) next
Internal StopIteration
> <doctest test.test_pdb.test_pdb_next_command_for_coroutine[2]>(3)test_main()
"""
def test_pdb_next_command_for_asyncgen():
- """Testing skip unwindng stack on yield for coroutines for "next" command
+ """Testing skip unwinding stack on yield for coroutines for "next" command
- >>> import asyncio
+ >>> from test.support import run_yielding_async_fn, async_yield
>>> async def agen():
... yield 1
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
... yield 2
>>> async def test_coro():
... await test_coro()
>>> def test_function():
- ... asyncio.run(test_main())
+ ... run_yielding_async_fn(test_main)
... print("finished")
>>> with PdbTestInput(['step',
-> yield 1
(Pdb) next
> <doctest test.test_pdb.test_pdb_next_command_for_asyncgen[1]>(3)agen()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) continue
2
finished
"""
def test_pdb_return_command_for_generator():
- """Testing no unwindng stack on yield for generators
+ """Testing no unwinding stack on yield for generators
for "return" command
>>> def test_gen():
finished
"""
-if not SKIP_ASYNCIO_TESTS:
+if not SKIP_CORO_TESTS:
def test_pdb_return_command_for_coroutine():
- """Testing no unwindng stack on yield for coroutines for "return" command
+ """Testing no unwinding stack on yield for coroutines for "return" command
- >>> import asyncio
+ >>> from test.support import run_yielding_async_fn, async_yield
>>> async def test_coro():
- ... await asyncio.sleep(0)
- ... await asyncio.sleep(0)
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
+ ... await async_yield(0)
+ ... await async_yield(0)
>>> async def test_main():
... import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
... await test_coro()
>>> def test_function():
- ... asyncio.run(test_main())
+ ... run_yielding_async_fn(test_main)
... print("finished")
>>> with PdbTestInput(['step',
-> async def test_coro():
(Pdb) step
> <doctest test.test_pdb.test_pdb_return_command_for_coroutine[1]>(2)test_coro()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) next
> <doctest test.test_pdb.test_pdb_return_command_for_coroutine[1]>(3)test_coro()
- -> await asyncio.sleep(0)
+ -> await async_yield(0)
(Pdb) continue
finished
"""
def test_pdb_until_command_for_generator():
- """Testing no unwindng stack on yield for generators
+ """Testing no unwinding stack on yield for generators
for "until" command if target breakpoint is not reached
>>> def test_gen():
finished
"""
-if not SKIP_ASYNCIO_TESTS:
+if not SKIP_CORO_TESTS:
def test_pdb_until_command_for_coroutine():
- """Testing no unwindng stack for coroutines
+ """Testing no unwinding stack for coroutines
for "until" command if target breakpoint is not reached
- >>> import asyncio
+ >>> from test.support import run_yielding_async_fn, async_yield
>>> async def test_coro():
... print(0)
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
... print(1)
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
... print(2)
- ... await asyncio.sleep(0)
+ ... await async_yield(0)
... print(3)
>>> async def test_main():
... await test_coro()
>>> def test_function():
- ... asyncio.run(test_main())
+ ... run_yielding_async_fn(test_main)
... print("finished")
>>> with PdbTestInput(['step',