Fixed cancel scope crash without a current task on the asyncio backend

docs/versionhistory.rst

@@ -57,6 +57,10 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   ``anyio.run()``; the options are now passed as keyword arguments to ``trio.run()``
   again, as documented (a regression from AnyIO 3)
   (`#1161 <https://github.com/agronholm/anyio/pull/1161>`_; PR by @Zac-HD)
+- Fixed ``TypeError: cannot create weak reference to 'NoneType' object`` on the asyncio
+  backend when a cancel scope or ``cancel_shielded_checkpoint()`` was used while
+  ``current_task()`` returned ``None``
+  (`#1163 <https://github.com/agronholm/anyio/pull/1163>`_; PR by @inoue22)
 
 **4.13.0**
 

src/anyio/_backends/_asyncio.py

@@ -416,7 +416,11 @@ def __enter__(self) -> CancelScope:
                 "Each CancelScope may only be used for a single 'with' block"
             )
 
-        self._host_task = host_task = cast(asyncio.Task, current_task())
+        host_task = current_task()
+        if host_task is None:
+            raise RuntimeError("A cancel scope can only be entered from within a task")
+
+        self._host_task = host_task
         self._tasks.add(host_task)
         try:
             task_state = _task_states[host_task]
@@ -2424,6 +2428,10 @@ async def checkpoint_if_cancelled(cls) -> None:
 
     @classmethod
     async def cancel_shielded_checkpoint(cls) -> None:
+        if current_task() is None:
+            await sleep(0)
+            return
+
         with CancelScope(shield=True):
             await sleep(0)
 

src/anyio/_core/_tasks.py

@@ -46,6 +46,8 @@ class CancelScope:
     :param shield: ``True`` to shield the cancel scope from external cancellation
     :raises NoEventLoopError: if no supported asynchronous event loop is running in the
         current thread
+    :raises RuntimeError: if the scope is entered while there is no current task in the
+        running event loop
     """
 
     def __new__(

tests/test_lowlevel.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 
 import pytest
 
 from anyio import create_task_group, run
+from anyio._backends import _asyncio
+from anyio._backends._asyncio import AsyncIOBackend, CancelScope
 from anyio.lowlevel import (
     RunVar,
     cancel_shielded_checkpoint,
@@ -86,6 +89,43 @@ async def second_func() -> None:
     assert second_finished
 
 
+def test_cancel_scope_without_running_task(
+    asyncio_event_loop: asyncio.AbstractEventLoop,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Entering a cancel scope without a current task raises a clear error (asyncio).
+
+    Regression test: previously this raised an obscure
+    ``TypeError: cannot create weak reference to 'NoneType' object`` because
+    ``current_task()`` returned ``None`` and that ``None`` was then used as a key in
+    a ``WeakKeyDictionary``.
+    """
+    monkeypatch.setattr(_asyncio, "current_task", lambda: None)
+
+    async def main() -> None:
+        with pytest.raises(
+            RuntimeError, match="A cancel scope can only be entered from within a task"
+        ):
+            with CancelScope():
+                pass
+
+    asyncio_event_loop.run_until_complete(main())
+
+
+def test_cancel_shielded_checkpoint_without_task(
+    asyncio_event_loop: asyncio.AbstractEventLoop,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """cancel_shielded_checkpoint() must not crash without a current task (asyncio).
+
+    Library primitives such as ``CapacityLimiter`` reach this code path while
+    ``current_task()`` may return ``None``; it must degrade to a plain checkpoint
+    instead of raising ``TypeError``.
+    """
+    monkeypatch.setattr(_asyncio, "current_task", lambda: None)
+    asyncio_event_loop.run_until_complete(AsyncIOBackend.cancel_shielded_checkpoint())
+
+
 class TestRunVar:
     def test_get_set(
         self,