Skip to content

Factor out common conftest logic between django-cf and runtime-sdk #244

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hoodmane merged 1 commit into from
Sep 2, 2026
Merged

Factor out common conftest logic between django-cf and runtime-sdk #244

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions packages/django-cf/pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,12 @@ dev = [
"requests",
"ruff",
"django",
"testlib[host]",
]

[tool.uv.sources]
testlib = { path = "../testlib", editable = true }

[project.urls]
"Homepage" = "https://github.com/cloudflare/workers-py"
"Bug Reports" = "https://github.com/cloudflare/workers-py/issues"
Expand Down
269 changes: 32 additions & 237 deletions packages/django-cf/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,26 +1,27 @@
"""Host-side fixtures that serve the test workers with ``pywrangler dev``.

TODO: reduce the duplication between this file and packages/runtime-sdk/tests/conftest.py
"""
"""Host-side fixtures that serve the test workers with ``pywrangler dev``."""

# pyright: reportMissingImports=false, reportMissingModuleSource=false

import ast
import contextlib
import functools
import os
import shutil
import signal
import socket
import subprocess
import time
from collections.abc import Callable, Generator
from dataclasses import dataclass, field
from collections.abc import Generator
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal, TypedDict

import pytest
import requests
from testlib.host import (
COMPAT_CONFIGS,
CompatConfig,
configure_compatibility,
)
from testlib.host import (
dev_server as run_dev_server,
)
from testlib.host import (
register_in_worker_suites as register_testlib_suites,
)

TEST_DIR: Path = Path(__file__).parent
PACKAGE_DIR: Path = TEST_DIR.parent
Expand All @@ -35,12 +36,7 @@
IN_WORKER_PROJECT: Path = TEST_DIR / "in_worker" / "worker"

DEV_STARTUP_TIMEOUT: int = 240
DEV_POLL_INTERVAL: float = 0.5
SEED_TIMEOUT: int = 180
TEARDOWN_TIMEOUT: int = 10
SUITE_CONNECT_TIMEOUT: int = 10
SUITE_READ_TIMEOUT: int = 300

GENERATED = shutil.ignore_patterns(
".venv",
".venv-workers",
Expand All @@ -52,128 +48,17 @@
)


@dataclass(frozen=True)
class CompatConfig:
compat_date: str
python_version: str
extra_compat_flags: list[str] = field(default_factory=list)


COMPAT_CONFIGS: list[CompatConfig] = [
CompatConfig(
compat_date="2025-09-01",
python_version="3.12",
extra_compat_flags=[
"enable_python_external_sdk",
"python_process_pth_files",
"python_request_headers_preserve_commas",
],
),
CompatConfig(
compat_date="2026-01-01",
python_version="3.13",
extra_compat_flags=[
"enable_python_external_sdk",
"python_process_pth_files",
"python_request_headers_preserve_commas",
],
),
CompatConfig(
compat_date="2026-07-01",
python_version="3.14",
# TODO: remove these when 3.14 is stable, and enabled by date
extra_compat_flags=["python_workers_314", "experimental"],
),
]


@dataclass(frozen=True)
class DevServer:
base_url: str


def get_free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]


def _terminate(process: subprocess.Popen[bytes]) -> None:
if process.poll() is not None:
return
# `pywrangler dev` spawns npx -> wrangler -> workerd; signalling the group
# is the only way to avoid orphaning workerd.
group = os.getpgid(process.pid)
os.killpg(group, signal.SIGTERM)
try:
process.wait(timeout=TEARDOWN_TIMEOUT)
except subprocess.TimeoutExpired:
os.killpg(group, signal.SIGKILL)
process.wait()


def _fail(log_path: Path, message: str) -> None:
# Callers run inside `_dev_server`, whose finally block stops the worker.
pytest.fail(
f"{message}\n\n--- pywrangler dev log ---\n{log_path.read_text(errors='replace')}"
)


def _wait_for_ready(
process: subprocess.Popen[bytes], base_url: str, log_path: Path
) -> None:
"""Block until the worker answers.

Any status counts: a 404 still proves workerd loaded the script and is
routing requests.
"""
deadline = time.monotonic() + DEV_STARTUP_TIMEOUT
while time.monotonic() < deadline:
if process.poll() is not None:
_fail(
log_path, f"pywrangler dev exited early with code {process.returncode}"
)
try:
requests.get(base_url, timeout=5)
return
except requests.RequestException:
time.sleep(DEV_POLL_INTERVAL)

_fail(log_path, f"pywrangler dev was not ready within {DEV_STARTUP_TIMEOUT}s")


@contextlib.contextmanager
def _dev_server(
target: Path, tmp_path: Path, env: dict[str, str], pywrangler: list[str]
) -> Generator[tuple[str, Path]]:
"""Run `pywrangler dev` on a free port, yielding its base URL and log path."""
port = get_free_port()
base_url = f"http://127.0.0.1:{port}"
log_path = tmp_path / f"{target.name}-dev.log"

with log_path.open("w") as log_file:
process = subprocess.Popen(
[
*pywrangler,
"dev",
"--port",
str(port),
"--persist-to",
str(tmp_path / "state"),
],
cwd=target,
stdout=log_file,
stderr=subprocess.STDOUT,
env=env,
start_new_session=True,
)
try:
_wait_for_ready(process, base_url, log_path)
yield base_url, log_path
finally:
_terminate(process)


