test_taskgroups.py

# Adapted with permission from the EdgeDB project;
# license: PSFL.

importweakref
importsys
importgc
importasyncio
importcontextvars
importcontextlib
fromasyncioimporttaskgroups
importunittest
importwarnings

fromtest.test_asyncio.utilsimportawait_without_task

# To prevent a warning "test altered the execution environment"
deftearDownModule():
asyncio.set_event_loop_policy(None)


classMyExc(Exception):
pass


classMyBaseExc(BaseException):
pass


defget_error_types(eg):
return {type(exc) forexcineg.exceptions}


defset_gc_state(enabled):
was_enabled=gc.isenabled()
ifenabled:
gc.enable()
else:
gc.disable()
returnwas_enabled


@contextlib.contextmanager
defdisable_gc():
was_enabled=set_gc_state(enabled=False)
try:
yield
finally:
set_gc_state(enabled=was_enabled)


classBaseTestTaskGroup:

asyncdeftest_taskgroup_01(self):

asyncdeffoo1():
awaitasyncio.sleep(0.1)
return42

asyncdeffoo2():
awaitasyncio.sleep(0.2)
return11

asyncwithtaskgroups.TaskGroup() asg:
t1=g.create_task(foo1())
t2=g.create_task(foo2())

self.assertEqual(t1.result(), 42)
self.assertEqual(t2.result(), 11)

asyncdeftest_taskgroup_02(self):

asyncdeffoo1():
awaitasyncio.sleep(0.1)
return42

asyncdeffoo2():
awaitasyncio.sleep(0.2)
return11

asyncwithtaskgroups.TaskGroup() asg:
t1=g.create_task(foo1())
awaitasyncio.sleep(0.15)
t2=g.create_task(foo2())

self.assertEqual(t1.result(), 42)
self.assertEqual(t2.result(), 11)

asyncdeftest_taskgroup_03(self):

asyncdeffoo1():
awaitasyncio.sleep(1)
return42

asyncdeffoo2():
awaitasyncio.sleep(0.2)
return11

asyncwithtaskgroups.TaskGroup() asg:
t1=g.create_task(foo1())
awaitasyncio.sleep(0.15)
# cancel t1 explicitly, i.e. everything should continue
# working as expected.
t1.cancel()

t2=g.create_task(foo2())

self.assertTrue(t1.cancelled())
self.assertEqual(t2.result(), 11)

asyncdeftest_taskgroup_04(self):

NUM=0
t2_cancel=False
t2=None

asyncdeffoo1():
awaitasyncio.sleep(0.1)
1/0

asyncdeffoo2():
nonlocalNUM, t2_cancel
try:
awaitasyncio.sleep(1)
exceptasyncio.CancelledError:
t2_cancel=True
raise
NUM+=1

asyncdefrunner():
nonlocalNUM, t2

asyncwithtaskgroups.TaskGroup() asg:
g.create_task(foo1())
t2=g.create_task(foo2())

NUM+=10

withself.assertRaises(ExceptionGroup) ascm:
awaitasyncio.create_task(runner())

self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})

self.assertEqual(NUM, 0)
self.assertTrue(t2_cancel)
self.assertTrue(t2.cancelled())

asyncdeftest_cancel_children_on_child_error(self):
# When a child task raises an error, the rest of the children
# are cancelled and the errors are gathered into an EG.

NUM=0
t2_cancel=False
runner_cancel=False

asyncdeffoo1():
awaitasyncio.sleep(0.1)
1/0

asyncdeffoo2():
nonlocalNUM, t2_cancel
try:
awaitasyncio.sleep(5)
exceptasyncio.CancelledError:
t2_cancel=True
raise
NUM+=1

asyncdefrunner():
nonlocalNUM, runner_cancel

asyncwithtaskgroups.TaskGroup() asg:
g.create_task(foo1())
g.create_task(foo1())
g.create_task(foo1())
g.create_task(foo2())
try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
runner_cancel=True
raise

