undercooked/tests/unit/test_protocol_doc_sync.py

"""Unit: `docs/PROTOCOL.md` and `protocol.py` cannot drift (phase 06, task 01).

The phase's completion criteria demand that "every field in code appears in
the doc" — and, in the other direction, that the doc never names a field
that does not exist. This test is the guard: for **every** message model
in both directions (the nine client→server, the eight server→client, and
the six wire objects they nest) it extracts the field names from the doc's
``### ModelName`` table and compares them with the model's ``model_fields``
— set-equal, in both directions. It also keeps the doc's error-code and
close-code tables equal to the `ErrorCode` / `CloseCode` constants.

Doc convention (what this test relies on): every ``###`` heading in
``docs/PROTOCOL.md`` is exactly one model section (no other ``###``
headings — use ``##`` for prose sections), and each section holds one
markdown table whose first column is the field name. So: an added field
that is not documented fails here, a documented field that is not in the
model fails here, a renamed model fails here, and a new error code without
a doc row fails here.
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest
from pydantic import BaseModel

from undercooked.server.protocol import (
    CloseCode,
    CountdownMsg,
    EmoteMsg,
    ErrMsg,
    ErrorCode,
    EventWire,
    EvMsg,
    HelloMsg,
    InputMsg,
    LobbyMsg,
    NextLevelMsg,
    PingMsg,
    PlayerScoreWire,
    PongMsg,
    ReadyMsg,
    RoundMsg,
    SeatWire,
    SetLevelMsg,
    SnapMsg,
    SpectatorWire,
    StartMsg,
    ToLobbyMsg,
    Vec2Msg,
    WelcomeMsg,
    YouWire,
)

pytestmark = pytest.mark.unit

DOCS = Path(__file__).resolve().parents[2] / "docs" / "PROTOCOL.md"

#: Every message model in both directions, plus the wire objects they nest.
C_TO_S_MODELS: dict[str, type[BaseModel]] = {
    "HelloMsg": HelloMsg,
    "ReadyMsg": ReadyMsg,
    "SetLevelMsg": SetLevelMsg,
    "StartMsg": StartMsg,
    "InputMsg": InputMsg,
    "EmoteMsg": EmoteMsg,
    "ToLobbyMsg": ToLobbyMsg,
    "NextLevelMsg": NextLevelMsg,
    "PingMsg": PingMsg,
}
S_TO_C_MODELS: dict[str, type[BaseModel]] = {
    "WelcomeMsg": WelcomeMsg,
    "LobbyMsg": LobbyMsg,
    "CountdownMsg": CountdownMsg,
    "SnapMsg": SnapMsg,
    "RoundMsg": RoundMsg,
    "EvMsg": EvMsg,
    "ErrMsg": ErrMsg,
    "PongMsg": PongMsg,
}
WIRE_MODELS: dict[str, type[BaseModel]] = {
    "Vec2Msg": Vec2Msg,
    "YouWire": YouWire,
    "SeatWire": SeatWire,
    "SpectatorWire": SpectatorWire,
    "EventWire": EventWire,
    "PlayerScoreWire": PlayerScoreWire,
}
ALL_MODELS: dict[str, type[BaseModel]] = {
    **C_TO_S_MODELS,
    **S_TO_C_MODELS,
    **WIRE_MODELS,
}

SECTION_HEADING = re.compile(r"^###\s+(\S+)\s*$")


def _clean(cell: str) -> str:
    return cell.strip().strip("`").strip()


def extract_model_sections(text: str) -> dict[str, list[str]]:
    """Field names (first table column) per ``###`` section.

    A section runs from its ``###`` heading to the next heading of any
    level; inside it, only table rows count, and the header row (first cell
    ``Field``) and the separator row (dashes/colons) are skipped.
    """
    sections: dict[str, list[str]] = {}
    current: str | None = None
    for line in text.splitlines():
        heading = SECTION_HEADING.match(line)
        if heading is not None:
            name = heading.group(1)
            sections.setdefault(name, [])
            current = name
            continue
        if line.lstrip().startswith("#"):
            current = None  # any other heading ends the section
            continue
        if current is None:
            continue
        stripped = line.strip()
        if not stripped.startswith("|"):
            continue
        cells = [_clean(cell) for cell in stripped.strip("|").split("|")]
        if not cells or cells[0] in {"Field", ""}:
            continue
        if set(cells[0]) <= {"-", ":"}:
            continue
        sections[current].append(cells[0])
    return sections


def extract_section_first_cells(text: str, heading: str) -> list[str]:
    """First table column of one ``##`` section (the code tables)."""
    lines = text.splitlines()
    try:
        start = next(i for i, line in enumerate(lines) if line.strip() == heading)
    except StopIteration:
        pytest.fail(f"{heading!r} section not found in docs/PROTOCOL.md")
    cells: list[str] = []
    for line in lines[start + 1 :]:
        if line.lstrip().startswith("#"):
            break
        stripped = line.strip()
        if not stripped.startswith("|"):
            continue
        first = _clean(stripped.strip("|").split("|")[0])
        if first in {"Code", ""} or set(first) <= {"-", ":"}:
            continue
        cells.append(first)
    return cells


