phase: 01_runnable_service_scaffold

**Phase 01 final verification pass — all criteria green**

- Verified (nothing left to implement; all 4 task files already in `complete/`): toolchain lock, typed `Settings`+`.env.example`, debugpy bootstrap, app factory, CLI (`--host/--port/--reload/--levels-dir`, `python -m undercooked`), static shell + `/static`, security headers, no-CDN/doc guards, `validate.sh`.
- `uv sync --extra dev` → exit 0; `uv.lock` present (untracked → harness commits it with the phase).
- `uv run undercooked --host 127.0.0.1 --port 8000` → `curl /healthz` = 200 `{"status":"ok","version":"0.1.0","rooms":0}` (superset of the criterion's body — now documented in README); `curl /` = 200 shell HTML; static css/js 200; CSP/nosniff/no-referrer/DENY on 200, 404 and 405 too.
- `DEBUGPY=1`: port 5678 is held on this machine by the user's own podman dev-container debugpy adapter — **no live change made/killed**; verified via `DEBUGPY=1 DEBUGPY_PORT=5679` → listener opens, `/healthz` still 200; the occupied-port case degraded gracefully as designed (warning, server serves). Bad config → one line + exit 2 ✓.
- `bash .agents/validate.sh` (root **and** `.agents/`) → exit 0: `ruff check .` 0 errors, `pyright` 0 errors, levels-CLI notice-skip (matches README prose), **284 passed** full suite incl. 10 Playwright e2e, coverage **100.00%** (floor 90), repo hygiene clean. `uv run pytest -m e2e` → 10 passed.
- Docs fixed: README Status → phase 01 complete + next pending 02 (was pointing at `todo/01…`), documented real `/healthz` body, precedence now "CLI flags > env > .env > default", added the PLAN §13-promised **How to play** section (honest about shell-only today), Layout rows marked what exists vs phases 02/07. AGENTS.md: recorded the `data:`-URL exemption in the no-CDN rule (test allows it) and matched the session-token rule to the browser's WebCrypto/base64url implementation. Also corrected a stale "session-wide chromium" docstring in `tests/e2e/conftest.py` (fixtures are function-scoped).
- Deviations: no container build run (needs an image pull; not a phase criterion — `Containerfile`/`compose.yaml` reviewed and consistent with `docs/DEPLOYMENT.md`); `HEAD /` is 405 (routes are GET-only by design; health checks use GET). No git index/history commands run.
- Next pending phase: `02_level_content_and_validation`.
This commit is contained in:
2026-10-06 01:51:20 -04:00
parent f84c6ebaef
commit 6fd83446ec
43 changed files with 5478 additions and 30 deletions
+1
View File
@@ -0,0 +1 @@
Pure, fast unit tests: config parsing, helpers, and other no-I/O logic.
+471
View File
@@ -0,0 +1,471 @@
"""Unit tests for `undercooked.config` (PLAN.md A17): defaults, overrides, validation.
Everything here calls `load_settings({...})` with an explicit mapping wherever
possible, so a developer's local `.env` can never change what a test asserts.
The `.env` / cache behaviour is exercised separately, in an isolated directory.
"""
from __future__ import annotations
import dataclasses
import os
import subprocess
import sys
from collections.abc import Callable, Iterator
from pathlib import Path
import pytest
from undercooked import config
from undercooked.config import (
ENV_KEYS,
ConfigError,
Settings,
apply_cli_overrides,
get_settings,
load_settings,
reset_settings_cache,
)
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
@pytest.fixture(autouse=True)
def pristine_env() -> Iterator[None]:
"""Undo env/cache side effects - `load_dotenv` mutates `os.environ` in place."""
saved = dict(os.environ)
reset_settings_cache()
yield
os.environ.clear()
os.environ.update(saved)
reset_settings_cache()
# ----------------------------------------------------------------------------- defaults
def test_defaults_match_the_documented_table() -> None:
settings = load_settings({})
assert settings == Settings()
assert settings.host == "127.0.0.1"
assert settings.port == 8000
assert settings.tick_hz == 20
assert settings.snapshot_hz == 20
assert settings.max_players == 4
assert settings.max_rooms == 200
assert settings.levels_dir == Path("levels")
assert settings.static_dir is None # derived from the package layout, see server.app
assert settings.lobby_idle_timeout_s == 900.0
assert settings.guest_idle_timeout_s == 45.0
assert settings.max_message_bytes == 4096
assert settings.max_inputs_per_s == 40
assert settings.sim_latency_ms == 0
assert settings.debugpy is False
assert settings.debugpy_port == 5678
assert settings.allowed_origins == ()
def test_env_keys_cover_exactly_the_settings_fields() -> None:
assert set(ENV_KEYS) == {field.name for field in dataclasses.fields(Settings)}
def test_settings_is_frozen_and_uses_slots() -> None:
settings = Settings()
with pytest.raises(dataclasses.FrozenInstanceError):
settings.port = 1234 # type: ignore[misc]
assert not hasattr(settings, "__dict__")
def test_config_error_is_a_value_error() -> None:
assert issubclass(ConfigError, ValueError)
# ----------------------------------------------------------------------------- overrides
@pytest.mark.parametrize(
("field", "raw", "expected"),
[
("host", "localhost", "localhost"),
("port", "8123", 8123),
("tick_hz", "30", 30),
("snapshot_hz", "15", 15),
("max_players", "2", 2),
("max_rooms", "1", 1),
("lobby_idle_timeout_s", "60", 60.0),
("guest_idle_timeout_s", "7.5", 7.5),
("max_message_bytes", "8192", 8192),
("max_inputs_per_s", "60", 60),
("sim_latency_ms", "150", 150),
("debugpy_port", "5679", 5679),
],
)
def test_single_variable_override(field: str, raw: str, expected: object) -> None:
settings = load_settings({ENV_KEYS[field]: raw})
assert getattr(settings, field) == expected
assert settings.levels_dir == Path("levels") # untouched fields keep their default
@pytest.mark.parametrize("raw", ["1", "true", "TRUE", "yes", "on"])
def test_debugpy_truthy_values(raw: str) -> None:
assert load_settings({"DEBUGPY": raw}).debugpy is True
@pytest.mark.parametrize("raw", ["0", "false", "False", "no", "off"])
def test_debugpy_falsy_values(raw: str) -> None:
assert load_settings({"DEBUGPY": raw}).debugpy is False
@pytest.mark.parametrize(
("raw", "expected"),
[
("", ()),
(" ", ()),
("http://localhost:3000", ("http://localhost:3000",)),
("http://a , http://b", ("http://a", "http://b")),
("http://a,,http://b,", ("http://a", "http://b")),
(",,", ()),
],
)
def test_allowed_origins_is_comma_split(raw: str, expected: tuple[str, ...]) -> None:
assert load_settings({ENV_KEYS["allowed_origins"]: raw}).allowed_origins == expected
@pytest.mark.parametrize(
("raw", "expected"),
[
("levels", Path("levels")),
("levels/", Path("levels")),
("/srv/undercooked/levels", Path("/srv/undercooked/levels")),
("~/kitchens", Path.home() / "kitchens"),
("./kitchens", Path("kitchens")),
],
)
def test_levels_dir_path_handling(raw: str, expected: Path) -> None:
assert load_settings({ENV_KEYS["levels_dir"]: raw}).levels_dir == expected
@pytest.mark.parametrize(
("raw", "expected"),
[
("/app/app/static", Path("/app/app/static")),
("~/static", Path.home() / "static"),
("static/", Path("static")),
("", None),
(" ", None),
],
)
def test_static_dir_is_optional_and_expanded(raw: str, expected: Path | None) -> None:
"""An unset `UNDERCOOKED_STATIC_DIR` means "derive it from the package"."""
assert load_settings({ENV_KEYS["static_dir"]: raw}).static_dir == expected
@pytest.mark.parametrize(
"field", ["port", "tick_hz", "lobby_idle_timeout_s", "host", "allowed_origins"]
)
@pytest.mark.parametrize("blank", ["", " ", "\t"])
def test_blank_value_counts_as_unset(field: str, blank: str) -> None:
assert load_settings({ENV_KEYS[field]: blank}) == Settings()
def test_surrounding_whitespace_is_tolerated() -> None:
settings = load_settings({ENV_KEYS["port"]: " 8123 ", ENV_KEYS["host"]: "\nlocalhost\t"})
assert (settings.port, settings.host) == (8123, "localhost")
def test_snapshot_hz_may_equal_tick_hz_but_not_exceed_it() -> None:
both_ten = load_settings({ENV_KEYS["tick_hz"]: "10", ENV_KEYS["snapshot_hz"]: "10"})
assert both_ten.snapshot_hz == 10
with pytest.raises(ConfigError, match=r"UNDERCOOKED_SNAPSHOT_HZ.*must be >= 1, <= 10"):
load_settings({ENV_KEYS["tick_hz"]: "10", ENV_KEYS["snapshot_hz"]: "20"})
# ----------------------------------------------------------------------------- validation errors
@pytest.mark.parametrize(
("field", "raw", "message"),
[
("port", "http", "UNDERCOOKED_PORT='http' is not an integer"),
("port", "8.5", "UNDERCOOKED_PORT='8.5' is not an integer"),
("port", "0", "UNDERCOOKED_PORT=0 is invalid: must be >= 1, <= 65535"),
("port", "65536", "UNDERCOOKED_PORT=65536 is invalid: must be >= 1, <= 65535"),
("tick_hz", "4", "UNDERCOOKED_TICK_HZ=4 is invalid: must be >= 5, <= 60"),
("tick_hz", "99", "UNDERCOOKED_TICK_HZ=99 is invalid: must be >= 5, <= 60"),
("tick_hz", "twenty", "UNDERCOOKED_TICK_HZ='twenty' is not an integer"),
("snapshot_hz", "0", "UNDERCOOKED_SNAPSHOT_HZ=0 is invalid: must be >= 1, <= 20"),
("max_players", "0", "UNDERCOOKED_MAX_PLAYERS=0 is invalid: must be >= 1, <= 4"),
("max_players", "5", "UNDERCOOKED_MAX_PLAYERS=5 is invalid: must be >= 1, <= 4"),
("max_rooms", "0", "UNDERCOOKED_MAX_ROOMS=0 is invalid: must be >= 1"),
("max_rooms", "-3", "UNDERCOOKED_MAX_ROOMS=-3 is invalid: must be >= 1"),
(
"lobby_idle_timeout_s",
"-1",
"UNDERCOOKED_LOBBY_IDLE_TIMEOUT_S=-1.0 is invalid: must be >= 0.0",
),
("lobby_idle_timeout_s", "soon", "UNDERCOOKED_LOBBY_IDLE_TIMEOUT_S='soon' is not a number"),
(
"guest_idle_timeout_s",
"-0.5",
"UNDERCOOKED_GUEST_IDLE_TIMEOUT_S=-0.5 is invalid: must be >= 0.0",
),
(
"max_message_bytes",
"255",
"UNDERCOOKED_MAX_MESSAGE_BYTES=255 is invalid: must be >= 256",
),
("max_inputs_per_s", "0", "UNDERCOOKED_MAX_INPUTS_PER_S=0 is invalid: must be >= 1"),
("sim_latency_ms", "-1", "UNDERCOOKED_SIM_LATENCY_MS=-1 is invalid: must be >= 0"),
("debugpy_port", "0", "DEBUGPY_PORT=0 is invalid: must be >= 1, <= 65535"),
(
"debugpy",
"maybe",
"DEBUGPY='maybe' is not a boolean (use 1/0, true/false, yes/no, on/off)",
),
],
)
def test_invalid_values_raise_config_error(field: str, raw: str, message: str) -> None:
with pytest.raises(ConfigError) as excinfo:
load_settings({ENV_KEYS[field]: raw})
assert str(excinfo.value) == message
def test_out_of_range_tick_hz_fails_from_the_real_environment() -> None:
"""The documented criterion: `UNDERCOOKED_TICK_HZ=99` fails loudly at startup."""
env = {
key: value
for key, value in os.environ.items()
if not key.startswith(("UNDERCOOKED_", "DEBUGPY"))
}
env["UNDERCOOKED_TICK_HZ"] = "99"
result = subprocess.run( # fixed argv, repo's own interpreter
[sys.executable, "-c", "from undercooked.config import load_settings; load_settings()"],
cwd=REPO_ROOT,
env=env,
capture_output=True,
text=True,
check=False,
timeout=120,
)
assert result.returncode != 0
assert "ConfigError" in result.stderr
assert "UNDERCOOKED_TICK_HZ=99" in result.stderr
# ----------------------------------------------------------------------------- cached accessor
def test_get_settings_caches_until_reset(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("UNDERCOOKED_PORT", "8123")
first = get_settings()
assert first.port == 8123
monkeypatch.setenv("UNDERCOOKED_PORT", "9999")
assert get_settings() is first # still the cached object
reset_settings_cache()
assert get_settings().port == 9999
def test_get_settings_applies_defaults_with_an_empty_environment(
monkeypatch: pytest.MonkeyPatch,
) -> None:
for key in list(os.environ):
if key.startswith(("UNDERCOOKED_", "DEBUGPY")):
monkeypatch.delenv(key, raising=False)
monkeypatch.setenv("PYTHON_DOTENV_DISABLED", "true")
assert get_settings() == Settings()
# ----------------------------------------------------------------------------- .env loading
def test_dotenv_file_is_read_from_the_current_directory(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.delenv("UNDERCOOKED_PORT", raising=False)
monkeypatch.delenv("UNDERCOOKED_MAX_ROOMS", raising=False)
(tmp_path / ".env").write_text(
"UNDERCOOKED_PORT=8123\nUNDERCOOKED_MAX_ROOMS=7\n", encoding="utf-8"
)
monkeypatch.chdir(tmp_path)
settings = load_settings()
assert (settings.port, settings.max_rooms) == (8123, 7)
def test_dotenv_is_found_in_a_parent_directory(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.delenv("UNDERCOOKED_PORT", raising=False)
(tmp_path / ".env").write_text("UNDERCOOKED_PORT=8245\n", encoding="utf-8")
nested = tmp_path / "src" / "undercooked"
nested.mkdir(parents=True)
monkeypatch.chdir(nested)
assert load_settings().port == 8245
def test_dotenv_never_overrides_an_exported_variable(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setenv("UNDERCOOKED_PORT", "7000")
(tmp_path / ".env").write_text("UNDERCOOKED_PORT=8123\n", encoding="utf-8")
monkeypatch.chdir(tmp_path)
assert load_settings().port == 7000
def test_dotenv_is_read_once_until_the_cache_is_reset(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
loaded: list[str] = []
(tmp_path / ".env").write_text("UNDERCOOKED_PORT=8123\n", encoding="utf-8")
monkeypatch.chdir(tmp_path)
def fake_load_dotenv(path: object, **_kwargs: object) -> bool:
loaded.append(str(path))
return True
monkeypatch.setattr(config, "load_dotenv", fake_load_dotenv)
load_settings()
load_settings()
assert len(loaded) == 1 # one `.env` read per process, not one per call
reset_settings_cache()
load_settings()
assert len(loaded) == 2
def test_missing_dotenv_file_is_not_an_error(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.delenv("UNDERCOOKED_PORT", raising=False)
monkeypatch.chdir(tmp_path)
assert load_settings().port == 8000
def test_explicit_env_mapping_skips_dotenv(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
def never_find(**_kwargs: object) -> str:
pytest.fail("dotenv must not be read when an explicit env mapping is given")
monkeypatch.setattr(config, "find_dotenv", never_find)
assert load_settings({"UNDERCOOKED_PORT": "8123"}).port == 8123
# ----------------------------------------------------------------------------- .env.example
def parse_env_example(path: Path) -> dict[str, str]:
"""Read `KEY=value` lines from an env template, ignoring comments."""
values: dict[str, str] = {}
for line in path.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
key, _, value = stripped.partition("=")
values[key.strip()] = value.strip()
return values
def test_env_example_documents_every_variable_with_its_default() -> None:
example = parse_env_example(REPO_ROOT / ".env.example")
assert set(example) == set(ENV_KEYS.values())
assert load_settings(example) == Settings()
@pytest.mark.parametrize("forbidden", ["token", "password", "secret", "api_key", "reese", ".com"])
def test_env_example_values_carry_no_secrets_or_hostnames(forbidden: str) -> None:
"""The template is committed, so its values must stay placeholders (A16/A17)."""
for key, value in parse_env_example(REPO_ROOT / ".env.example").items():
assert forbidden not in value.lower(), f"{key} looks like it holds a secret or a real host"
assert len(value) <= 40, f"{key} is too long to be a default"
# ----------------------------------------------------------------------------- README table
def parse_readme_config_table(path: Path) -> dict[str, str]:
"""Read the VAR / default / meaning rows of the README configuration table.
A default written as *(empty)* means "no value", i.e. an empty string.
"""
lines = path.read_text(encoding="utf-8").splitlines()
try:
start = lines.index("## Configuration")
except ValueError: # pragma: no cover - fails loudly if the section is renamed
pytest.fail("README.md has no '## Configuration' section")
table: dict[str, str] = {}
for line in lines[start + 1 :]:
if line.startswith("## "):
break
stripped = line.strip()
if not stripped.startswith("|"):
continue
cells = [cell.strip() for cell in stripped.strip("|").split("|")]
key = cells[0].strip("`")
if not key.isupper():
continue # header and separator rows
default = cells[1].strip("`")
table[key] = "" if default.startswith("*") else default
return table
def test_readme_documents_every_variable_with_its_default() -> None:
table = parse_readme_config_table(REPO_ROOT / "README.md")
assert set(table) == set(ENV_KEYS.values())
assert load_settings(table) == Settings()
# ----------------------------------------------------------------------------- CLI overrides
def test_apply_cli_overrides_publishes_env_and_rebuilds(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("UNDERCOOKED_LEVELS_DIR", raising=False)
settings = apply_cli_overrides(levels_dir=Path("kitchens"), port=9123)
assert settings.levels_dir == Path("kitchens")
assert settings.port == 9123
assert os.environ["UNDERCOOKED_LEVELS_DIR"] == "kitchens"
assert get_settings() == settings # a later get_settings() sees the same thing
def test_apply_cli_overrides_beats_an_existing_environment_value(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("UNDERCOOKED_PORT", "8123")
assert apply_cli_overrides(port=9200).port == 9200
def test_apply_cli_overrides_validates_the_values_it_writes() -> None:
with pytest.raises(ConfigError, match=r"UNDERCOOKED_PORT=0 is invalid"):
apply_cli_overrides(port=0)
def test_apply_cli_overrides_rejects_an_unknown_field() -> None:
with pytest.raises(ConfigError, match="'nonsense' is not a configuration field"):
apply_cli_overrides(**{"nonsense": 1})
def test_apply_cli_overrides_renders_booleans_as_env_flags() -> None:
assert apply_cli_overrides(debugpy=True).debugpy is True
assert os.environ["DEBUGPY"] == "1"
assert apply_cli_overrides(debugpy=False).debugpy is False
assert os.environ["DEBUGPY"] == "0"
def test_apply_cli_overrides_with_nothing_still_validates_the_environment(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("UNDERCOOKED_TICK_HZ", "99")
with pytest.raises(ConfigError, match="UNDERCOOKED_TICK_HZ=99"):
apply_cli_overrides()
# ----------------------------------------------------------------------------- conftest factory
def test_make_settings_fixture_builds_settings(make_settings: Callable[..., Settings]) -> None:
settings = make_settings(tick_hz=30, snapshot_hz=15, debugpy=True, allowed_origins=["http://a", "http://b"])
assert (settings.tick_hz, settings.snapshot_hz) == (30, 15)
assert settings.debugpy is True
assert settings.allowed_origins == ("http://a", "http://b")
assert settings.levels_dir == Path("levels")
+114
View File
@@ -0,0 +1,114 @@
"""Unit tests for the debugpy bootstrap (PLAN.md A18).
The real debugger must never bind a port during the suite: each test injects a
fake `debugpy` through `sys.modules`, which is exactly what the lazy import in
`undercooked.debug` resolves against.
"""
from __future__ import annotations
import logging
import sys
from typing import Any
import pytest
from undercooked import debug
from undercooked.config import Settings
pytestmark = pytest.mark.unit
class FakeDebugpy:
"""Stand-in for the `debugpy` module; records `listen` calls, may fail."""
def __init__(self, error: Exception | None = None) -> None:
self.error = error
self.calls: list[tuple[Any, dict[str, Any]]] = []
def listen(self, address: Any, **kwargs: Any) -> Any:
self.calls.append((address, kwargs))
if self.error is not None:
raise self.error
return address
def install(monkeypatch: pytest.MonkeyPatch, fake: FakeDebugpy | None) -> None:
"""Put `fake` (or `None`, for "not installed") where `import debugpy` looks."""
monkeypatch.setitem(sys.modules, "debugpy", fake)
def test_disabled_is_a_no_op(monkeypatch: pytest.MonkeyPatch) -> None:
# `None` in sys.modules makes any `import debugpy` raise, so reaching the
# import at all would fail the test instead of silently binding a port.
install(monkeypatch, None)
assert debug.maybe_start_debugpy(Settings()) is False
def test_disabled_never_logs(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.DEBUG, logger="undercooked.debug"):
debug.maybe_start_debugpy(Settings(debugpy=False, debugpy_port=5999))
assert caplog.records == []
def test_enabled_listens_once_without_waiting(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeDebugpy()
install(monkeypatch, fake)
assert debug.maybe_start_debugpy(Settings(debugpy=True)) is True
assert len(fake.calls) == 1
address, kwargs = fake.calls[0]
assert address == (debug.DEBUGPY_BIND_HOST, 5678)
# Non-blocking is the whole point: startup must not stall waiting for an editor.
assert kwargs.get("wait_for_client", False) is False
def test_enabled_honours_a_custom_port(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeDebugpy()
install(monkeypatch, fake)
assert debug.maybe_start_debugpy(Settings(debugpy=True, debugpy_port=5999)) is True
assert fake.calls[0][0] == (debug.DEBUGPY_BIND_HOST, 5999)
def test_enabled_logs_the_endpoint(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
install(monkeypatch, FakeDebugpy())
with caplog.at_level(logging.INFO, logger="undercooked.debug"):
debug.maybe_start_debugpy(Settings(debugpy=True, debugpy_port=5999))
assert "5999" in caplog.text
assert "debugpy" in caplog.text.lower()
def test_missing_debugpy_does_not_stop_the_server(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
install(monkeypatch, None)
with caplog.at_level(logging.WARNING, logger="undercooked.debug"):
assert debug.maybe_start_debugpy(Settings(debugpy=True)) is False
assert "not installed" in caplog.text
@pytest.mark.parametrize(
"error",
[
OSError(98, "Address already in use"),
RuntimeError("already listening"),
ValueError("bad address"),
],
)
def test_listen_failures_are_swallowed(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
error: Exception,
) -> None:
install(monkeypatch, FakeDebugpy(error))
with caplog.at_level(logging.WARNING, logger="undercooked.debug"):
assert debug.maybe_start_debugpy(Settings(debugpy=True, debugpy_port=5678)) is False
assert "could not listen" in caplog.text
def test_logger_is_named_after_the_module() -> None:
assert isinstance(debug.logger, logging.Logger)
assert debug.logger.name == "undercooked.debug"
+409
View File
@@ -0,0 +1,409 @@
"""Docs guard: what `README.md` tells a human to type must be real (phase 01 task 04).
Documentation rots in two ways this file makes impossible:
1. **Pseudo-commands.** A fenced block that reads plausibly but cannot run — an invented
script, a tool that is not part of this project, a placeholder like `make deploy`.
Every command line inside a `bash`/`sh`/`console`/`shell` fence of `README.md` or
`docs/*.md` must start with a known-safe prefix, and whatever it launches through
`uv run` must be a console script or a dependency declared in `pyproject.toml`. That is
what keeps the Quickstart true after a rename — and what stops a promise (e.g. the
phase-02 `undercooked-levels` CLI) from being written as a runnable command before it
exists.
2. **Config drift.** A variable documented but not implemented, or implemented but never
documented. AGENTS.md §5 requires the `Settings` field, the `ENV_KEYS` entry, the
`.env.example` line and the README row to land together; here the documentation side is
compared against `config.py` and `.env.example`, so they cannot disagree.
"""
from __future__ import annotations
import re
import tomllib
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
import pytest
from undercooked.config import ENV_KEYS
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
README = REPO_ROOT / "README.md"
ENV_EXAMPLE = REPO_ROOT / ".env.example"
PYPROJECT = REPO_ROOT / "pyproject.toml"
#: Markdown documents whose command blocks a human may copy.
DOCUMENTED_MARKDOWN: tuple[Path, ...] = (README, *sorted((REPO_ROOT / "docs").glob("*.md")))
#: Shell fences that promise "type this".
COMMAND_FENCE_LANGUAGES = frozenset({"bash", "sh", "console", "shell"})
#: The only things a documented command line may start with. Anything else is prose
#: wearing a code fence, and it will be deleted rather than run.
SAFE_PREFIXES = ("uv ", "podman ", "bash .agents/validate.sh", "curl ", "cd ", "#")
#: `FOO=bar cmd …` is still a command; only documented variables may lead a line.
ENV_ASSIGNMENT = re.compile(r"""^(?:[A-Z][A-Z0-9_]*=(?:"[^"]*"|'[^']*'|\S)+\s+)+""")
#: Every `UNDERCOOKED_*` / `DEBUGPY*` name written anywhere in prose, a table or a fence.
#: A name has to start with a real character, so the `UNDERCOOKED_*` wildcard in prose is
#: not read as a variable nobody implemented.
DOCUMENTED_VARIABLE = re.compile(r"\bUNDERCOOKED_[A-Z0-9_]*[A-Z0-9]\b|\bDEBUGPY(?:_[A-Z0-9]+)?\b")
#: `uv run` options that consume the token after them, so the program name is further along.
UV_VALUE_OPTIONS = frozenset({"--extra", "--python", "--with", "--directory", "--project", "-m"})
#: `KEY=value` lines of an env template.
_TEMPLATE_VARIABLE = re.compile(r"^([A-Z][A-Z0-9_]*)\s*=")
@dataclass(frozen=True, slots=True)
class Command:
"""One logical command line from a fenced block (continuations joined)."""
document: str
line: int
text: str
def __str__(self) -> str:
return f"{self.document}:{self.line}: {self.text}"
def _strip_assignment(command: str) -> tuple[tuple[str, ...], str]:
"""Split `A=1 B=2 uv run …` into its variable names and the command underneath."""
names: list[str] = []
while match := ENV_ASSIGNMENT.match(command):
names.extend(part.split("=", 1)[0] for part in match.group(0).split())
command = command[match.end(0) :].lstrip()
return tuple(names), command
def fenced_commands(markdown: str, *, document: str = "<memory>") -> list[Command]:
"""Every command inside a `bash`-family fence, with the line it started on.
A trailing-backslash continuation becomes one `Command`, because only the first line
of a wrapped command carries the program name.
"""
commands: list[Command] = []
fence: str | None = None
pending: list[str] = []
pending_line = 0
for number, raw in enumerate(markdown.splitlines(), start=1):
marker = re.match(r"^\s*```(\w*)\s*$", raw)
if marker:
fence = None if fence is not None else marker.group(1).lower()
if pending:
commands.append(Command(document, pending_line, " ".join(pending)))
pending = []
continue
if fence is None or fence not in COMMAND_FENCE_LANGUAGES:
continue
stripped = raw.strip()
if not stripped:
if pending:
commands.append(Command(document, pending_line, " ".join(pending)))
pending = []
continue
if not pending:
pending_line = number
if stripped.endswith("\\"):
pending.append(stripped[:-1].strip())
continue
pending.append(stripped)
commands.append(Command(document, pending_line, " ".join(pending)))
pending = []
if pending: # pragma: no cover - a document ending mid-continuation
commands.append(Command(document, pending_line, " ".join(pending)))
return commands
def uv_run_program(command: str) -> str | None:
"""The program a `uv run …` line launches, with uv's own options skipped; else `None`.
That program must be declared in `pyproject.toml`, which is what makes a doc line
inventing a script fail the build.
"""
if not command.startswith("uv run"):
return None
tokens = command.split()[2:]
index = 0
while index < len(tokens):
token = tokens[index]
if token in UV_VALUE_OPTIONS:
index += 2
continue
if token.startswith("-"):
index += 1
continue
return token.lower()
return None
def declared_uv_programs() -> frozenset[str]:
"""Program names `uv run` may launch: this project's scripts plus its dependencies."""
def names(requirements: list[str]) -> set[str]:
return {re.split(r"[<>=!;\[ ]", name)[0].strip().lower() for name in requirements}
data = tomllib.loads(PYPROJECT.read_text(encoding="utf-8"))
project = data["project"]
packages = names(list(project.get("dependencies", [])))
for extra in project.get("optional-dependencies", {}).values():
packages |= names(list(extra))
return frozenset(set(project.get("scripts", {})) | packages | {"python"})
def template_variables() -> set[str]:
"""Names declared in the committed `.env.example` template (comments ignored)."""
return {
match.group(1)
for line in ENV_EXAMPLE.read_text(encoding="utf-8").splitlines()
if (match := _TEMPLATE_VARIABLE.match(line.strip()))
}
def variables_in(path: Path) -> set[str]:
"""Every configuration variable named anywhere in one markdown file."""
return set(DOCUMENTED_VARIABLE.findall(path.read_text(encoding="utf-8")))
def _fail(header: str, offenders: list[str]) -> None:
pytest.fail(f"{header}\n" + "\n".join(f" {offender}" for offender in offenders))
def unsafe_commands(commands: Iterable[Command]) -> list[str]:
"""Commands whose first token is none of `SAFE_PREFIXES`, once env assignments are peeled."""
offenders: list[str] = []
for command in commands:
_, bare = _strip_assignment(command.text)
if not bare.startswith(SAFE_PREFIXES):
offenders.append(str(command))
return offenders
def undeclared_programs(commands: Iterable[Command], allowed: frozenset[str]) -> list[str]:
"""`uv run <thing>` lines whose `<thing>` pyproject.toml does not declare."""
offenders: list[str] = []
for command in commands:
_, bare = _strip_assignment(command.text)
program = uv_run_program(bare)
if program is not None and program not in allowed:
offenders.append(f"{command} -> {program!r} is not in pyproject.toml")
return offenders
def documented_but_unimplemented(documented: set[str], implemented: set[str]) -> list[str]:
"""Variables the docs name and `config.py` does not read."""
return sorted(documented - implemented)
def implemented_but_undocumented(implemented: set[str], documented: set[str]) -> list[str]:
"""Variables `config.py` reads and the docs never mention."""
return sorted(set(implemented) - documented)
# --------------------------------------------------------------------------- command blocks
def test_readme_has_command_blocks_worth_checking() -> None:
"""Non-vacuous: if the Quickstart stops being a bash fence, this has to notice."""
commands = fenced_commands(README.read_text(encoding="utf-8"), document="README.md")
assert len(commands) >= 8, f"README.md has lost its command blocks: {commands}"
assert any(command.text.startswith("uv run undercooked") for command in commands)
assert any(command.text.startswith("uv run pytest") for command in commands)
def test_documented_command_lines_start_with_a_known_prefix() -> None:
"""No pseudo-commands in the docs a player or an operator copy-pastes from."""
offenders: list[str] = []
for document in DOCUMENTED_MARKDOWN:
commands = fenced_commands(document.read_text(encoding="utf-8"), document=document.name)
offenders.extend(unsafe_commands(commands))
if offenders:
_fail(f"documented commands must start with one of {SAFE_PREFIXES}:", offenders)
def test_documented_commands_only_name_declared_programs() -> None:
"""`uv run <thing>` may only launch a script or dependency of this project."""
allowed = declared_uv_programs()
offenders: list[str] = []
for document in DOCUMENTED_MARKDOWN:
commands = fenced_commands(document.read_text(encoding="utf-8"), document=document.name)
offenders.extend(undeclared_programs(commands, allowed))
if offenders:
_fail(f"`uv run` may only run {sorted(allowed)}:", offenders)
def test_command_fence_is_a_fence_and_not_a_table() -> None:
"""Pins the extractor: fenced bash is a command, indented prose and other fences are not."""
sample = "\n".join(
[
"text `not-a-command --yolo`", # 1: inline code is not a command
"",
"```bash", # 3
"uv run undercooked \\", # 4 \\ the continuation joins the next line
" --port 8137", # 5
"", # 6: a blank line ends a command, it never joins across it
"uv run pytest -q", # 7
"# a comment", # 8
"uv run ruff check \\", # 9 \\ a continuation cut off by a blank line still counts
"",
"uv run pyright \\", # 11 \\ and one cut off by the closing fence counts too
"```",
"",
"```text", # 14: not a command fence
"made-up-command --yolo",
"```",
]
)
assert fenced_commands(sample) == [
Command("<memory>", 4, "uv run undercooked --port 8137"),
Command("<memory>", 7, "uv run pytest -q"),
Command("<memory>", 8, "# a comment"),
Command("<memory>", 9, "uv run ruff check"),
Command("<memory>", 11, "uv run pyright"),
]
assert str(Command("README.md", 12, "uv run undercooked")) == "README.md:12: uv run undercooked"
def test_leading_environment_assignments_are_peeled_off() -> None:
names, bare = _strip_assignment("DEBUGPY=1 UNDERCOOKED_PORT=8137 uv run undercooked")
assert names == ("DEBUGPY", "UNDERCOOKED_PORT")
assert bare == "uv run undercooked"
@pytest.mark.parametrize(
("command", "program"),
[
("uv run pytest -q", "pytest"),
("uv run --extra dev ruff check .", "ruff"),
("uv run undercooked --reload", "undercooked"),
("uv run python -c 'print(1)'", "python"),
],
)
def test_the_uv_run_extractor_finds_the_program(command: str, program: str) -> None:
assert uv_run_program(command) == program
@pytest.mark.parametrize("command", ["podman build .", "uv sync --extra dev", "cd .", "uv run", ""])
def test_the_uv_run_extractor_returns_none_for_everything_else(command: str) -> None:
assert uv_run_program(command) is None
def test_the_declared_programs_cover_the_real_toolchain() -> None:
"""Derived from pyproject.toml, so a dependency removal breaks the docs loudly."""
allowed = declared_uv_programs()
assert {"undercooked", "pytest", "ruff", "pyright", "playwright", "python"} <= allowed
assert "undercooked-levels" in allowed # declared now, importable from phase 02
# --------------------------------------------------------------------------- variable names
def test_every_variable_named_in_the_docs_is_implemented() -> None:
"""README and docs may only name a variable `config.py` or `.env.example` defines."""
implemented = set(ENV_KEYS.values()) | template_variables()
offenders = [
f"{document.name}: {name}"
for document in DOCUMENTED_MARKDOWN
for name in documented_but_unimplemented(variables_in(document), implemented)
]
if offenders:
_fail("documented but not implemented:", offenders)
def test_every_implemented_variable_is_documented() -> None:
"""The reverse direction: a knob nobody documented is a knob nobody can find."""
documented = variables_in(README) | variables_in(ENV_EXAMPLE)
missing = implemented_but_undocumented(set(ENV_KEYS.values()), documented)
if missing:
_fail("config.py defines variables neither README.md nor .env.example mentions:", missing)
def test_readme_and_env_template_name_the_same_variables() -> None:
"""The two operator-facing lists cannot diverge from each other."""
documented, template = variables_in(README), variables_in(ENV_EXAMPLE)
only_readme = sorted(documented - template)
only_template = sorted(template - documented)
assert not only_readme, f"named in README.md but absent from .env.example: {only_readme}"
assert not only_template, f"named in .env.example but absent from README.md: {only_template}"
# --------------------------------------------------------------------------- the gate is documented
def test_readme_documents_the_gate_and_the_cdn_rule() -> None:
"""The commands AGENTS.md calls binding are findable in the human-facing docs."""
readme = README.read_text(encoding="utf-8")
assert "bash .agents/validate.sh" in readme, "the pipeline gate is not documented in README"
assert re.search(r"no external cdns", readme, re.IGNORECASE), "A12 is not stated in README"
def test_a_pseudo_command_is_reported() -> None:
"""The prefix rule actually fires: planted prose-commands are named, real ones are not."""
sample = "\n".join(
[
"```bash", # 1
"uv run undercooked --port 8137", # 2
"DEBUGPY=1 uv run undercooked", # 3: an env lead is still a command
"make deploy-to-staging", # 4
"kubectl port-forward svc/undercooked 8000", # 5
"# a comment is a valid line", # 6
"```",
]
)
offenders = unsafe_commands(fenced_commands(sample, document="<memory>"))
assert offenders == [
"<memory>:4: make deploy-to-staging",
"<memory>:5: kubectl port-forward svc/undercooked 8000",
]
def test_an_invented_program_is_reported() -> None:
"""A doc line cannot run a script pyproject.toml does not declare."""
sample = "\n".join(
[
"```bash",
"uv run --frozen undercooked-levels validate",
"uv run made-up-cli validate",
"```",
]
)
commands = fenced_commands(sample, document="<memory>")
offenders = undeclared_programs(commands, frozenset({"undercooked-levels"}))
assert len(offenders) == 1, offenders
assert "made-up-cli" in offenders[0]
def test_the_variable_diffs_work_in_both_directions() -> None:
"""The two drift directions are tested, not just asserted to be empty on today's docs."""
implemented = {"UNDERCOOKED_PORT", "DEBUGPY"}
documented = {"UNDERCOOKED_PORT", "DEBUGPY", "UNDERCOOKED_MADE_UP", "DEBUGPY_PORT"}
assert documented_but_unimplemented(documented, implemented) == [
"DEBUGPY_PORT",
"UNDERCOOKED_MADE_UP",
]
assert implemented_but_undocumented(implemented, documented) == []
assert implemented_but_undocumented({"UNDERCOOKED_NEW_KNOB"}, documented) == [
"UNDERCOOKED_NEW_KNOB"
]
def test_the_failure_helper_lists_every_offender() -> None:
"""`_fail` is the message an executor reads; it has to enumerate, not truncate."""
with pytest.raises(pytest.fail.Exception, match="two"):
_fail("broken:", ["one", "two"])
def test_agents_md_only_names_test_paths_that_exist() -> None:
"""Every concrete `tests/**.py|md` path AGENTS.md cites is a real file."""
agents = (REPO_ROOT / "AGENTS.md").read_text(encoding="utf-8")
cited = re.findall(r"tests/[A-Za-z0-9_./-]+\.(?:py|md)", agents)
assert "tests/unit/test_no_external_assets.py" in cited
assert cited, "AGENTS.md cites no test files at all"
missing = sorted({path for path in cited if not (REPO_ROOT / path).is_file()})
assert missing == [], f"AGENTS.md points at files that do not exist: {missing}"
+357
View File
@@ -0,0 +1,357 @@
"""PLAN §13 doc-index guard: the promised documentation set and the `docs/` tree cannot drift.
`.agents/PLAN.md` §13 closes with the canonical documentation index and the sentence
"`tests/unit/test_docs_index.py` (phase 01) keeps this table and the `docs/` tree identical,
so neither can drift silently." This is that file. It closes both directions of drift:
* **A promised doc is missing.** Every index row whose creating phase has already completed
(according to the `.agents/phases/complete/` tree) must exist on disk, so no phase can
reference a document that was never written. A range in the "Created by" column —
`phase 02/03` — promises the file by the *end* of that range, so its obligation starts at
the later phase; a `→` marks the stub-then-fill pattern, where the earlier phase owes the
stub.
* **A doc exists that nothing promises.** Every `docs/*.md` must be listed in the index or be
a document a phase file of the roadmap itself creates — `docs/ROOMS.md` comes from phase 05
and is not in §13, so the roadmap may extend the doc set, but an untracked file may not
appear there silently.
Only markdown is compared: the index promises *documents*, and a diagram or fixture a phase
puts under `docs/` is not a documentation promise.
"""
from __future__ import annotations
import re
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
import pytest
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
PLAN = REPO_ROOT / ".agents/PLAN.md"
DOCS_DIR = REPO_ROOT / "docs"
PHASES_DIR = REPO_ROOT / ".agents/phases"
COMPLETE_DIR = PHASES_DIR / "complete"
INDEX_HEADING = "### Documentation index (canonical)"
PHASE_DIRECTORY = re.compile(r"^(\d{2})_")
PHASE_REFERENCE = re.compile(r"phase\s+(\d{1,2})", re.IGNORECASE)
NUMBER = re.compile(r"\d{1,2}")
DOC_REFERENCE = re.compile(r"docs/[A-Za-z0-9_-]+\.md")
#: The index exactly as PLAN §13 writes it. Changing this set means editing PLAN.md first.
CANONICAL_DOCS = frozenset(
{
"README.md",
"AGENTS.md",
"BUDGETS.md",
"docs/DESIGN.md",
"docs/SIMULATION.md",
"docs/LEVELS.md",
"docs/PROTOCOL.md",
"docs/CLIENT.md",
"docs/DEPLOYMENT.md",
"docs/TESTING.md",
"docs/ACCESSIBILITY.md",
"docs/SECURITY.md",
"tests/e2e/README.md",
}
)
@dataclass(frozen=True, slots=True)
class DocRow:
"""One row of the PLAN §13 index: path, contents, and the phase(s) that create it."""
path: str
contents: str
created_by: str
@property
def promised_from(self) -> int:
"""First phase after which this file must exist.
The first comma/arrow-separated group is the creation promise. Inside it, a slash
range (`phase 02/03`) is owed by the end of the range, so the later number wins;
anything else is owed by the earliest number mentioned.
"""
first_group = re.split(r"[→,;]", self.created_by, maxsplit=1)[0]
numbers = [int(number) for number in NUMBER.findall(first_group)]
if not numbers:
numbers = [int(number) for number in PHASE_REFERENCE.findall(self.created_by)]
if not numbers:
pytest.fail(f"PLAN index row {self.path!r} names no phase: {self.created_by!r}")
return numbers[-1] if "/" in first_group and len(numbers) > 1 else numbers[0]
def parse_documentation_index(plan_text: str) -> list[DocRow]:
"""Read the canonical documentation-index table out of PLAN.md text."""
lines = plan_text.splitlines()
try:
start = lines.index(INDEX_HEADING)
except ValueError: # pragma: no cover - PLAN is locked; fail loudly if it is renamed
pytest.fail(f".agents/PLAN.md has no {INDEX_HEADING!r} heading")
rows: list[DocRow] = []
for raw in lines[start + 1 :]:
stripped = raw.strip()
if not stripped.startswith("|"):
if rows:
break
continue
cells = [cell.strip() for cell in stripped.strip("|").split("|")]
if len(cells) < 3 or cells[0].strip("`") in {"Path", ""} or set(cells[0]) <= {"-", ":"}:
continue
rows.append(DocRow(path=cells[0].strip("`"), contents=cells[1], created_by=cells[2]))
return rows
def latest_completed_phase(complete_dir: Path = COMPLETE_DIR) -> int:
"""Highest phase number with anything in `complete/` (0 before the first phase finishes)."""
return max(
(
int(match.group(1))
for entry in complete_dir.glob("*")
if (match := PHASE_DIRECTORY.match(entry.name))
),
default=0,
)
def roadmap_promised_docs(phases_dir: Path = PHASES_DIR) -> set[str]:
"""Every `docs/*.md` a phase overview or task file itself names."""
return {
match.group(0)
for phase_file in sorted(phases_dir.rglob("*.md"))
for match in DOC_REFERENCE.finditer(phase_file.read_text(encoding="utf-8"))
}
def documentation_index(plan: Path = PLAN) -> list[DocRow]:
return parse_documentation_index(plan.read_text(encoding="utf-8"))
def written_docs(docs_dir: Path = DOCS_DIR, *, root: Path = REPO_ROOT) -> set[str]:
"""The markdown currently in `docs/`, as repo-relative paths."""
return {path.relative_to(root).as_posix() for path in sorted(docs_dir.rglob("*.md"))}
# --------------------------------------------------------------------------- the index itself
def test_the_documentation_index_is_parseable() -> None:
rows = documentation_index()
assert len(rows) >= len(CANONICAL_DOCS), rows
paths = [row.path for row in rows]
assert len(paths) == len(set(paths)), f"duplicate index rows: {paths}"
assert all(row.contents and row.created_by for row in rows), paths
def test_the_index_still_lists_the_canonical_doc_set() -> None:
"""Adding or dropping a row is a PLAN §13 change and must fail a test first."""
assert {row.path for row in documentation_index()} == CANONICAL_DOCS
@pytest.mark.parametrize(
("created_by", "expected"),
[
("phase 01", 1),
("phase 01, refined throughout", 1),
("phase 01 (stub) \u2192 phase 09", 1),
("phase 02, phase 11", 2),
("phase 02/03", 3),
("phase 14", 14),
("refined throughout, phase 05", 5),
],
)
def test_the_promised_from_rule_reads_the_phase_column(created_by: str, expected: int) -> None:
row = DocRow(path="docs/X.md", contents="whatever", created_by=created_by)
assert row.promised_from == expected
def test_an_index_row_that_names_no_phase_fails_loudly() -> None:
row = DocRow(path="docs/X.md", contents="whatever", created_by="someday, maybe")
with pytest.raises(pytest.fail.Exception, match="names no phase"):
_ = row.promised_from
# --------------------------------------------------------------------------- promised vs present
def test_every_doc_promised_by_a_completed_phase_exists() -> None:
"""No phase may reference a document whose creating phase is already in `complete/`."""
completed = latest_completed_phase()
assert completed >= 1, "no completed phase found — the guard would be vacuous"
missing = [
f"{row.path} (owed by phase {row.promised_from:02d}, {completed:02d} is complete)"
for row in documentation_index()
if row.promised_from <= completed and not (REPO_ROOT / row.path).is_file()
]
report = "\n ".join(missing)
assert missing == [], f"documentation the project owes but does not have:\n {report}"
def test_no_doc_file_exists_that_nothing_promises() -> None:
"""The `docs/` tree is exactly the index plus what the phase roadmap creates."""
promised = {row.path for row in documentation_index()} | roadmap_promised_docs()
orphans = sorted(written_docs() - promised)
report = "\n ".join(orphans)
assert orphans == [], (
f"unpromised files under docs/ — add each to PLAN §13's index or to the phase "
f"that creates it:\n {report}"
)
def test_the_docs_on_disk_are_the_three_phase_01_documents() -> None:
"""Pins today's tree so that adding a document is a deliberate, named act."""
assert written_docs() == {"docs/DESIGN.md", "docs/DEPLOYMENT.md", "docs/TESTING.md"}
def stub_without_owner(relative: str, text: str) -> str | None:
"""One sentence of complaint about a document, or `None` when it is honest.
Split out of the test below so a planted stub can prove each check fires; a guard that
only ever sees healthy input is not a guard.
"""
if not text.startswith("# "):
return f"{relative} does not open with a title"
if "## Purpose" not in text:
return f"{relative} has no Purpose section"
purpose = text.split("## Purpose", 1)[1].split("\n##", 1)[0].strip()
if len(purpose.split()) < 25:
return f"{relative}'s Purpose paragraph is too thin to guide"
names_stub = re.search(r"\bstub\b", text, re.IGNORECASE) is not None
has_owner = re.search(r"[Ff]illed in by phase \d\d", text) is not None
if names_stub and not has_owner:
return f"{relative} is a stub with no 'Filled in by phase NN' marker"
return None
def dishonest_references(
references: dict[str, set[str]], *, exists: Callable[[str], bool], owed_later: set[str]
) -> list[str]:
"""`document: path` for each cited doc that neither exists nor is owed by a later phase."""
offenders: list[str] = []
for name, cited in sorted(references.items()):
offenders.extend(
f"{name}: {path}"
for path in sorted(cited)
if not exists(path) and path not in owed_later
)
return offenders
def test_the_stub_docs_say_who_fills_them_in() -> None:
"""A stub with no owner stays a stub forever, so each one names its phase."""
complaints = [
complaint
for relative in sorted(written_docs())
if (complaint := stub_without_owner(relative, (REPO_ROOT / relative).read_text("utf-8")))
]
report = "\n ".join(complaints)
assert complaints == [], f"documents that are not self-describing:\n {report}"
def test_readme_and_agents_reference_only_honest_doc_paths() -> None:
"""A doc path in README/AGENTS either exists or is owed by a phase that has not run."""
completed = latest_completed_phase()
rows = documentation_index()
owed_later = {row.path for row in rows if row.promised_from > completed}
owed_later |= roadmap_promised_docs()
root_docs = {row.path for row in rows if not row.path.startswith("docs/")}
references: dict[str, set[str]] = {}
for name in ("README.md", "AGENTS.md"):
text = (REPO_ROOT / name).read_text(encoding="utf-8")
references[name] = set(DOC_REFERENCE.findall(text)) | {
path for path in root_docs if path in text
}
offenders = dishonest_references(
references, exists=lambda path: (REPO_ROOT / path).is_file(), owed_later=owed_later
)
report = "\n ".join(offenders)
assert offenders == [], f"references to documentation nothing promises to create:\n {report}"
# --------------------------------------------------------------------------- the guard itself
def test_a_missing_promised_doc_is_named(tmp_path: Path) -> None:
"""Plant a row whose phase already shipped: the parse must expose it as owed."""
plan = tmp_path / "PLAN.md"
plan.write_text(
"\n".join(
[
"## Risks",
"",
INDEX_HEADING,
"",
"| Path | Contents | Created by |",
"|---|---|---|",
"| `docs/GONE.md` | whatever it says | phase 01 |",
"| `docs/LATER.md` | not owed yet | phase 09 |",
"",
"prose after the table",
]
),
encoding="utf-8",
)
rows = parse_documentation_index(plan.read_text(encoding="utf-8"))
assert [row.path for row in rows] == ["docs/GONE.md", "docs/LATER.md"]
owed = [row.path for row in rows if row.promised_from <= 1]
assert owed == ["docs/GONE.md"]
assert not (tmp_path / owed[0]).is_file()
def test_an_unpromised_doc_file_is_reported(tmp_path: Path) -> None:
"""The orphan rule works on a temp tree too — it is not README-specific magic."""
(tmp_path / "docs").mkdir()
(tmp_path / "docs" / "WANTED.md").write_text("# WANTED\n", encoding="utf-8")
(tmp_path / "docs" / "MYSTERY.md").write_text("# MYSTERY\n", encoding="utf-8")
(tmp_path / "docs" / "tile.svg").write_text("<svg/>", encoding="utf-8")
assert written_docs(tmp_path, root=tmp_path) == {
"docs/MYSTERY.md",
"docs/WANTED.md",
}
assert written_docs(tmp_path, root=tmp_path) - {"docs/WANTED.md"} == {"docs/MYSTERY.md"}
def test_an_unhonest_document_is_described_precisely() -> None:
"""Each complaint the stub check can make is exercised on planted text."""
body = "word " * 30
good = f"# Title\n\n## Purpose\n{body}\nA stub here. Filled in by phase 09.\n"
assert stub_without_owner("docs/GOOD.md", good) is None
assert stub_without_owner("docs/ORPHAN.md", good.replace(" Filled in by phase 09.", "")) == (
"docs/ORPHAN.md is a stub with no 'Filled in by phase NN' marker"
)
assert stub_without_owner("docs/UNTITLED.md", f"## Purpose\n{body}") == (
"docs/UNTITLED.md does not open with a title"
)
assert stub_without_owner("docs/NOPURPOSE.md", "# Title\nProse only.\n") == (
"docs/NOPURPOSE.md has no Purpose section"
)
assert stub_without_owner("docs/THIN.md", "# Title\n\n## Purpose\ntoo short\n") == (
"docs/THIN.md's Purpose paragraph is too thin to guide"
)
def test_a_reference_to_a_doc_nobody_promises_is_named() -> None:
references = {
"AGENTS.md": {"docs/HERE.md"},
"README.md": {"docs/HERE.md", "docs/GHOST.md", "docs/OWED.md"},
}
offenders = dishonest_references(
references,
exists=lambda path: path in {"docs/HERE.md", "docs/OWED.md"},
owed_later={"docs/OWED.md"},
)
assert offenders == ["README.md: docs/GHOST.md"]
def test_the_roadmap_extension_of_the_doc_set_is_tracked() -> None:
"""`docs/ROOMS.md` is phase 05's, not PLAN §13's — and must not read as an orphan."""
roadmap = roadmap_promised_docs()
assert "docs/ROOMS.md" in roadmap
assert "docs/ROOMS.md" not in CANONICAL_DOCS
assert latest_completed_phase() >= 1
+90
View File
@@ -0,0 +1,90 @@
"""Phase-01 task-01 harness smoke tests: package metadata and pytest wiring."""
from __future__ import annotations
import os
import re
import subprocess
import sys
from pathlib import Path
import pytest
import undercooked
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
HARNESS_MARKERS = ("unit", "integration", "e2e", "slow")
TEST_SUBDIRS = ("unit", "integration", "sim", "e2e")
def test_package_metadata() -> None:
assert undercooked.__version__ == "0.1.0"
assert undercooked.__all__ == []
def test_harness_markers_registered(pytestconfig: pytest.Config) -> None:
registered = {line.split(":", 1)[0].strip() for line in pytestconfig.getini("markers")}
assert set(HARNESS_MARKERS) <= registered
def test_strict_markers_enabled(pytestconfig: pytest.Config) -> None:
assert "--strict-markers" in pytestconfig.getini("addopts")
@pytest.mark.parametrize("name", TEST_SUBDIRS)
def test_test_tree_directories_exist_with_readme(name: str) -> None:
directory = REPO_ROOT / "tests" / name
assert directory.is_dir()
readme = directory / "README.md"
assert readme.is_file()
assert readme.read_text(encoding="utf-8").strip()
def test_tests_dir_is_not_a_package() -> None:
assert not (REPO_ROOT / "tests" / "__init__.py").exists()
def test_fixture_levels_dir_exists(isolated_levels_dir: None) -> None:
levels_dir = Path(os.environ["UNDERCOOKED_LEVELS_DIR"])
assert levels_dir == REPO_ROOT / "tests" / "fixtures" / "levels"
assert levels_dir.is_dir()
assert not list(levels_dir.glob("*.yaml"))
def test_e2e_tests_follow_the_story_naming_convention() -> None:
"""`pytest -m e2e --collect-only` selects only `test_usNN_*.py` story suites.
Task 01 could only assert "the marker collects nothing yet". Now that the first
browser suite exists the durable invariant is the naming contract in
`tests/e2e/README.md`: a story suite that is not named for its story is one that
nobody can find, and one that cannot be collected is one that never runs.
"""
result = subprocess.run( # fixed argv, repo's own interpreter
[
sys.executable,
"-m",
"pytest",
"-m",
"e2e",
"--collect-only",
"-q",
"-o",
"addopts=", # drop the outer coverage flags: no nested .coverage writes
"-p",
"no:cacheprovider",
],
cwd=REPO_ROOT,
capture_output=True,
text=True,
check=False,
timeout=120,
)
assert result.returncode == pytest.ExitCode.OK, result.stdout + result.stderr
assert "error" not in result.stderr.lower(), result.stderr
files = sorted({line.split("::")[0] for line in result.stdout.splitlines() if "::" in line})
assert files, f"the e2e marker selected nothing:\n{result.stdout}"
for path in files:
assert path.startswith("tests/e2e/"), path
assert re.fullmatch(r"test_us\d{2}_.+\.py", Path(path).name), path
+105
View File
@@ -0,0 +1,105 @@
"""Unit tests for the security headers (PLAN.md §11, A12).
These assert the *values*, not just their presence: a header that silently
weakened (a `unsafe-inline`, a `*`, a reporting endpoint) is a security
regression dressed up as a passing test.
"""
from __future__ import annotations
import pytest
from starlette.responses import JSONResponse, PlainTextResponse, Response
from undercooked.server.headers import (
CONTENT_SECURITY_POLICY,
SECURITY_HEADERS,
SecurityHeadersMiddleware,
apply_security_headers,
)
pytestmark = pytest.mark.unit
EXPECTED = {
"Content-Security-Policy": CONTENT_SECURITY_POLICY,
"X-Content-Type-Options": "nosniff",
"Referrer-Policy": "no-referrer",
"X-Frame-Options": "DENY",
"Permissions-Policy": "camera=(), microphone=(), geolocation=()",
}
def test_the_five_documented_headers_are_defined() -> None:
assert dict(SECURITY_HEADERS) == EXPECTED
def test_apply_adds_every_header_to_a_response() -> None:
response = PlainTextResponse("hi")
apply_security_headers(response)
for header, value in EXPECTED.items():
assert response.headers[header] == value
def test_apply_works_on_json_and_empty_bodies() -> None:
payload = JSONResponse({"a": 1})
apply_security_headers(payload)
assert payload.headers["content-security-policy"] == CONTENT_SECURITY_POLICY
empty = Response(status_code=204)
apply_security_headers(empty)
assert empty.headers["x-content-type-options"] == "nosniff"
def test_apply_is_idempotent_and_overrides_weaker_values() -> None:
response = PlainTextResponse("hi", headers={"X-Frame-Options": "SAMEORIGIN"})
apply_security_headers(response)
apply_security_headers(response)
assert response.headers["x-frame-options"] == "DENY"
def _csp() -> dict[str, str]:
"""The CSP as `directive -> whole clause`, so assertions read like the header."""
return {part.split(" ", 1)[0]: part for part in CONTENT_SECURITY_POLICY.split("; ")}
def test_csp_locks_scripts_and_styles_to_self() -> None:
directives = _csp()
assert directives["default-src"] == "default-src 'self'"
assert directives["script-src"] == "script-src 'self'"
assert directives["style-src"] == "style-src 'self'"
assert directives["font-src"] == "font-src 'self'"
assert directives["base-uri"] == "base-uri 'none'"
assert directives["frame-ancestors"] == "frame-ancestors 'none'"
def test_csp_allows_only_same_origin_websockets_and_data_images() -> None:
directives = _csp()
assert directives["connect-src"] == "connect-src 'self' ws: wss:"
assert directives["img-src"] == "img-src 'self' data:"
@pytest.mark.parametrize(
"forbidden", ["unsafe-inline", "unsafe-eval", " https://", "http://", "*", "report-uri"]
)
def test_csp_contains_no_unsafe_source_or_remote_endpoint(forbidden: str) -> None:
assert forbidden not in CONTENT_SECURITY_POLICY
def test_no_header_points_at_a_remote_reporting_service() -> None:
"""`Report-To`/`report-uri` would phone a third party: A12 forbids it."""
assert not any(name.lower() in {"report-to", "nel"} for name in SECURITY_HEADERS)
for value in SECURITY_HEADERS.values():
assert "://" not in value
assert "report" not in value.lower()
def test_headers_are_immutable() -> None:
with pytest.raises(TypeError):
SECURITY_HEADERS["X-Frame-Options"] = "ALLOWALL" # type: ignore[index]
def test_middleware_is_constructible_with_an_app_stub() -> None:
async def stub_app(scope: object, receive: object, send: object) -> None: # pragma: no cover
return None
middleware = SecurityHeadersMiddleware(stub_app)
assert middleware.app is stub_app
+257
View File
@@ -0,0 +1,257 @@
"""Unit tests for the `undercooked` CLI (task 03): precedence, exit codes, wiring.
`_resolve` and `_uvicorn_log_config` are pure, so precedence is testable without
a socket. `main()` is tested with uvicorn's `run` replaced by a recorder — the
real server is started by the integration/phase criteria, not by a unit test.
"""
from __future__ import annotations
import copy
import os
import subprocess
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pytest
import uvicorn
import uvicorn.config
from undercooked.__main__ import (
APP_IMPORT_STRING,
EXIT_CONFIG_ERROR,
EXIT_OK,
RunOptions,
_cli_overrides,
_resolve,
_uvicorn_log_config,
build_parser,
main,
)
from undercooked.config import (
ENV_KEYS,
ConfigError,
Settings,
apply_cli_overrides,
get_settings,
reset_settings_cache,
)
if TYPE_CHECKING:
from collections.abc import Iterator
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
# A string in an assertion, never a socket bind: the CLI flag is what is under test.
ALL_INTERFACES = "0.0.0.0" # noqa: S104 - a test assertion, not a socket bind
@pytest.fixture(autouse=True)
def pristine_env() -> Iterator[None]:
"""`main()` publishes CLI flags into `os.environ`; nothing may leak out."""
saved = dict(os.environ)
reset_settings_cache()
yield
os.environ.clear()
os.environ.update(saved)
reset_settings_cache()
@pytest.fixture
def uvicorn_calls(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
"""Replace uvicorn's `run` with a recorder; returns the recorded invocations."""
calls: list[dict[str, Any]] = []
def fake_run(target: str, **kwargs: Any) -> None:
calls.append({"target": target, **kwargs})
monkeypatch.setattr(uvicorn, "run", fake_run)
return calls
@pytest.fixture
def bare_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""No `UNDERCOOKED_*`/`DEBUGPY*` in the environment: defaults are observable."""
for key in list(os.environ):
if key.startswith(("UNDERCOOKED_", "DEBUGPY")):
monkeypatch.delenv(key, raising=False)
monkeypatch.setenv("PYTHON_DOTENV_DISABLED", "true")
def parse(argv: list[str]) -> Any:
"""Parse ``argv`` the way `main` would, without serving anything."""
return build_parser().parse_args(argv)
# ----------------------------------------------------------------------------- parser
def test_every_flag_defaults_to_unset() -> None:
args = parse([])
assert (args.host, args.port, args.reload, args.levels_dir) == (None, None, False, None)
def test_flags_are_parsed() -> None:
args = parse(
["--host", ALL_INTERFACES, "--port", "9000", "--reload", "--levels-dir", "kitchens"]
)
assert (args.host, args.port, args.reload) == (ALL_INTERFACES, 9000, True)
assert args.levels_dir == Path("kitchens")
def test_help_names_the_env_fallback() -> None:
text = build_parser().format_help()
assert "--levels-dir" in text
assert "UNDERCOOKED_PORT" in text
# ----------------------------------------------------------------------------- precedence
def test_cli_beats_environment_and_defaults() -> None:
settings = Settings(host="10.0.0.1", port=8123, levels_dir=Path("env-levels"))
options = _resolve(
parse(["--host", "127.0.0.1", "--port", "9999", "--levels-dir", "cli-levels"]), settings
)
assert options == RunOptions(
host="127.0.0.1", port=9999, reload=False, levels_dir=Path("cli-levels")
)
def test_settings_supply_everything_the_cli_left_unset() -> None:
settings = Settings(host=ALL_INTERFACES, port=8123, levels_dir=Path("env-levels"))
options = _resolve(parse([]), settings)
assert options == RunOptions(
host=ALL_INTERFACES, port=8123, reload=False, levels_dir=Path("env-levels")
)
def test_settings_default_to_the_documented_defaults_in_a_bare_environment(
bare_env: None,
) -> None:
options = _resolve(parse([]), get_settings())
assert options == RunOptions(
host="127.0.0.1", port=8000, reload=False, levels_dir=Path("levels")
)
def test_environment_beats_the_builtin_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("UNDERCOOKED_PORT", "8123")
monkeypatch.setenv("UNDERCOOKED_HOST", ALL_INTERFACES)
reset_settings_cache()
options = _resolve(parse([]), get_settings())
assert (options.host, options.port) == (ALL_INTERFACES, 8123)
def test_reload_is_a_flag_not_an_environment_variable() -> None:
assert _resolve(parse(["--reload"]), Settings()).reload is True
assert _resolve(parse([]), Settings()).reload is False
def test_cli_overrides_only_report_flags_that_were_given() -> None:
assert _cli_overrides(parse([])) == {}
assert _cli_overrides(parse(["--port", "9000", "--levels-dir", "x"])) == {
"port": 9000,
"levels_dir": Path("x"),
}
# ----------------------------------------------------------------------------- main()
def test_main_serves_the_app_string_with_resolved_options(
uvicorn_calls: list[dict[str, Any]], bare_env: None
) -> None:
assert main([]) == EXIT_OK
assert len(uvicorn_calls) == 1
call = uvicorn_calls[0]
assert call["target"] == APP_IMPORT_STRING # reload-capable "module:attribute" form
assert (call["host"], call["port"], call["reload"]) == ("127.0.0.1", 8000, False)
assert "undercooked" in call["log_config"]["loggers"]
def test_main_passes_cli_flags_to_uvicorn(
uvicorn_calls: list[dict[str, Any]], bare_env: None
) -> None:
assert main(["--host", ALL_INTERFACES, "--port", "9123", "--reload"]) == EXIT_OK
call = uvicorn_calls[0]
assert (call["host"], call["port"], call["reload"]) == (ALL_INTERFACES, 9123, True)
def test_main_publishes_cli_flags_to_the_environment(
uvicorn_calls: list[dict[str, Any]], bare_env: None
) -> None:
"""A uvicorn --reload child re-reads the environment, so flags must live there."""
assert main(["--levels-dir", "kitchens", "--port", "9124"]) == EXIT_OK
assert os.environ[ENV_KEYS["levels_dir"]] == "kitchens"
assert os.environ[ENV_KEYS["port"]] == "9124"
assert get_settings().levels_dir == Path("kitchens")
assert get_settings().port == 9124
def test_main_validates_cli_values_like_environment_values(
uvicorn_calls: list[dict[str, Any]], capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["--port", "0"]) == EXIT_CONFIG_ERROR
assert uvicorn_calls == []
err = capsys.readouterr().err
assert err.strip().startswith("undercooked:")
assert "UNDERCOOKED_PORT" in err
assert err.count("\n") == 1 # a one-line message, not a traceback
def test_main_reports_an_invalid_environment(
uvicorn_calls: list[dict[str, Any]],
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
monkeypatch.setenv("UNDERCOOKED_TICK_HZ", "99")
reset_settings_cache()
assert main([]) == EXIT_CONFIG_ERROR
assert uvicorn_calls == []
assert "UNDERCOOKED_TICK_HZ=99" in capsys.readouterr().err
def test_unknown_flag_is_an_argument_error() -> None:
with pytest.raises(SystemExit) as excinfo:
main(["--nope"])
assert excinfo.value.code == EXIT_CONFIG_ERROR
def test_an_unknown_settings_field_is_a_config_error() -> None:
with pytest.raises(ConfigError, match="not a configuration field"):
apply_cli_overrides(**{"nonsense": 1})
# ----------------------------------------------------------------------------- logging
def test_uvicorn_log_config_adds_our_loggers_without_touching_uvicorns() -> None:
base = copy.deepcopy(uvicorn.config.LOGGING_CONFIG)
config = _uvicorn_log_config(base)
assert config is not base
assert set(base["loggers"]) == {"uvicorn", "uvicorn.error", "uvicorn.access"}
ours = config["loggers"]["undercooked"]
assert ours["handlers"] == ["default"]
assert ours["level"] == "INFO"
assert config["loggers"]["uvicorn"] is base["loggers"]["uvicorn"]
assert set(config["formatters"]) == set(base["formatters"]) # the handler exists
# ----------------------------------------------------------------------------- the console script
def test_python_m_undercooked_prints_usage_and_exits_zero() -> None:
result = subprocess.run( # fixed argv, the repo's own interpreter
[sys.executable, "-m", "undercooked", "--help"],
cwd=REPO_ROOT,
capture_output=True,
text=True,
check=False,
timeout=120,
)
assert result.returncode == EXIT_OK, result.stderr
assert "usage: undercooked" in result.stdout
assert "--reload" in result.stdout
+392
View File
@@ -0,0 +1,392 @@
"""PLAN A12 guard: the browser must load nothing this process does not serve.
This is the mechanical half of "no external CDNs" (LOCKED decision A12, AGENTS.md §5).
It has two halves on purpose:
1. **A text scan** of every `app/static/**` asset (`.html`, `.css`, `.js`, `.svg`) for the
shapes that reach off-origin: `http(s)://`, protocol-relative `//host/`, hosted-library
mirrors (`jsdelivr`, `unpkg`, …), CSS `@import url(`, protocol-relative stylesheets,
`srcset`, `fetch()` / `import` of an absolute URL, an absolute-URL `WebSocket`, and
`EventSource` (SSE is not this project's transport). A single remote reference is both
an availability and a privacy defect, so the scan is blunt and needs no allow-list —
with one documented exception, the `xmlns` namespace URIs of SVG/XML, which are
identifiers the browser never fetches (A11 ships in-repo SVG art).
2. **An element check** of `index.html`: every `<script src>`, `<link href>` and
`<img src>` must be a same-origin `/static/...` path (a `data:` URL is allowed where
the CSP allows it — `img-src 'self' data:` — because it is the app's own bytes).
The root it scans is imported from `undercooked.server.app.STATIC_DIR`, i.e. the very
directory the server mounts at `/static`, so the guard cannot pass by scanning some other
folder. The last group of tests plants each forbidden shape in a temporary tree and
asserts the scan names the offending file, which is what keeps this file from rotting
into a test that scans nothing.
"""
from __future__ import annotations
import re
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
import pytest
from undercooked.config import Settings
from undercooked.server.app import STATIC_DIR, resolve_static_dir
pytestmark = pytest.mark.unit
REPO_ROOT = Path(__file__).resolve().parents[2]
#: Every file type a browser can be told to load or parse.
ASSET_SUFFIXES = frozenset({".html", ".css", ".js", ".svg"})
#: Files whose text may name a hosted library. CSS/SVG are covered by the URL rules below;
#: HTML and JS are where a `<script>`/`import` of a hosted bundle actually shows up.
HOSTED_TOKEN_SUFFIXES = frozenset({".html", ".js"})
#: Hosted-library / hosted-font mirrors that must never appear in shipped client text.
HOSTED_LIBRARY_TOKENS = ("googleapis", "gstatic", "jsdelivr", "unpkg", "cdnjs", "tailwindcss")
#: An absolute or protocol-relative authority in a URL-ish position.
_ABS = r"(?:https?:)?//"
#: XML namespace declarations: identifiers, not requests (SVG art is first-class, A11).
_NAMESPACE_DECL = re.compile(
r"""xmlns(?::[\w.-]+)?\s*=\s*["']?https?://www\.w3\.org/[^\s"'<>]*["']?""",
re.IGNORECASE,
)
#: (rule name, pattern) pairs applied to each masked line of every asset.
BANNED_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
(
"absolute URL",
re.compile(r"https?://", re.IGNORECASE),
),
(
"protocol-relative URL",
# `//host.tld/...` — the scheme is deliberately NOT part of this pattern, so the
# lookbehind really does keep `https://` from being reported twice, and the
# host-shaped lookahead keeps JS/CSS comments (`// note`) from matching at all.
re.compile(r"(?<![A-Za-z0-9:])//(?![\s/])(?:[\w-]+\.)+[a-z]{2,}", re.IGNORECASE),
),
(
"CSS @import url()",
re.compile(r"@import\s+url\s*\(", re.IGNORECASE),
),
(
"protocol-relative stylesheet",
re.compile(
rf"""<link\b[^>]*?rel=["']?[^"'<>]*stylesheet[^>"']*["']?[^>]*?href=["']?{_ABS}""",
re.IGNORECASE,
),
),
(
"srcset attribute",
re.compile(r"\bsrcset\s*=", re.IGNORECASE),
),
(
"fetch() of an off-origin URL",
re.compile(rf"\bfetch\s*\(\s*[\"'`]{_ABS}", re.IGNORECASE),
),
(
"import of an off-origin URL",
re.compile(rf"\b(?:import|export)\b[^;\n]*?\bfrom\s*[\"'`]{_ABS}", re.IGNORECASE),
),
(
"bare import of an off-origin URL",
re.compile(rf"\bimport\s*[\"'`]{_ABS}", re.IGNORECASE),
),
(
"WebSocket with an absolute URL",
# Same-origin is built from `location`, never written out (`WebSocket("ws…`).
re.compile(r"\bWebSocket\s*\(\s*[\"'`](?:wss?|https?)://", re.IGNORECASE),
),
(
"EventSource (server-sent events)",
re.compile(r"\bnew\s+EventSource\s*\(|\bEventSource\s*\(", re.IGNORECASE),
),
)
#: Same-origin prefixes an element URL may start with.
STATIC_PREFIX = "/static/"
@dataclass(frozen=True, slots=True)
class Violation:
"""One forbidden reference, in one file, on one line."""
path: Path
line: int
rule: str
text: str
def describe(self) -> str:
"""Human-readable failure line naming the offending file."""
return f"{self.path}:{self.line}: {self.rule}: {self.text.strip()}"
def _mask(text: str) -> str:
"""Blank out the text that looks external but cannot cause a request."""
return _NAMESPACE_DECL.sub(lambda match: " " * len(match.group(0)), text)
def iter_asset_files(root: Path) -> Iterator[Path]:
"""Every scannable asset under ``root``, depth-first and in a stable order."""
if not root.is_dir(): # pragma: no cover - guarded by the caller's own assertion
return
for path in sorted(root.rglob("*")):
if path.is_file() and path.suffix.lower() in ASSET_SUFFIXES:
yield path
def scan_text(path: Path, text: str) -> list[Violation]:
"""Every forbidden reference in ``text``, attributed to ``path``."""
violations: list[Violation] = []
for number, raw in enumerate(_mask(text).splitlines(), start=1):
for rule, pattern in BANNED_PATTERNS:
if pattern.search(raw):
violations.append(Violation(path=path, line=number, rule=rule, text=raw))
if path.suffix.lower() in HOSTED_TOKEN_SUFFIXES:
lowered = raw.lower()
violations.extend(
Violation(path=path, line=number, rule=f"hosted library {token!r}", text=raw)
for token in HOSTED_LIBRARY_TOKENS
if token in lowered
)
if path.suffix.lower() == ".html":
violations.extend(scan_element_urls(path, text))
return violations
def scan_file(path: Path) -> list[Violation]:
"""Read one asset and return its forbidden references."""
return scan_text(path, path.read_text(encoding="utf-8"))
def scan_tree(root: Path) -> list[Violation]:
"""Scan every asset under ``root`` (the shipped client, when called with STATIC_DIR)."""
return [violation for path in iter_asset_files(root) for violation in scan_file(path)]
def _element_attributes(tag: str) -> dict[str, str]:
"""Parse `name="value"` pairs of one tag into a lower-cased mapping."""
found: dict[str, str] = {}
for name, value in re.findall(r"""([\w-]+)\s*=\s*["']([^"']*)["']""", tag):
found[name.lower()] = value
return found
def _is_allowed(value: str, *, data_ok: bool) -> bool:
if value.startswith(STATIC_PREFIX):
return True
return data_ok and value.startswith("data:")
def scan_element_urls(path: Path, text: str) -> list[Violation]:
"""Every `<script src>` / `<link href>` / `<img src>` in an HTML document that this
app does not serve.
`<script src>` and stylesheet `<link>`s must be `/static/...`. `<img>` and
non-stylesheet `<link>`s (the `data:` favicon) may additionally be `data:`, matching
the CSP's `img-src 'self' data:` — in-app bytes either way.
"""
violations: list[Violation] = []
for match in re.finditer(r"<(script|link|img)\b((?:[^>'\"]|\"[^\"]*\"|'[^']*')*)>", text, re.I):
name = match.group(1).lower()
attrs = _element_attributes(match.group(2))
if name == "script":
attribute, data_ok = "src", False
elif name == "link":
attribute, data_ok = "href", "stylesheet" not in attrs.get("rel", "").lower()
else:
attribute, data_ok = "src", True
value = attrs.get(attribute)
if value is None or _is_allowed(value, data_ok=data_ok):
continue
line = text.count("\n", 0, match.start()) + 1
rule = f"<{name} {attribute}> is not a same-origin /static/ path"
violations.append(Violation(path=path, line=line, rule=rule, text=match.group(0)))
return violations
def format_violations(violations: list[Violation]) -> str:
"""Render violations for a pytest failure message."""
return "\n".join(violation.describe() for violation in violations)
@pytest.fixture
def static_tree(tmp_path: Path) -> Path:
"""A throwaway static root, shaped like the real one, for planted violations."""
root = tmp_path / "static"
(root / "js").mkdir(parents=True)
(root / "css").mkdir(parents=True)
(root / "index.html").write_text("<!doctype html><title>t</title>", encoding="utf-8")
return root
# --------------------------------------------------------------------------- the scanned root
def test_the_guard_scans_the_directory_the_server_actually_serves() -> None:
"""Re-deriving the root here is what stops a green run against the wrong folder."""
assert STATIC_DIR == REPO_ROOT / "app" / "static"
assert STATIC_DIR.is_dir(), f"{STATIC_DIR} is not a directory"
assert (STATIC_DIR / "index.html").is_file()
assert resolve_static_dir(Settings()) == STATIC_DIR
def test_the_scan_walks_the_shipped_client_files() -> None:
names = {path.relative_to(STATIC_DIR).as_posix() for path in iter_asset_files(STATIC_DIR)}
assert {"index.html", "js/app.js", "css/base.css"} <= names, names
# --------------------------------------------------------------------------- the shipped client
def test_shipped_client_references_nothing_external() -> None:
"""A12: zero forbidden references anywhere under app/static/."""
violations = scan_tree(STATIC_DIR)
report = format_violations(violations)
assert violations == [], f"the client loads bytes this app does not serve:\n{report}"
def test_shipped_index_html_loads_only_same_origin_assets() -> None:
"""Non-vacuous on purpose: the index must actually declare a script and a stylesheet."""
html = (STATIC_DIR / "index.html").read_text(encoding="utf-8")
tags = re.findall(r"<(script|link|img)\b", html, re.I)
assert len(tags) >= 3, "index.html no longer declares the assets this test guards"
assert 'src="/static/js/app.js"' in html
assert scan_element_urls(STATIC_DIR / "index.html", html) == []
# --------------------------------------------------------------------------- planted violations
@pytest.mark.parametrize(
("name", "content"),
[
("script_src", '<!doctype html><script src="https://cdn.example.com/x.js"></script>'),
(
"module_src",
'<!doctype html><script type="module" src="//cdn.example.com/x.js"></script>',
),
("stylesheet_link", '<link rel="stylesheet" href="https://fonts.example.com/f.css">'),
("protocol_relative_stylesheet", '<link rel="stylesheet" href="//cdn.example.com/a.css">'),
("img_src", '<img src="https://evil.example.com/pixel.png">'),
("img_srcset", '<img src="/static/a.svg" srcset="https://cdn.example.com/b.svg 2x">'),
],
)
def test_a_planted_html_reference_fails_the_scan(
static_tree: Path, name: str, content: str
) -> None:
page = static_tree / f"{name}.html"
page.write_text(content, encoding="utf-8")
violations = scan_tree(static_tree)
assert violations, content
assert all(violation.path == page for violation in violations), format_violations(violations)
assert page.name in format_violations(violations)
@pytest.mark.parametrize(
("suffix", "content", "rule"),
[
(
".js",
'import { x } from "https://cdn.example.com/lib.js";',
"import of an off-origin URL",
),
(
".js",
'import "https://cdn.example.com/side-effect.js";',
"bare import of an off-origin URL",
),
(".js", 'export * from "//cdn.example.com/lib.js";', "import of an off-origin URL"),
(".js", 'fetch("https://api.example.com/state");', "fetch() of an off-origin URL"),
(".js", 'fetch("//api.example.com/state");', "fetch() of an off-origin URL"),
(
".js",
'new WebSocket("wss://remote.example.com/ws/ABCDE");',
"WebSocket with an absolute URL",
),
(".js", 'new EventSource("/events");', "EventSource (server-sent events)"),
(".js", 'const u = "https://cdn.example.com/x.js";', "absolute URL"),
(".css", "@import url(https://fonts.example.com/one.css);", "CSS @import url()"),
(".css", "@import url(//fonts.example.com/one.css);", "CSS @import url()"),
(".css", "body { background: url(//cdn.example.com/bg.png); }", "protocol-relative URL"),
(
".svg",
'<svg xmlns="http://www.w3.org/2000/svg">'
'<image href="https://x.example.com/a.png"/></svg>',
"absolute URL",
),
(".html", '<a href="https://external.example.com">out</a>', "absolute URL"),
],
)
def test_a_planted_asset_reference_fails_the_scan(
static_tree: Path, suffix: str, content: str, rule: str
) -> None:
planted = static_tree / "js" / f"planted{suffix}"
planted.write_text(content, encoding="utf-8")
violations = scan_tree(static_tree)
assert violations, content
assert rule in {violation.rule for violation in violations}, format_violations(violations)
assert any(violation.path == planted for violation in violations)
assert planted.name in format_violations(violations)
@pytest.mark.parametrize("token", HOSTED_LIBRARY_TOKENS)
@pytest.mark.parametrize("suffix", [".js", ".html"])
def test_a_planted_hosted_library_token_fails_the_scan(
static_tree: Path, token: str, suffix: str
) -> None:
planted = static_tree / f"planted{suffix}"
planted.write_text(f"/* vendored from {token} once */", encoding="utf-8")
violations = [v for v in scan_tree(static_tree) if v.path == planted]
assert violations, token
assert token in format_violations(violations)
# --------------------------------------------------------------------------- what must stay green
@pytest.mark.parametrize(
("suffix", "content"),
[
(
".svg",
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 16 16">'
'<rect width="16" height="16"/></svg>',
),
(
".svg",
'<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">'
'<use xlink:href="#tile"/></svg>',
),
(".js", "// a comment about docs/PROTOCOL.md and a/b/c paths\nexport const ok = 1;\n"),
(".js", 'const ws = new WebSocket(`${location.origin}/ws/${code}`);\n'),
(".js", 'fetch("/static/api/levels");\n'),
(".css", ".a { background: url(/static/assets/tile.svg); }\n"),
(".html", '<img src="data:image/svg+xml,<svg/>">\n'),
],
)
def test_same_origin_and_namespace_references_are_allowed(
static_tree: Path, suffix: str, content: str
) -> None:
"""The guard must not train anyone to delete real code: same-origin stays green."""
(static_tree / "js" / f"fine{suffix}").write_text(content, encoding="utf-8")
assert scan_tree(static_tree) == []
def test_non_asset_files_are_out_of_scope(static_tree: Path) -> None:
"""Notes and images are not client instructions; the scan is about loadable code."""
(static_tree / "NOTES.md").write_text("see https://example.com for details", encoding="utf-8")
(static_tree / "logo.png").write_bytes(b"\x89PNG\r\n")
assert scan_tree(static_tree) == []
def test_an_empty_tree_is_not_a_passing_signal(tmp_path: Path) -> None:
"""Documented explicitly: an empty root scans clean, so the root tests above matter."""
assert scan_tree(tmp_path) == []
assert list(iter_asset_files(tmp_path)) == []