NUM+=10

# The 3 foo1 sub tasks can be racy when the host is busy - if the
# cancellation happens in the middle, we'll see partial sub errors here
withself.assertRaises(ExceptionGroup) ascm:
awaitasyncio.create_task(runner())

self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})
self.assertEqual(NUM, 0)
self.assertTrue(t2_cancel)
self.assertTrue(runner_cancel)

asyncdeftest_cancellation(self):

NUM=0

asyncdeffoo():
nonlocalNUM
try:
awaitasyncio.sleep(5)
exceptasyncio.CancelledError:
NUM+=1
raise

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
for_inrange(5):
g.create_task(foo())

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(asyncio.CancelledError) ascm:
awaitr

self.assertEqual(NUM, 5)

asyncdeftest_taskgroup_07(self):

NUM=0

asyncdeffoo():
nonlocalNUM
try:
awaitasyncio.sleep(5)
exceptasyncio.CancelledError:
NUM+=1
raise

asyncdefrunner():
nonlocalNUM
asyncwithtaskgroups.TaskGroup() asg:
for_inrange(5):
g.create_task(foo())

try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
NUM+=10
raise

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(asyncio.CancelledError):
awaitr

self.assertEqual(NUM, 15)

asyncdeftest_taskgroup_08(self):

asyncdeffoo():
try:
awaitasyncio.sleep(10)
finally:
1/0

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
for_inrange(5):
g.create_task(foo())

awaitasyncio.sleep(10)

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(ExceptionGroup) ascm:
awaitr
self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})

asyncdeftest_taskgroup_09(self):

t1=t2=None

asyncdeffoo1():
awaitasyncio.sleep(1)
return42

asyncdeffoo2():
awaitasyncio.sleep(2)
return11

asyncdefrunner():
nonlocalt1, t2
asyncwithtaskgroups.TaskGroup() asg:
t1=g.create_task(foo1())
t2=g.create_task(foo2())
awaitasyncio.sleep(0.1)
1/0

try:
awaitrunner()
exceptExceptionGroupast:
self.assertEqual(get_error_types(t), {ZeroDivisionError})
else:
self.fail('ExceptionGroup was not raised')

self.assertTrue(t1.cancelled())
self.assertTrue(t2.cancelled())

asyncdeftest_taskgroup_10(self):

t1=t2=None

asyncdeffoo1():
awaitasyncio.sleep(1)
return42

asyncdeffoo2():
awaitasyncio.sleep(2)
return11

asyncdefrunner():
nonlocalt1, t2
asyncwithtaskgroups.TaskGroup() asg:
t1=g.create_task(foo1())
t2=g.create_task(foo2())
1/0

try:
awaitrunner()
exceptExceptionGroupast:
self.assertEqual(get_error_types(t), {ZeroDivisionError})
else:
self.fail('ExceptionGroup was not raised')

self.assertTrue(t1.cancelled())
self.assertTrue(t2.cancelled())

asyncdeftest_taskgroup_11(self):

asyncdeffoo():
try:
awaitasyncio.sleep(10)
finally:
1/0

asyncdefrunner():
asyncwithtaskgroups.TaskGroup():
asyncwithtaskgroups.TaskGroup() asg2:
for_inrange(5):
g2.create_task(foo())

awaitasyncio.sleep(10)

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(ExceptionGroup) ascm:
awaitr

self.assertEqual(get_error_types(cm.exception), {ExceptionGroup})
self.assertEqual(get_error_types(cm.exception.exceptions[0]), {ZeroDivisionError})

asyncdeftest_taskgroup_12(self):

asyncdeffoo():
try:
awaitasyncio.sleep(10)
finally:
1/0

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg1:
g1.create_task(asyncio.sleep(10))

asyncwithtaskgroups.TaskGroup() asg2:
for_inrange(5):
g2.create_task(foo())

