settled-reach/tooling/test_check.py

#!/usr/bin/env python3
"""Behaviour of the five `reach check` gates (T-1281).

**This file used to be test_check_parity.py.** Every case here began as a parity
case, running the new implementation beside the shell or Python script it
replaced and requiring identical exit codes and no lost facts. That evidence is
recorded in T-1262 and T-1281; the scripts themselves are now retired, so there
is nothing left to compare against and these assertions become the spec.

A parity test is scaffolding with a defined lifetime. Keeping one after its
subject is deleted would mean keeping the subject alive to be compared with,
which is the opposite of a migration.

What did NOT change: every case still exercises a failure, not just a pass. A
gate that has only ever been run against a healthy tree has never been tested,
and four of these five gates exist because something shipped broken.

Run: python3 tooling/test_check.py
"""

import json
import os
import shutil
import sqlite3
import subprocess
import sys
import tempfile
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent


def _reach(*args: str, root: Path | None = None) -> subprocess.CompletedProcess[str]:
    """Invoke by BARE NAME — never a path or interpreter (T-1261)."""
    env = {**os.environ, "SR_OUTPUT_FORMAT": "text"}
    if root is not None:
        env["SR_REPO_ROOT"] = str(root)
    return subprocess.run(
        ["reach", "check", *args],
        capture_output=True,
        text=True,
        cwd=REPO_ROOT,
        env=env,
    )


def _out(result: subprocess.CompletedProcess[str]) -> str:
    return result.stdout + result.stderr


# --- client-version -------------------------------------------------------

VERSION_CASES = [
    ("ok", "0.9.1", "0.9.1", 0),
    ("drift", "0.9.1", "0.8.0", 1),
    ("missing-godot-file", "0.9.1", None, 1),
]


def _version_fixture(tmp: Path, yaml_version: str, godot_version: str | None) -> Path:
    root = tmp / "fixture"
    (root / "client").mkdir(parents=True)
    # Commentary quoting OTHER version numbers, mirroring the real file: a regex
    # not anchored to line start would match these and read a comment edit as a
    # version bump.
    (root / "project.yaml").write_text(
        "# history: 0.1.0, 0.2.0, 0.3.0 were pre-cascade\n"
        f"version: {yaml_version}\nname: fixture\n",
        encoding="utf-8",
    )
    if godot_version is not None:
        (root / "client" / "project.godot").write_text(
            f'[application]\n\nconfig/name="Fixture"\nconfig/version="{godot_version}"\n',
            encoding="utf-8",
        )
    return root


def check_client_version(failures: list[str]) -> None:
    for label, yaml_version, godot_version, expected in VERSION_CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _version_fixture(Path(tmpdir), yaml_version, godot_version)
            result = _reach("client-version", root=root)
            if result.returncode != expected:
                failures.append(
                    f"[client-{label}] exited {result.returncode}, expected {expected}"
                )
            if label == "drift":
                for fact in (yaml_version, godot_version):
                    if fact not in _out(result):
                        failures.append(
                            f"[client-{label}] output omits {fact!r} — a drift report "
                            "that does not name both versions cannot be acted on"
                        )


# --- fact-ids -------------------------------------------------------------

FACT_CASES = [
    ("ok", ["alpha", "beta"], ["alpha"], 0),
    ("unknown", ["alpha"], ["alpha", "ghost"], 1),
    ("advisory", [], ["alpha"], 0),  # catalogs unpopulated: advisory, exit 0
]


def _fact_fixture(tmp: Path, catalog: list[str], referenced: list[str]) -> Path:
    root = tmp / "fixture"
    knowledge = root / "server" / "content" / "global" / "knowledge"
    campaigns = root / "server" / "content" / "campaigns"
    knowledge.mkdir(parents=True)
    campaigns.mkdir(parents=True)
    (root / "project.yaml").write_text("version: 0.0.0\n", encoding="utf-8")

    if catalog:
        (knowledge / "facts.yaml").write_text(
            "# a comment mentioning fact_id: decoy_should_not_count\n"
            + "".join(f"  - fact_id: {name}\n" for name in catalog),
            encoding="utf-8",
        )
    # Its schema is attribute keys, not fact_ids. If this ever starts counting,
    # the canonical set inflates and the advisory case flips to enforcing.
    (knowledge / "entity-attributes.yaml").write_text(
        "  - fact_id: attribute_not_a_fact\n", encoding="utf-8"
    )
    (campaigns / "story.yaml").write_text(
        "".join(f'  - fact_id: "{name}"\n' for name in referenced), encoding="utf-8"
    )
    return root


