Add pytest-xdist

.gitignore

@@ -1,3 +1,3 @@
 __pycache__
 dist
-.coverage
+.coverage*

pyproject.toml

@@ -59,6 +59,7 @@ filterwarnings = ["error"]
 
 [tool.coverage.run]
 source_pkgs = ["uvicorn_worker", "tests"]
+parallel = true
 
 [tool.coverage.report]
 show_missing = true

requirements.txt

@@ -8,5 +8,9 @@ twine==4.0.2
 ruff==0.3.4
 mypy==1.9.0
 pytest==8.0.0
+pytest-xdist==1.3.0
 coverage==7.4.1
+coverage_enable_subprocess==1.0
 httpx==0.27.0
+trustme==1.1.0
+cryptography==42.0.4

scripts/coverage

@@ -7,4 +7,5 @@ fi
 
 set -x
 
+${PREFIX}coverage combine -a -q
 ${PREFIX}coverage report

scripts/test

@@ -11,7 +11,11 @@ if [ -z $GITHUB_ACTIONS ]; then
     scripts/check
 fi
 
-${PREFIX}coverage run --debug config -m pytest "$@"
+# enable subprocess coverage
+# https://coverage.readthedocs.io/en/latest/subprocess.html
+# https://github.com/nedbat/coveragepy/issues/367
+export COVERAGE_PROCESS_START=$PWD/pyproject.toml
+${PREFIX}coverage run --debug config -m pytest "$@" -n auto
 
 if [ -z $GITHUB_ACTIONS ]; then
     scripts/coverage

tests/conftest.py

@@ -1,32 +1,14 @@
 from __future__ import annotations
 
 import contextlib
-import importlib.util
-import os
 import socket
 import ssl
-from copy import deepcopy
-from hashlib import md5
-from pathlib import Path
-from tempfile import TemporaryDirectory
-from threading import Thread
-from time import sleep
-from typing import Any
-from uuid import uuid4
 
 import pytest
-
-try:
-    import trustme
-    from cryptography.hazmat.backends import default_backend
-    from cryptography.hazmat.primitives import serialization
-
-    HAVE_TRUSTME = True
-except ImportError:  # pragma: no cover
-    HAVE_TRUSTME = False
-
+import trustme
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
 from uvicorn.config import LOGGING_CONFIG
-from uvicorn.importer import import_from_string
 
 # Note: We explicitly turn the propagate on just for tests, because pytest
 # caplog not able to capture no-propagate loggers.
@@ -42,8 +24,6 @@
 
 @pytest.fixture
 def tls_certificate_authority() -> trustme.CA:
-    if not HAVE_TRUSTME:
-        pytest.skip("trustme not installed")  # pragma: no cover
     return trustme.CA()
 
 
@@ -69,7 +49,7 @@ def tls_ca_certificate_private_key_path(tls_certificate_authority: trustme.CA):
 
 
 @pytest.fixture
-def tls_certificate_private_key_encrypted_path(tls_certificate):
+def tls_certificate_private_key_encrypted_path(tls_certificate: trustme.LeafCert):
     private_key = serialization.load_pem_private_key(
         tls_certificate.private_key_pem.bytes(),
         password=None,
@@ -81,7 +61,7 @@ def tls_certificate_private_key_encrypted_path(tls_certificate):
         serialization.BestAvailableEncryption(b"uvicorn password for the win"),
     )
     with trustme.Blob(encrypted_key).tempfile() as private_encrypted_key:
-        yield private_encrypted_key
+        yield private_encrypted_key  # pragma: no cover
 
 
 @pytest.fixture
@@ -92,7 +72,7 @@ def tls_certificate_private_key_path(tls_certificate: trustme.CA):
 
 @pytest.fixture
 def tls_certificate_key_and_chain_path(tls_certificate: trustme.LeafCert):
-    with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem:
+    with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem:  # pragma: no cover
         yield cert_pem
 
 