awaitasyncio.sleep(10)

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(ExceptionGroup) ascm:
awaitr

self.assertEqual(get_error_types(cm.exception), {ExceptionGroup})
self.assertEqual(get_error_types(cm.exception.exceptions[0]), {ZeroDivisionError})

asyncdeftest_taskgroup_13(self):

asyncdefcrash_after(t):
awaitasyncio.sleep(t)
raiseValueError(t)

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg1:
g1.create_task(crash_after(0.1))

asyncwithtaskgroups.TaskGroup() asg2:
g2.create_task(crash_after(10))

r=asyncio.create_task(runner())
withself.assertRaises(ExceptionGroup) ascm:
awaitr

self.assertEqual(get_error_types(cm.exception), {ValueError})

asyncdeftest_taskgroup_14(self):

asyncdefcrash_after(t):
awaitasyncio.sleep(t)
raiseValueError(t)

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg1:
g1.create_task(crash_after(10))

asyncwithtaskgroups.TaskGroup() asg2:
g2.create_task(crash_after(0.1))

r=asyncio.create_task(runner())
withself.assertRaises(ExceptionGroup) ascm:
awaitr

self.assertEqual(get_error_types(cm.exception), {ExceptionGroup})
self.assertEqual(get_error_types(cm.exception.exceptions[0]), {ValueError})

asyncdeftest_taskgroup_15(self):

asyncdefcrash_soon():
awaitasyncio.sleep(0.3)
1/0

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg1:
g1.create_task(crash_soon())
try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
awaitasyncio.sleep(0.5)
raise

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(ExceptionGroup) ascm:
awaitr
self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})

asyncdeftest_taskgroup_16(self):

asyncdefcrash_soon():
awaitasyncio.sleep(0.3)
1/0

asyncdefnested_runner():
asyncwithtaskgroups.TaskGroup() asg1:
g1.create_task(crash_soon())
try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
awaitasyncio.sleep(0.5)
raise

asyncdefrunner():
t=asyncio.create_task(nested_runner())
awaitt

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(ExceptionGroup) ascm:
awaitr
self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})

asyncdeftest_taskgroup_17(self):
NUM=0

asyncdefrunner():
nonlocalNUM
asyncwithtaskgroups.TaskGroup():
try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
NUM+=10
raise

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()
withself.assertRaises(asyncio.CancelledError):
awaitr

self.assertEqual(NUM, 10)

asyncdeftest_taskgroup_18(self):
NUM=0

asyncdefrunner():
nonlocalNUM
asyncwithtaskgroups.TaskGroup():
try:
awaitasyncio.sleep(10)
exceptasyncio.CancelledError:
NUM+=10
# This isn't a good idea, but we have to support
# this weird case.
raiseMyExc

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.1)

self.assertFalse(r.done())
r.cancel()

try:
awaitr
exceptExceptionGroupast:
self.assertEqual(get_error_types(t),{MyExc})
else:
self.fail('ExceptionGroup was not raised')

self.assertEqual(NUM, 10)

asyncdeftest_taskgroup_19(self):
asyncdefcrash_soon():
awaitasyncio.sleep(0.1)
1/0

asyncdefnested():
try:
awaitasyncio.sleep(10)
finally:
raiseMyExc

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(crash_soon())
awaitnested()

r=asyncio.create_task(runner())
try:
awaitr
exceptExceptionGroupast:
self.assertEqual(get_error_types(t), {MyExc, ZeroDivisionError})
else:
self.fail('TasgGroupError was not raised')

asyncdeftest_taskgroup_20(self):
asyncdefcrash_soon():
awaitasyncio.sleep(0.1)
1/0

asyncdefnested():
try:
awaitasyncio.sleep(10)
finally:
raiseKeyboardInterrupt

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(crash_soon())
awaitnested()

withself.assertRaises(KeyboardInterrupt):
awaitrunner()

