test(supervisors): fix occasional assertion failures and hangs

tests/supervisors/test_multiprocess.py

@@ -10,6 +10,7 @@
 
 import pytest
 
+from tests.utils import with_retry
 from uvicorn import Config
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.supervisors import Multiprocess
@@ -76,22 +77,29 @@ def test_multiprocess_run() -> None:
 
 
 @new_console_in_windows
+@with_retry(3)
 def test_multiprocess_health_check() -> None:
     """
     Ensure that the health check works as expected.
     """
     config = Config(app=app, workers=2)
     supervisor = Multiprocess(config, target=run, sockets=[])
     threading.Thread(target=supervisor.run, daemon=True).start()
-    time.sleep(1)
+    time.sleep(0)  # release gil.
+    time.sleep(1)  # ensure server is up.
     process = supervisor.processes[0]
+    assert process.is_alive()
     process.kill()
-    assert not process.is_alive()
-    time.sleep(1)
-    for p in supervisor.processes:
-        assert p.is_alive()
-    supervisor.signal_queue.append(signal.SIGINT)
-    supervisor.join_all()
+    process.join()
+    try:
+        assert not process.is_alive(0.5)
+        time.sleep(0)  # release gil.
+        time.sleep(1)  # ensure process restart complete.
+        for p in supervisor.processes:
+            assert p.is_alive()
+    finally:
+        supervisor.signal_queue.append(signal.SIGINT)
+        supervisor.join_all()
 
 
 @new_console_in_windows

tests/utils.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 import asyncio
+import functools
+import inspect
 import os
 import signal
 import sys
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager, contextmanager
 from pathlib import Path
 from socket import socket
+from typing import Any, Callable
 
 from uvicorn import Config, Server
 
@@ -53,3 +56,36 @@ def get_asyncio_default_loop_per_os() -> type[asyncio.AbstractEventLoop]:
         return asyncio.ProactorEventLoop  # type: ignore  # pragma: nocover
     else:
         return asyncio.SelectorEventLoop  # pragma: nocover
+
+
+def with_retry(retry_count: int = 1) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        if inspect.iscoroutinefunction(func):
+
+            @functools.wraps(func)  # pragma: nocover
+            async def async_wrapper(*args, **kwargs):
+                for attempt in range(retry_count):
+                    try:
+                        return await func(*args, **kwargs)
+                    except Exception as e:
+                        if attempt == retry_count - 1:
+                            raise e
+                return None
+
+            return async_wrapper  # pragma: nocover
+
+        else:
+            # Maintain the original calling method of the test case, e.g. test_multiprocess_health_check.
+            @functools.wraps(func)  # pragma: nocover
+            def sync_wrapper(*args, **kwargs):
+                for attempt in range(retry_count):
+                    try:
+                        return func(*args, **kwargs)
+                    except Exception as e:
+                        if attempt == retry_count - 1:
+                            raise e
+                return None
+
+            return sync_wrapper
+
+    return decorator