@@ -109,132 +89,6 @@ def tls_ca_ssl_context(tls_certificate_authority: trustme.CA) -> ssl.SSLContext:
     return ssl_ctx
 
 
-@pytest.fixture(scope="package")
-def reload_directory_structure(tmp_path_factory: pytest.TempPathFactory):
-    """
-    This fixture creates a directory structure to enable reload parameter tests
-
-    The fixture has the following structure:
-    root
-    ├── [app, app_first, app_second, app_third]
-    │   ├── css
-    │   │   └── main.css
-    │   ├── js
-    │   │   └── main.js
-    │   ├── src
-    │   │   └── main.py
-    │   └── sub
-    │       └── sub.py
-    ├── ext
-    │   └── ext.jpg
-    ├── .dotted
-    ├── .dotted_dir
-    │   └── file.txt
-    └── main.py
-    """
-    root = tmp_path_factory.mktemp("reload_directory")
-    apps = ["app", "app_first", "app_second", "app_third"]
-
-    root_file = root / "main.py"
-    root_file.touch()
-
-    dotted_file = root / ".dotted"
-    dotted_file.touch()
-
-    dotted_dir = root / ".dotted_dir"
-    dotted_dir.mkdir()
-    dotted_dir_file = dotted_dir / "file.txt"
-    dotted_dir_file.touch()
-
-    for app in apps:
-        app_path = root / app
-        app_path.mkdir()
-        dir_files = [
-            ("src", ["main.py"]),
-            ("js", ["main.js"]),
-            ("css", ["main.css"]),
-            ("sub", ["sub.py"]),
-        ]
-        for directory, files in dir_files:
-            directory_path = app_path / directory
-            directory_path.mkdir()
-            for file in files:
-                file_path = directory_path / file
-                file_path.touch()
-    ext_dir = root / "ext"
-    ext_dir.mkdir()
-    ext_file = ext_dir / "ext.jpg"
-    ext_file.touch()
-
-    yield root
-
-
-@pytest.fixture
-def anyio_backend() -> str:
-    return "asyncio"
-
-
-@pytest.fixture(scope="function")
-def logging_config() -> dict[str, Any]:
-    return deepcopy(LOGGING_CONFIG)
-
-
-@pytest.fixture
-def short_socket_name(tmp_path, tmp_path_factory):  # pragma: py-win32
-    max_sock_len = 100
-    socket_filename = "my.sock"
-    identifier = f"{uuid4()}-"
-    identifier_len = len(identifier.encode())
-    tmp_dir = Path("/tmp").resolve()
-    os_tmp_dir = Path(os.getenv("TMPDIR", "/tmp")).resolve()
-    basetemp = Path(
-        str(tmp_path_factory.getbasetemp()),
-    ).resolve()
-    hash_basetemp = md5(
-        str(basetemp).encode(),
-    ).hexdigest()
-
-    def make_tmp_dir(base_dir):
-        return TemporaryDirectory(
-            dir=str(base_dir),
-            prefix="p-",
-            suffix=f"-{hash_basetemp}",
-        )
-
-    paths = basetemp, os_tmp_dir, tmp_dir
-    for _num, tmp_dir_path in enumerate(paths, 1):
-        with make_tmp_dir(tmp_dir_path) as tmpd:
-            tmpd = Path(tmpd).resolve()
-            sock_path = str(tmpd / socket_filename)
-            sock_path_len = len(sock_path.encode())
-            if sock_path_len <= max_sock_len:
-                if max_sock_len - sock_path_len >= identifier_len:  # pragma: no cover
-                    sock_path = str(tmpd / "".join((identifier, socket_filename)))
-                yield sock_path
-                return
-
-
-def sleep_touch(*paths: Path):
-    sleep(0.1)
-    for p in paths:
-        p.touch()
-
-
-@pytest.fixture
-def touch_soon():
-    threads = []
-
-    def start(*paths: Path):
-        thread = Thread(target=sleep_touch, args=paths)
-        thread.start()
-        threads.append(thread)
-
-    yield start
-
-    for t in threads:
-        t.join()
-
-
 def _unused_port(socket_type: int) -> int:
     """Find an unused localhost port from 1024-65535 and return it."""
     with contextlib.closing(socket.socket(type=socket_type)) as sock:
@@ -247,37 +101,3 @@ def _unused_port(socket_type: int) -> int:
 @pytest.fixture
 def unused_tcp_port() -> int:
     return _unused_port(socket.SOCK_STREAM)
-
-
-@pytest.fixture(
-    params=[
-        pytest.param(
-            "uvicorn.protocols.websockets.wsproto_impl:WSProtocol",
-            marks=pytest.mark.skipif(not importlib.util.find_spec("wsproto"), reason="wsproto not installed."),
-            id="wsproto",
-        ),
-        pytest.param(
-            "uvicorn.protocols.websockets.websockets_impl:WebSocketProtocol",
-            id="websockets",
-        ),
-    ]
-)
-def ws_protocol_cls(request: pytest.FixtureRequest):
-    return import_from_string(request.param)
-
-
-@pytest.fixture(
-    params=[
-        pytest.param(
-            "uvicorn.protocols.http.httptools_impl:HttpToolsProtocol",
-            marks=pytest.mark.skipif(
-                not importlib.util.find_spec("httptools"),
-                reason="httptools not installed.",
-            ),
-            id="httptools",
-        ),
-        pytest.param("uvicorn.protocols.http.h11_impl:H11Protocol", id="h11"),
-    ]
-)
-def http_protocol_cls(request: pytest.FixtureRequest):
-    return import_from_string(request.param)

tests/test_workers.py

@@ -2,30 +2,17 @@
 
 import signal
 import subprocess
-import sys
 import tempfile
 import time
-from typing import TYPE_CHECKING
+from ssl import SSLContext
+from typing import IO, Generator
 
 import httpx
 import pytest
+from gunicorn.arbiter import Arbiter
+from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, LifespanStartupFailedEvent, Scope
 
-if TYPE_CHECKING:
-    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")
+from uvicorn_worker import UvicornH11Worker, UvicornWorker
 
 
 class Process(subprocess.Popen):
@@ -37,7 +24,7 @@ def read_output(self) -> str:
         return self.output.read().decode()
 
 
-@pytest.fixture(params=(uvicorn_workers.UvicornWorker, uvicorn_workers.UvicornH11Worker))
+@pytest.fixture(params=(UvicornWorker, UvicornH11Worker))
 def worker_class(request: pytest.FixtureRequest) -> str:
     """Gunicorn worker class names to test."""
     worker_class = request.param
@@ -46,18 +33,8 @@ def worker_class(request: pytest.FixtureRequest) -> str:
 
 async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
     assert scope["type"] == "http"
-    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)
+    await send({"type": "http.response.start", "status": 204, "headers": []})
+    await send({"type": "http.response.body", "body": b"", "more_body": False})
 
 
 @pytest.fixture(
@@ -131,7 +108,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", gunicorn_arbiter.Arbiter.SIGNALS)
+@pytest.mark.parametrize("signal_to_send", Arbiter.SIGNALS)
 def test_gunicorn_arbiter_signal_handling(gunicorn_process: Process, signal_to_send: signal.Signals) -> None:
     """Test Gunicorn arbiter signal handling.
 
@@ -141,7 +118,7 @@ def test_gunicorn_arbiter_signal_handling(gunicorn_process: Process, signal_to_s
 
     https://docs.gunicorn.org/en/latest/signals.html
     """
-    signal_abbreviation = gunicorn_arbiter.Arbiter.SIG_NAMES[signal_to_send]
+    signal_abbreviation = Arbiter.SIG_NAMES[signal_to_send]
     expected_text = f"Handling signal: {signal_abbreviation}"
     gunicorn_process.send_signal(signal_to_send)
     time.sleep(0.5)
@@ -208,9 +185,7 @@ 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,