coverage

tests/test_workers.py

@@ -2,19 +2,30 @@
 
 import signal
 import subprocess
+import sys
 import tempfile
 import time
-from ssl import SSLContext
-from typing import IO, TYPE_CHECKING, Generator
+from typing import TYPE_CHECKING
 
 import httpx
 import pytest
-from gunicorn.arbiter import Arbiter
-
-import uvicorn_worker._workers
 
 if TYPE_CHECKING:
-    from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, LifespanStartupFailedEvent, Scope
+    from ssl import SSLContext
+    from typing import IO, Generator
+
+    from uvicorn._types import (
+        ASGIReceiveCallable,
+        ASGISendCallable,
+        HTTPResponseBodyEvent,
+        HTTPResponseStartEvent,
+        LifespanStartupFailedEvent,
+        Scope,
+    )
+
+pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="requires unix")
+gunicorn_arbiter = pytest.importorskip("gunicorn.arbiter", reason="requires gunicorn")
+uvicorn_workers = pytest.importorskip("uvicorn.workers", reason="requires gunicorn")
 
 
 class Process(subprocess.Popen):
@@ -26,17 +37,40 @@ def read_output(self) -> str:
         return self.output.read().decode()
 
 
-@pytest.fixture(params=(uvicorn_worker._workers.UvicornWorker, uvicorn_worker._workers.UvicornH11Worker))
+@pytest.fixture(
+    params=(
+        pytest.param(uvicorn_workers.UvicornWorker),
+        pytest.param(uvicorn_workers.UvicornH11Worker),
+    )
+)
 def worker_class(request: pytest.FixtureRequest) -> str:
-    """Gunicorn worker class names to test."""
+    """Gunicorn worker class names to test.
+
+    This is a parametrized fixture. When the fixture is used in a test, the test
+    will be automatically parametrized, running once for each fixture parameter. All
+    tests using the fixture will be automatically marked with `pytest.mark.subprocess`.
+
+    https://docs.pytest.org/en/latest/how-to/fixtures.html
+    https://docs.pytest.org/en/latest/proposals/parametrize_with_fixtures.html
+    """
     worker_class = request.param
     return f"{worker_class.__module__}.{worker_class.__name__}"
 
 
 async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
     assert scope["type"] == "http"
-    await send({"type": "http.response.start", "status": 204, "headers": []})
-    await send({"type": "http.response.body", "body": b"", "more_body": False})
+    start_event: HTTPResponseStartEvent = {
+        "type": "http.response.start",
+        "status": 204,
+        "headers": [],
+    }
+    body_event: HTTPResponseBodyEvent = {
+        "type": "http.response.body",
+        "body": b"",
+        "more_body": False,
+    }
+    await send(start_event)
+    await send(body_event)
 
 
 @pytest.fixture(
@@ -110,7 +144,7 @@ def test_get_request_to_asgi_app(gunicorn_process: Process) -> None:
     assert "uvicorn.workers", "startup complete" in output_text
 
 
-@pytest.mark.parametrize("signal_to_send", Arbiter.SIGNALS)
+@pytest.mark.parametrize("signal_to_send", gunicorn_arbiter.Arbiter.SIGNALS)
 def test_gunicorn_arbiter_signal_handling(gunicorn_process: Process, signal_to_send: signal.Signals) -> None:
     """Test Gunicorn arbiter signal handling.
 
@@ -120,7 +154,7 @@ def test_gunicorn_arbiter_signal_handling(gunicorn_process: Process, signal_to_s
 
     https://docs.gunicorn.org/en/latest/signals.html
     """
-    signal_abbreviation = Arbiter.SIG_NAMES[signal_to_send]
+    signal_abbreviation = gunicorn_arbiter.Arbiter.SIG_NAMES[signal_to_send]
     expected_text = f"Handling signal: {signal_abbreviation}"
     gunicorn_process.send_signal(signal_to_send)
     time.sleep(0.5)
@@ -187,7 +221,9 @@ def gunicorn_process_with_lifespan_startup_failure(
             process.wait(timeout=2)
 
 
-def test_uvicorn_worker_boot_error(gunicorn_process_with_lifespan_startup_failure: Process) -> None:
+def test_uvicorn_worker_boot_error(
+    gunicorn_process_with_lifespan_startup_failure: Process,
+) -> None:
     """Test Gunicorn arbiter shutdown behavior after Uvicorn worker boot errors.
 
     Previously, if Uvicorn workers raised exceptions during startup,