diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py index af36fbd4..7224f39a 100644 --- a/tests/test_taskgroups.py +++ b/tests/test_taskgroups.py @@ -1586,75 +1586,103 @@ async def test_exception_refcycles_direct(self) -> None: Note: This test never failed on anyio, but keeping this test to align with the tests from cpython. """ - tg = create_task_group() - exc = None - class _Done(Exception): - pass + async def test() -> object: + tg = create_task_group() + exc = None - try: - async with tg: - raise _Done - except ExceptionGroup as e: - exc = e + class _Done(Exception): + pass + + try: + async with tg: + raise _Done + except ExceptionGroup as e: + exc = e + + assert exc is not None + try: + return exc + finally: + del exc - assert exc is not None - assert gc.get_referrers(exc) == no_other_refs() + assert gc.get_referrers(await test()) == no_other_refs() async def test_exception_refcycles_errors(self) -> None: """Test that TaskGroup deletes self._exceptions, and __aexit__ args""" - tg = create_task_group() - exc = None - class _Done(Exception): - pass + async def test() -> object: + tg = create_task_group() + exc = None - try: - async with tg: - raise _Done - except ExceptionGroup as excs: - exc = excs.exceptions[0] + class _Done(Exception): + pass + + try: + async with tg: + raise _Done + except ExceptionGroup as excs: + exc = excs.exceptions[0] - assert isinstance(exc, _Done) - assert gc.get_referrers(exc) == no_other_refs() + assert isinstance(exc, _Done) + try: + return exc + finally: + del exc + + assert gc.get_referrers(await test()) == no_other_refs() async def test_exception_refcycles_parent_task(self) -> None: """Test that TaskGroup's cancel_scope deletes self._host_task""" - tg = create_task_group() - exc = None - class _Done(Exception): - pass + async def test() -> object: + tg = create_task_group() + exc = None - async def coro_fn() -> None: - async with tg: - raise _Done + class _Done(Exception): + pass - try: - async with anyio.create_task_group() as tg2: - tg2.start_soon(coro_fn) - except ExceptionGroup as excs: - exc = excs.exceptions[0].exceptions[0] + async def coro_fn() -> None: + async with tg: + raise _Done + + try: + async with anyio.create_task_group() as tg2: + tg2.start_soon(coro_fn) + except ExceptionGroup as excs: + exc = excs.exceptions[0].exceptions[0] + + assert isinstance(exc, _Done) + try: + return exc + finally: + del exc - assert isinstance(exc, _Done) - assert gc.get_referrers(exc) == no_other_refs() + assert gc.get_referrers(await test()) == no_other_refs() async def test_exception_refcycles_propagate_cancellation_error(self) -> None: """Test that TaskGroup deletes cancelled_exc""" - tg = anyio.create_task_group() - exc = None - with CancelScope() as cs: - cs.cancel() + async def test() -> object: + tg = anyio.create_task_group() + exc = None + + with CancelScope() as cs: + cs.cancel() + try: + async with tg: + await checkpoint() + except get_cancelled_exc_class() as e: + exc = e + raise + + assert isinstance(exc, get_cancelled_exc_class()) try: - async with tg: - await checkpoint() - except get_cancelled_exc_class() as e: - exc = e - raise + return exc + finally: + del exc - assert isinstance(exc, get_cancelled_exc_class()) - assert gc.get_referrers(exc) == no_other_refs() + assert gc.get_referrers(await test()) == no_other_refs() async def test_exception_refcycles_base_error(self) -> None: """ @@ -1664,20 +1692,26 @@ async def test_exception_refcycles_base_error(self) -> None: but copied from CPython's asyncio.TaskGroup tests for completion. """ - class MyKeyboardInterrupt(KeyboardInterrupt): - pass + async def test() -> object: + class MyKeyboardInterrupt(KeyboardInterrupt): + pass - tg = create_task_group() - exc = None + tg = create_task_group() + exc = None - try: - async with tg: - raise MyKeyboardInterrupt - except BaseExceptionGroup as excs: - exc = excs.exceptions[0] + try: + async with tg: + raise MyKeyboardInterrupt + except BaseExceptionGroup as excs: + exc = excs.exceptions[0] + + assert isinstance(exc, MyKeyboardInterrupt) + try: + return exc + finally: + del exc - assert isinstance(exc, MyKeyboardInterrupt) - assert gc.get_referrers(exc) == no_other_refs() + assert gc.get_referrers(await test()) == no_other_refs() class TestTaskStatusTyping: