Added test and changelog entry for #837

docs/versionhistory.rst

@@ -3,6 +3,13 @@ Version history
 
 This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
+**UNRELEASED**
+
+- Fixed cancellation edge case on asyncio where a task spawning another with
+  ``TaskGroup.start()`` is not protected from external cancellation even when the
+  subtask has not yet called ``task_status.started()`` and is in a shielded cancel scope
+  (`#837 <https://github.com/agronholm/anyio/issues/837>`_)
+
 **4.7.0**
 
 - Updated ``TaskGroup`` to work with asyncio's eager task factories

tests/test_taskgroups.py

@@ -18,6 +18,7 @@
 from anyio import (
     TASK_STATUS_IGNORED,
     CancelScope,
+    Event,
     create_task_group,
     current_effective_deadline,
     current_time,
@@ -1602,6 +1603,40 @@ async def in_task_group(task_status: TaskStatus[None]) -> None:
     assert not tg.cancel_scope.cancel_called
 
 
+async def test_cancel_shielding_start() -> None:
+    """
+    Test that if the host task that has spawned a subtask via ``start()`` is cancelled,
+    a shielded cancel scope in the child task will shield it from cancellation.
+
+    Regression test for #837.
+
+    """
+
+    async def taskfunc(*, task_status: TaskStatus[None]) -> None:
+        with CancelScope(shield=True):
+            entered_inner_scope.set()
+            try:
+                await checkpoint()
+                await checkpoint()
+            except get_cancelled_exc_class():
+                pytest.fail("Shouldn't be cancelled in a shielded scope")
+
+        # The cancellation should be triggered here, and not any earlier
+        await checkpoint()
+
+    async def start_inner_task() -> None:
+        await inner_tg.start(taskfunc)
+
+    entered_inner_scope = Event()
+    async with (
+        create_task_group() as tg,
+        create_task_group() as inner_tg,
+    ):
+        tg.start_soon(start_inner_task)
+        await entered_inner_scope.wait()
+        tg.cancel_scope.cancel()
+
+
 if sys.version_info <= (3, 11):
 
     def no_other_refs() -> list[object]: