Widen check_multiprocess_shutdown_event type hint via Protocol (closes #247)

src/hypercorn/utils.py

@@ -7,9 +7,8 @@
 from collections.abc import Awaitable, Callable, Iterable
 from enum import Enum
 from importlib import import_module
-from multiprocessing.synchronize import Event as EventType
 from pathlib import Path
-from typing import Any, cast, Literal, TYPE_CHECKING
+from typing import Any, cast, Literal, Protocol, TYPE_CHECKING
 
 from .app_wrappers import ASGIWrapper, WSGIWrapper
 from .config import Config
@@ -154,8 +153,22 @@ async def raise_shutdown(shutdown_event: Callable[..., Awaitable]) -> None:
     raise ShutdownError()
 
 
+class IsSetEvent(Protocol):
+    """Anything that exposes a non-async ``is_set()`` checkpoint.
+
+    Widens the previous ``multiprocessing.synchronize.Event`` annotation so
+    that callers can pass the equally-valid alternatives that only require
+    ``is_set()`` here — notably ``multiprocessing.Manager().Event()`` (a
+    proxy wrapping ``threading.Event``) and plain ``threading.Event`` for
+    in-process testing. See https://github.com/pgjones/hypercorn/issues/247.
+    """
+
+    def is_set(self) -> bool:
+        pass
+
+
 async def check_multiprocess_shutdown_event(
-    shutdown_event: EventType, sleep: Callable[[float], Awaitable[Any]]
+    shutdown_event: IsSetEvent, sleep: Callable[[float], Awaitable[Any]]
 ) -> None:
     while True:
         if shutdown_event.is_set():

tests/test_utils.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import asyncio
+import multiprocessing
+import threading
 from collections.abc import Callable, Iterable
 from typing import Any
 
@@ -8,6 +11,7 @@
 from hypercorn.typing import Scope
 from hypercorn.utils import (
     build_and_validate_headers,
+    check_multiprocess_shutdown_event,
     filter_pseudo_headers,
     is_asgi,
     suppress_body,
@@ -80,3 +84,31 @@ def test_filter_pseudo_headers_no_authority() -> None:
         [(b"host", b"quart"), (b":path", b"/"), (b"user-agent", b"something")]
     )
     assert result == [(b"host", b"quart"), (b"user-agent", b"something")]
+
+
+@pytest.mark.parametrize(
+    "make_event",
+    [
+        # multiprocessing.Event() — the originally hinted concrete type.
+        pytest.param(lambda: multiprocessing.Event(), id="multiprocessing.Event"),
+        # Manager().Event() — a proxy wrapping threading.Event; valid per
+        # https://docs.python.org/3/library/multiprocessing.html#multiprocessing.managers.SyncManager.Event
+        pytest.param(lambda: multiprocessing.Manager().Event(), id="multiprocessing.Manager.Event"),
+        # threading.Event — equally usable in single-process tests.
+        pytest.param(lambda: threading.Event(), id="threading.Event"),
+    ],
+)
+def test_check_multiprocess_shutdown_event_accepts_is_set_protocol(make_event: Callable) -> None:
+    """Regression test for https://github.com/pgjones/hypercorn/issues/247.
+
+    ``check_multiprocess_shutdown_event`` should accept anything that exposes a
+    synchronous ``.is_set()`` method, not only ``multiprocessing.synchronize.Event``.
+    """
+    event = make_event()
+    event.set()
+
+    async def call() -> None:
+        await check_multiprocess_shutdown_event(event, asyncio.sleep)
+
+    # Should return immediately because the event is already set.
+    asyncio.run(asyncio.wait_for(call(), timeout=1.0))