def check_fact_ids(failures: list[str]) -> None:
    for label, catalog, referenced, expected in FACT_CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _fact_fixture(Path(tmpdir), catalog, referenced)
            result = _reach("fact-ids", root=root)
            combined = _out(result)
            if result.returncode != expected:
                failures.append(
                    f"[fact-{label}] exited {result.returncode}, expected {expected}"
                )
            if label == "unknown" and "ghost" not in combined:
                failures.append(
                    "[fact-unknown] output does not name the unknown fact_id, so the "
                    "failure says something is wrong without saying what"
                )
            if "attribute_not_a_fact" in combined:
                failures.append(
                    f"[fact-{label}] entity-attributes.yaml was counted as a catalog"
                )


# --- canvas-version -------------------------------------------------------
#
# Ranges from this repo's own history. Two of them are GENUINE instances of the
# regression this gate exists to catch — canvas generation changed, the version
# was bumped only after the fact. The history that produced the check, used as
# its own fixture, so nothing can drift from the thing it models.

CANVAS_CASES = [
    ("needs-bump", "4e503c356~1", "4e503c356", 1),  # T-1237 river courses
    ("needs-bump-2", "566b56651~1", "566b56651", 1),  # T-1194 relief texture
    ("bumped", "9b146f9e1~1", "9b146f9e1", 0),
    ("clean", "6b31111cd~1", "6b31111cd", 0),  # governance only
    ("no-base", "no-such-ref-anywhere", "HEAD", 0),  # a skip, not a failure
]

CANVAS_EXPECTED_FILES = {
    "needs-bump": ["server/src/atlas/step_canvas.rs", "server/src/atlas/river_course.rs"],
}


def check_canvas_version(failures: list[str]) -> None:
    for label, base, head, expected in CANVAS_CASES:
        result = _reach("canvas-version", "--base", base, "--head", head)
        if result.returncode != expected:
            failures.append(
                f"[canvas-{label}] exited {result.returncode}, expected {expected}"
                + (
                    " — if history was rewritten this range no longer sets up the case"
                    if expected == 1
                    else ""
                )
            )
        for path in CANVAS_EXPECTED_FILES.get(label, []):
            if path not in _out(result):
                failures.append(
                    f"[canvas-{label}] output omits {path} — the changed-file list is "
                    "the actionable half of this failure"
                )


# --- dataflow-graph -------------------------------------------------------

DIAGRAM_CASES = [
    ("ok", '"server/data/systems.db\\nthe snapshot"', True, 0),
    ("stale", '"server/data/gone-away.db\\nmoved"', False, 1),
    # No slash means no path token: the checker finds nothing to check and must
    # FAIL, because a gate that silently stops checking reports success forever.
    ("no-tokens", '"just prose, no paths"', False, 1),
]


def _diagram_fixture(tmp: Path, label: str, create: bool) -> Path:
    root = tmp / "fixture"
    diagram_dir = root / "docs" / "diagrams" / "data-flow"
    diagram_dir.mkdir(parents=True)
    (root / "project.yaml").write_text("version: 0.0.0\n", encoding="utf-8")
    # A token only counts if its first segment is a real top-level entry, so
    # server/ must exist for the token to even be considered.
    (root / "server" / "data").mkdir(parents=True)
    if create:
        (root / "server" / "data" / "systems.db").write_text("", encoding="utf-8")
    (diagram_dir / "wiki-generator-flow.d2").write_text(f"a: {label}\n", encoding="utf-8")
    return root


def check_dataflow_graph(failures: list[str]) -> None:
    for label, text, create, expected in DIAGRAM_CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _diagram_fixture(Path(tmpdir), text, create)
            result = _reach("dataflow-graph", root=root)
            if result.returncode != expected:
                failures.append(
                    f"[diagram-{label}] exited {result.returncode}, expected {expected}"
                )


