gh-115957: Close coroutine if TaskGroup.create_task() raises an error (#116009)

Doc/library/asyncio-task.rst

@@ -334,6 +334,13 @@ 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`.
+      If the task group is inactive (e.g. not yet entered,
+      already finished, or in the process of shutting down),
+      we will close the given ``coro``.
+
+      .. versionchanged:: 3.13
+
+         Close the given coroutine if the task group is not active.
 
 Example::
 

Doc/whatsnew/3.13.rst

@@ -185,6 +185,12 @@ Other Language Changes
 
   (Contributed by Sebastian Pipping in :gh:`115623`.)
 
+* When :func:`asyncio.TaskGroup.create_task` is called on an inactive
+  :class:`asyncio.TaskGroup`, the given coroutine will be closed (which
+  prevents a :exc:`RuntimeWarning` about the given coroutine being
+  never awaited).
+
+  (Contributed by Arthur Tacca and Jason Zhang in :gh:`115957`.)
 
 New Modules
 ===========

Lib/asyncio/taskgroups.py

@@ -154,10 +154,13 @@ class TaskGroup:
         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

@@ -7,6 +7,7 @@ import contextvars
 import contextlib
 from asyncio import taskgroups
 import unittest
+import warnings
 
 from test.test_asyncio.utils import await_without_task
 
@@ -738,10 +739,7 @@ class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
                 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:
@@ -797,22 +795,25 @@ class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
                 pass
 
     async def test_taskgroup_finished(self):
-        tg = taskgroups.TaskGroup()
-        async with tg:
-            pass
-        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 create_task_after_tg_finish():
+            tg = taskgroups.TaskGroup()
+            async with tg:
+                pass
+            coro = asyncio.sleep(0)
+            with self.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)
+        with warnings.catch_warnings(record=True) as w:
+            await create_task_after_tg_finish()
+        self.assertEqual(len(w), 0)
 
     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 +822,16 @@ class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
         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__":

Misc/NEWS.d/next/Library/2024-03-02-11-31-49.gh-issue-115957.C-3Z_U.rst

@@ -0,0 +1 @@
+When ``asyncio.TaskGroup.create_task`` is called on an inactive ``asyncio.TaskGroup``, the given coroutine will be closed (which prevents a  ``RuntimeWarning``).