def _seed(base_url: str, log_path: Path) -> None:
for endpoint in ("__run_migrations__", "__create_admin__"):
try:
Expand Down Expand Up @@ -219,7 +104,13 @@ def _serve(project_dir: Path, tmp_path: Path) -> Generator[DevServer]:
DJANGO_CF_SRC, vendored, ignore=shutil.ignore_patterns("__pycache__")
)

with _dev_server(target, tmp_path, env, pywrangler) as (base_url, log_path):
with run_dev_server(
target,
tmp_path,
env,
pywrangler,
startup_timeout=DEV_STARTUP_TIMEOUT,
) as (base_url, log_path):
_seed(base_url, log_path)
yield DevServer(base_url)

Expand All @@ -243,19 +134,6 @@ def r2_web_server(tmp_path_factory: pytest.TempPathFactory) -> Generator[DevServ
yield from _serve(R2_PROJECT, tmp_path_factory.mktemp("r2"))


def replace_compat_date(file: Path, compat_date: str) -> None:
file.write_text(file.read_text().replace("%COMPAT_DATE", compat_date))


def inject_compat_flags(file: Path, extra_flags: list[str]) -> None:
if not extra_flags:
return
content = file.read_text()
for flag in extra_flags:
content = content.replace('"python_workers"', f'"python_workers", "{flag}"')
file.write_text(content)


@pytest.fixture(scope="session", autouse=True)
def build_testlib():
subprocess.run(["uv", "build"], cwd=TESTLIB, check=True)
Expand All @@ -271,7 +149,7 @@ def compat_config(request: pytest.FixtureRequest) -> CompatConfig:


@pytest.fixture(scope="module")
def in_worker_server(
def dev_server(
tmp_path_factory: pytest.TempPathFactory, compat_config: CompatConfig
) -> Generator[str]:
"""Serve ``tests/in_worker/worker``, once per compat config.
Expand All @@ -286,8 +164,7 @@ def in_worker_server(
shutil.copytree(TESTLIB, tmp_path / "testlib", ignore=GENERATED)

wrangler_jsonc = target / "wrangler.jsonc"
replace_compat_date(wrangler_jsonc, compat_config.compat_date)
inject_compat_flags(wrangler_jsonc, compat_config.extra_compat_flags)
configure_compatibility(wrangler_jsonc, compat_config)

pywrangler = [
"uv",
Expand All @@ -310,97 +187,15 @@ def in_worker_server(
ignore=shutil.ignore_patterns("__pycache__"),
)

with _dev_server(target, tmp_path, env, pywrangler) as (base_url, _):
with run_dev_server(
target,
tmp_path,
env,
pywrangler,
startup_timeout=DEV_STARTUP_TIMEOUT,
) as (base_url, _):
yield base_url


class InWorkerTestResult(TypedDict):
status: Literal["passed", "failed", "error", "skipped"]
error: str
traceback: str
reason: str


SuiteResults = dict[str, InWorkerTestResult]


@functools.cache
def get_suite_results(in_worker_server: str, suite: str) -> SuiteResults | str:
try:
resp = requests.get(
f"{in_worker_server}/run-tests/{suite}",
timeout=(SUITE_CONNECT_TIMEOUT, SUITE_READ_TIMEOUT),
)
except requests.RequestException as error:
return f"Suite '{suite}' request failed: {error}"
if not resp.ok:
return f"Suite '{suite}' returned {resp.status_code}: {resp.text}"
return resp.json()


def _make_test(suite: str, test_name: str) -> Callable:
def test_fn(self: Any, in_worker_server: str) -> None:
results = get_suite_results(in_worker_server, suite)
if isinstance(results, str):
pytest.fail(results)
return
result: InWorkerTestResult | None = results.get(test_name)
assert result is not None, (
f"Test {suite}::{test_name} not found in results; "
f"available keys: {sorted(results)}"
)
if result["status"] == "skipped":
pytest.skip(result.get("reason", ""))
elif result["status"] == "failed":
pytest.fail(result["error"])
elif result["status"] == "error":
pytest.fail(f"{result['error']}\n{result.get('traceback', '')}")

test_fn.__name__ = f"test_{test_name}"
return test_fn


def make_suite_class(suite: str, tests: list[str]) -> type:
camel = "".join(part.title() for part in suite.split("_"))
return type(
f"Test{camel}",
(),
{f"test_{name}": _make_test(suite, name) for name in tests},
)


def _normalize_test_name(*parts: str) -> str:
normalized = []
for part in parts:
normalized.append(part[len("test_") :] if part.startswith("test_") else part)
return "__".join(normalized)


def discover_test_names(module_path: Path) -> list[str]:
tree = ast.parse(module_path.read_text())
names = []
for node in tree.body:
if isinstance(
node, ast.FunctionDef | ast.AsyncFunctionDef
) and node.name.startswith("test_"):
names.append(_normalize_test_name(node.name))
elif isinstance(node, ast.ClassDef):
for child in node.body:
if isinstance(
child, ast.FunctionDef | ast.AsyncFunctionDef
) and child.name.startswith("test_"):
names.append(_normalize_test_name(node.name, child.name))
return names


def discover_suites(src_dir: Path) -> dict[str, list[str]]:
return {
module_path.stem[len("test_") :]: discover_test_names(module_path)
for module_path in sorted(src_dir.glob("test_*.py"))
}


def register_in_worker_suites(namespace: dict[str, Any], src_dir: Path) -> None:
for suite, test_names in discover_suites(src_dir).items():
suite_cls = make_suite_class(suite, test_names)
namespace[suite_cls.__name__] = suite_cls
def register_in_worker_suites(namespace: dict, src_dir: Path) -> None:
register_testlib_suites(namespace, src_dir)
Loading
Loading