def test_the_doc_exists() -> None:
    assert DOCS.is_file(), "docs/PROTOCOL.md is the phase deliverable — it must exist"


def test_documented_sections_are_exactly_the_model_registry() -> None:
    sections = extract_model_sections(DOCS.read_text(encoding="utf-8"))
    missing = set(ALL_MODELS) - set(sections)
    extra = set(sections) - set(ALL_MODELS)
    assert not missing, f"models with no ### section in the doc: {sorted(missing)}"
    assert not extra, (
        f"### sections that are not model sections (doc convention): {sorted(extra)}"
    )


@pytest.mark.parametrize("name", sorted(ALL_MODELS))
def test_the_doc_fields_match_the_model_fields_in_both_directions(name: str) -> None:
    model = ALL_MODELS[name]
    sections = extract_model_sections(DOCS.read_text(encoding="utf-8"))
    assert name in sections, f"no ### {name} section in docs/PROTOCOL.md"
    documented = set(sections[name])
    implemented = set(model.model_fields)
    undocumented = implemented - documented
    ghost = documented - implemented
    assert not undocumented, (
        f"{name}: fields in the model but missing from the doc: {sorted(undocumented)}"
    )
    assert not ghost, (
        f"{name}: fields documented but not in the model: {sorted(ghost)}"
    )


def test_both_directions_are_covered_by_the_registry() -> None:
    assert len(C_TO_S_MODELS) == 9
    assert len(S_TO_C_MODELS) == 8
    assert len(WIRE_MODELS) == 6
    assert set(ALL_MODELS) == set(C_TO_S_MODELS) | set(S_TO_C_MODELS) | set(WIRE_MODELS)


def test_the_error_code_table_matches_the_constants() -> None:
    cells = extract_section_first_cells(DOCS.read_text(encoding="utf-8"), "## Error codes")
    assert len(cells) == len(set(cells)), "duplicate rows in the error-code table"
    documented = set(cells)
    implemented = {
        ErrorCode.UNKNOWN_TYPE,
        ErrorCode.BAD_JSON,
        ErrorCode.BAD_FIELD,
        ErrorCode.TOO_BIG,
        ErrorCode.RATE_LIMITED,
        ErrorCode.NOT_SEATED,
        ErrorCode.NOT_HOST,
        ErrorCode.LEVEL_LOCKED,
        ErrorCode.WRONG_STATE,
        ErrorCode.ROOM_FULL,
        ErrorCode.ROOM_GONE,
        ErrorCode.PROTOCOL_MISMATCH,
        ErrorCode.EVICTED,
        ErrorCode.INTERNAL,
    }
    assert documented == implemented, (
        f"error-code table drift: missing {sorted(implemented - documented)}, "
        f"ghosts {sorted(documented - implemented)}"
    )


def test_the_close_code_table_matches_the_constants() -> None:
    cells = extract_section_first_cells(DOCS.read_text(encoding="utf-8"), "## Close codes")
    documented = {int(cell) for cell in cells}
    implemented = {
        CloseCode.NORMAL,
        CloseCode.PROTOCOL_MISMATCH,
        CloseCode.ROOM_GONE,
        CloseCode.ROOM_FULL,
        CloseCode.EVICTED,
        CloseCode.ABUSE,
    }
    assert documented == implemented, (
        f"close-code table drift: missing {sorted(implemented - documented)}, "
        f"ghosts {sorted(documented - implemented)}"
    )


def test_the_whole_contract_round_trips_through_the_models() -> None:
    """Every model here is a real Pydantic model with a `type` discriminator
    (C→S) or a defaulted one (S→C) — a smoke over the registry itself."""
    for name, model in ALL_MODELS.items():
        assert isinstance(model, type)
        assert issubclass(model, BaseModel)
        fields = set(model.model_fields)
        if name in C_TO_S_MODELS:
            assert "type" in fields, f"{name} has no discriminator"