asyncdeftest_taskgroup_20a(self):
asyncdefcrash_soon():
awaitasyncio.sleep(0.1)
1/0

asyncdefnested():
try:
awaitasyncio.sleep(10)
finally:
raiseMyBaseExc

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(crash_soon())
awaitnested()

withself.assertRaises(BaseExceptionGroup) ascm:
awaitrunner()

self.assertEqual(
get_error_types(cm.exception), {MyBaseExc, ZeroDivisionError}
 )

asyncdef_test_taskgroup_21(self):
# This test doesn't work as asyncio, currently, doesn't
# correctly propagate KeyboardInterrupt (or SystemExit) --
# those cause the event loop itself to crash.
# (Compare to the previous (passing) test -- that one raises
# a plain exception but raises KeyboardInterrupt in nested();
# this test does it the other way around.)

asyncdefcrash_soon():
awaitasyncio.sleep(0.1)
raiseKeyboardInterrupt

asyncdefnested():
try:
awaitasyncio.sleep(10)
finally:
raiseTypeError

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(crash_soon())
awaitnested()

withself.assertRaises(KeyboardInterrupt):
awaitrunner()

asyncdeftest_taskgroup_21a(self):

asyncdefcrash_soon():
awaitasyncio.sleep(0.1)
raiseMyBaseExc

asyncdefnested():
try:
awaitasyncio.sleep(10)
finally:
raiseTypeError

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(crash_soon())
awaitnested()

withself.assertRaises(BaseExceptionGroup) ascm:
awaitrunner()

self.assertEqual(get_error_types(cm.exception), {MyBaseExc, TypeError})

asyncdeftest_taskgroup_22(self):

asyncdeffoo1():
awaitasyncio.sleep(1)
return42

asyncdeffoo2():
awaitasyncio.sleep(2)
return11

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(foo1())
g.create_task(foo2())

r=asyncio.create_task(runner())
awaitasyncio.sleep(0.05)
r.cancel()

withself.assertRaises(asyncio.CancelledError):
awaitr

asyncdeftest_taskgroup_23(self):

asyncdefdo_job(delay):
awaitasyncio.sleep(delay)

asyncwithtaskgroups.TaskGroup() asg:
forcountinrange(10):
awaitasyncio.sleep(0.1)
g.create_task(do_job(0.3))
ifcount==5:
self.assertLess(len(g._tasks), 5)
awaitasyncio.sleep(1.35)
self.assertEqual(len(g._tasks), 0)

asyncdeftest_taskgroup_24(self):

asyncdefroot(g):
awaitasyncio.sleep(0.1)
g.create_task(coro1(0.1))
g.create_task(coro1(0.2))

asyncdefcoro1(delay):
awaitasyncio.sleep(delay)

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(root(g))

awaitrunner()

asyncdeftest_taskgroup_25(self):
nhydras=0

asyncdefhydra(g):
nonlocalnhydras
nhydras+=1
awaitasyncio.sleep(0.01)
g.create_task(hydra(g))
g.create_task(hydra(g))

asyncdefhercules():
whilenhydras<10:
awaitasyncio.sleep(0.015)
1/0

asyncdefrunner():
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(hydra(g))
g.create_task(hercules())

withself.assertRaises(ExceptionGroup) ascm:
awaitrunner()

self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})
self.assertGreaterEqual(nhydras, 10)

asyncdeftest_taskgroup_task_name(self):
asyncdefcoro():
awaitasyncio.sleep(0)
asyncwithtaskgroups.TaskGroup() asg:
t=g.create_task(coro(), name="yolo")
self.assertEqual(t.get_name(), "yolo")

asyncdeftest_taskgroup_task_context(self):
cvar=contextvars.ContextVar('cvar')

asyncdefcoro(val):
awaitasyncio.sleep(0)
cvar.set(val)