# --- systems-db-stamp -----------------------------------------------------
#
# The old script could not be tested in a fixture at all: generator_sources
# raises at IMPORT time when the economy-db tree is absent, so every case died
# before reaching any logic. The ported service imports it lazily and after the
# absent/unstamped checks, which is why these states are reachable here.
# STALE still is not — it needs the whole registered source set — but it is the
# state the live repo exercises on every push.

STAMP_CASES = [
    ("absent", "no-db", 0),
    ("unstamped", None, 2),  # exit 2 is unique to this state (T-857)
    ("empty", [], 2),
    ("unknown", [("not_a_registered_generator", "1.0.0", "abc")], 1),
    ("bad-version", [("not_a_registered_generator", "deadbeef", "abc")], 1),
]


def _stamp_fixture(tmp: Path, rows) -> Path:
    root = tmp / "fixture"
    (root / "server" / "data").mkdir(parents=True)
    (root / "project.yaml").write_text("version: 0.0.0\n", encoding="utf-8")
    if rows == "no-db":
        return root

    connection = sqlite3.connect(str(root / "server" / "data" / "systems.db"))
    if rows is None:
        connection.execute("CREATE TABLE unrelated (x INTEGER)")  # no meta table
    else:
        connection.execute(
            "CREATE TABLE meta (generator_name TEXT, schema_version TEXT, "
            "generator_sha TEXT)"
        )
        connection.executemany("INSERT INTO meta VALUES (?, ?, ?)", rows)
    connection.commit()
    connection.close()
    return root


def check_systems_db_stamp(failures: list[str]) -> None:
    for label, rows, expected in STAMP_CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _stamp_fixture(Path(tmpdir), rows)
            result = _reach("systems-db-stamp", root=root)
            if result.returncode != expected:
                note = (
                    " — exit 2 is the unstamped signal the pre-push hook has relied "
                    "on since T-857"
                    if expected == 2
                    else ""
                )
                failures.append(
                    f"[stamp-{label}] exited {result.returncode}, expected "
                    f"{expected}{note}"
                )


# --- every failure names a remedy ----------------------------------------


def check_failures_carry_remedies(failures: list[str]) -> None:
    """A failing gate must say what to do next, structurally.

    The conformance suite proves every ReachError is RAISED with fix=; this
    proves the remedy actually reaches the caller, which is a different claim
    and the one that matters at 3am.
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        root = _version_fixture(Path(tmpdir), "0.9.1", "0.8.0")
        result = subprocess.run(
            ["reach", "check", "client-version"],
            capture_output=True,
            text=True,
            cwd=REPO_ROOT,
            env={**os.environ, "SR_REPO_ROOT": str(root), "SR_OUTPUT_FORMAT": "json"},
        )
        verdicts = [
            json.loads(line)
            for line in result.stderr.splitlines()
            if line.strip().startswith("{")
        ]
        verdict = next((v for v in verdicts if v.get("kind") == "verdict"), None)
        if verdict is None:
            failures.append("[remedy] a failing gate emitted no verdict event")
        elif not verdict.get("fix"):
            failures.append(
                "[remedy] a failing gate's verdict carries no fix — the D-263 "
                "contract requires every non-zero exit to name the next command"
            )


def main() -> int:
    if shutil.which("reach") is None:
        print(
            "test_check: `reach` is not on PATH.\n  Fix: make install-reach",
            file=sys.stderr,
        )
        return 1

    failures: list[str] = []
    check_client_version(failures)
    check_fact_ids(failures)
    check_canvas_version(failures)
    check_dataflow_graph(failures)
    check_systems_db_stamp(failures)
    check_failures_carry_remedies(failures)

    if failures:
        print("test_check: FAIL", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1

    total = (
        len(VERSION_CASES)
        + len(FACT_CASES)
        + len(CANVAS_CASES)
        + len(DIAGRAM_CASES)
        + len(STAMP_CASES)
    )
    print(f"test_check: OK — {total} cases across five gates, failures included")
    return 0


if __name__ == "__main__":
    sys.exit(main())