pythongh-115957: Close coroutine if the TaskGroup has been closed

Doc/library/asyncio-task.rst

@@ -335,6 +335,10 @@ and reliable way to wait for all tasks in the group to finish.
       Create a task in this task group.
       The signature matches that of :func:`asyncio.create_task`.
 
+      .. versionchanged:: 3.13
+
+         Close the given coroutine if the task group has been closed.
+
 Example::
 
     async def main():

Lib/asyncio/taskgroups.py

@@ -154,10 +154,13 @@ def create_task(self, coro, *, name=None, context=None):
         Similar to `asyncio.create_task`.
         """
         if not self._entered:
+            coro.close()
             raise RuntimeError(f"TaskGroup {self!r} has not been entered")
         if self._exiting and not self._tasks:
+            coro.close()
             raise RuntimeError(f"TaskGroup {self!r} is finished")
         if self._aborting:
+            coro.close()
             raise RuntimeError(f"TaskGroup {self!r} is shutting down")
         if context is None:
             task = self._loop.create_task(coro, name=name)

Lib/test/test_asyncio/test_taskgroups.py

@@ -738,10 +738,7 @@ async def coro2(g):
                 await asyncio.sleep(1)
             except asyncio.CancelledError:
                 with self.assertRaises(RuntimeError):
-                    g.create_task(c1 := coro1())
-                # We still have to await c1 to avoid a warning
-                with self.assertRaises(ZeroDivisionError):
-                    await c1
+                    g.create_task(coro1())
 
         with self.assertRaises(ExceptionGroup) as cm:
             async with taskgroups.TaskGroup() as g:
@@ -803,16 +800,12 @@ async def test_taskgroup_finished(self):
         coro = asyncio.sleep(0)
         with self.assertRaisesRegex(RuntimeError, "is finished"):
             tg.create_task(coro)
-        # We still have to await coro to avoid a warning
-        await coro
 
     async def test_taskgroup_not_entered(self):
         tg = taskgroups.TaskGroup()
         coro = asyncio.sleep(0)
         with self.assertRaisesRegex(RuntimeError, "has not been entered"):
             tg.create_task(coro)
-        # We still have to await coro to avoid a warning
-        await coro
 
     async def test_taskgroup_without_parent_task(self):
         tg = taskgroups.TaskGroup()
@@ -821,8 +814,16 @@ async def test_taskgroup_without_parent_task(self):
         coro = asyncio.sleep(0)
         with self.assertRaisesRegex(RuntimeError, "has not been entered"):
             tg.create_task(coro)
-        # We still have to await coro to avoid a warning
-        await coro
+
+    def test_coro_closed_when_tg_closed(self):
+        async def run_coro_after_tg_closes():
+            async with taskgroups.TaskGroup() as tg:
+                pass
+            coro = asyncio.sleep(0)
+            with self.assertRaisesRegex(RuntimeError, "is finished"):
+                tg.create_task(coro)
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(run_coro_after_tg_closes())
 
 
 if __name__ == "__main__":