asyncwithtaskgroups.TaskGroup() asg:
ctx=contextvars.copy_context()
self.assertIsNone(ctx.get(cvar))
t1=g.create_task(coro(1), context=ctx)
awaitt1
self.assertEqual(1, ctx.get(cvar))
t2=g.create_task(coro(2), context=ctx)
awaitt2
self.assertEqual(2, ctx.get(cvar))

asyncdeftest_taskgroup_no_create_task_after_failure(self):
asyncdefcoro1():
awaitasyncio.sleep(0.001)
1/0
asyncdefcoro2(g):
try:
awaitasyncio.sleep(1)
exceptasyncio.CancelledError:
withself.assertRaises(RuntimeError):
g.create_task(coro1())

withself.assertRaises(ExceptionGroup) ascm:
asyncwithtaskgroups.TaskGroup() asg:
g.create_task(coro1())
g.create_task(coro2(g))

self.assertEqual(get_error_types(cm.exception), {ZeroDivisionError})

asyncdeftest_taskgroup_context_manager_exit_raises(self):
# See https://github.com/python/cpython/issues/95289
classCustomException(Exception):
pass

asyncdefraise_exc():
raiseCustomException

@contextlib.asynccontextmanager
asyncdefdatabase():
try:
yield
finally:
raiseCustomException

asyncdefmain():
task=asyncio.current_task()
try:
asyncwithtaskgroups.TaskGroup() astg:
asyncwithdatabase():
tg.create_task(raise_exc())
awaitasyncio.sleep(1)
 except* CustomExceptionaserr:
self.assertEqual(task.cancelling(), 0)
self.assertEqual(len(err.exceptions), 2)

else:
self.fail('CustomException not raised')

awaitasyncio.create_task(main())

asyncdeftest_taskgroup_already_entered(self):
tg=taskgroups.TaskGroup()
asyncwithtg:
withself.assertRaisesRegex(RuntimeError, "has already been entered"):
asyncwithtg:
pass

asyncdeftest_taskgroup_double_enter(self):
tg=taskgroups.TaskGroup()
asyncwithtg:
pass
withself.assertRaisesRegex(RuntimeError, "has already been entered"):
asyncwithtg:
pass

asyncdeftest_taskgroup_finished(self):
asyncdefcreate_task_after_tg_finish():
tg=taskgroups.TaskGroup()
asyncwithtg:
pass
coro=asyncio.sleep(0)
withself.assertRaisesRegex(RuntimeError, "is finished"):
tg.create_task(coro)

# Make sure the coroutine was closed when submitted to the inactive tg
# (if not closed, a RuntimeWarning should have been raised)
withwarnings.catch_warnings(record=True) asw:
awaitcreate_task_after_tg_finish()
self.assertEqual(len(w), 0)

asyncdeftest_taskgroup_not_entered(self):
tg=taskgroups.TaskGroup()
coro=asyncio.sleep(0)
withself.assertRaisesRegex(RuntimeError, "has not been entered"):
tg.create_task(coro)

asyncdeftest_taskgroup_without_parent_task(self):
tg=taskgroups.TaskGroup()
withself.assertRaisesRegex(RuntimeError, "parent task"):
awaitawait_without_task(tg.__aenter__())
coro=asyncio.sleep(0)
withself.assertRaisesRegex(RuntimeError, "has not been entered"):
tg.create_task(coro)

asyncdeftest_coro_closed_when_tg_closed(self):
asyncdefrun_coro_after_tg_closes():
asyncwithtaskgroups.TaskGroup() astg:
pass
coro=asyncio.sleep(0)
withself.assertRaisesRegex(RuntimeError, "is finished"):
tg.create_task(coro)

awaitrun_coro_after_tg_closes()

asyncdeftest_cancelling_level_preserved(self):
asyncdefraise_after(t, e):
awaitasyncio.sleep(t)
raisee()

try:
asyncwithasyncio.TaskGroup() astg:
tg.create_task(raise_after(0.0, RuntimeError))
 except* RuntimeError:
