#!/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())