pass
self.assertEqual(asyncio.current_task().cancelling(), 0)

asyncdeftest_nested_groups_both_cancelled(self):
asyncdefraise_after(t, e):
awaitasyncio.sleep(t)
raisee()

try:
asyncwithasyncio.TaskGroup() asouter_tg:
try:
asyncwithasyncio.TaskGroup() asinner_tg:
inner_tg.create_task(raise_after(0, RuntimeError))
outer_tg.create_task(raise_after(0, ValueError))
 except* RuntimeError:
pass
else:
self.fail("RuntimeError not raised")
self.assertEqual(asyncio.current_task().cancelling(), 1)
 except* ValueError:
pass
else:
self.fail("ValueError not raised")
self.assertEqual(asyncio.current_task().cancelling(), 0)

asyncdeftest_error_and_cancel(self):
event=asyncio.Event()

asyncdefraise_error():
event.set()
awaitasyncio.sleep(0)
raiseRuntimeError()

asyncdefinner():
try:
asyncwithtaskgroups.TaskGroup() astg:
tg.create_task(raise_error())
awaitasyncio.sleep(1)
self.fail("Sleep in group should have been cancelled")
 except* RuntimeError:
self.assertEqual(asyncio.current_task().cancelling(), 1)
self.assertEqual(asyncio.current_task().cancelling(), 1)
awaitasyncio.sleep(1)
self.fail("Sleep after group should have been cancelled")

asyncdefouter():
t=asyncio.create_task(inner())
awaitevent.wait()
self.assertEqual(t.cancelling(), 0)
t.cancel()
self.assertEqual(t.cancelling(), 1)
withself.assertRaises(asyncio.CancelledError):
awaitt
self.assertTrue(t.cancelled())

awaitouter()

asyncdeftest_exception_refcycles_direct(self):
"""Test that TaskGroup doesn't keep a reference to the raised ExceptionGroup"""
tg=asyncio.TaskGroup()
exc=None

class_Done(Exception):
pass

try:
asyncwithtg:
raise_Done
exceptExceptionGroupase:
exc=e

self.assertIsNotNone(exc)
self.assertListEqual(gc.get_referrers(exc), [])


asyncdeftest_exception_refcycles_errors(self):
"""Test that TaskGroup deletes self._errors, and __aexit__ args"""
tg=asyncio.TaskGroup()
exc=None

class_Done(Exception):
pass

try:
asyncwithtg:
raise_Done
 except* _Doneasexcs:
exc=excs.exceptions[0]

self.assertIsInstance(exc, _Done)
self.assertListEqual(gc.get_referrers(exc), [])


asyncdeftest_exception_refcycles_parent_task(self):
"""Test that TaskGroup deletes self._parent_task"""
tg=asyncio.TaskGroup()
exc=None

class_Done(Exception):
pass

asyncdefcoro_fn():
asyncwithtg:
raise_Done

try:
asyncwithasyncio.TaskGroup() astg2:
tg2.create_task(coro_fn())
 except* _Doneasexcs:
exc=excs.exceptions[0].exceptions[0]

self.assertIsInstance(exc, _Done)
self.assertListEqual(gc.get_referrers(exc), [])


asyncdeftest_exception_refcycles_parent_task_wr(self):
"""Test that TaskGroup deletes self._parent_task and create_task() deletes task"""
tg=asyncio.TaskGroup()
exc=None

class_Done(Exception):
pass

asyncdefcoro_fn():
asyncwithtg:
raise_Done

withdisable_gc():
try:
asyncwithasyncio.TaskGroup() astg2:
task_wr=weakref.ref(tg2.create_task(coro_fn()))
 except* _Doneasexcs:
exc=excs.exceptions[0].exceptions[0]

self.assertIsNone(task_wr())
self.assertIsInstance(exc, _Done)
self.assertListEqual(gc.get_referrers(exc), [])