settled-reach/tooling/test_check_parity.py

#!/usr/bin/env python3
"""Parity between `reach check client-version` and the script it replaces (T-1262).

Both implementations are live: the old `tooling/check-client-version` is still
wired to the pre-push hook, and the deprecation window is deliberate (T-1253
retires it). While both exist they must agree, or the migration silently changes
what the gate does.

WHAT PARITY MEANS HERE, because the ticket asked for byte-for-byte and that is
no longer the right requirement. D-263 was amended after this ticket was written
to give `reach` a streaming output model: the event stream — progress and the
final verdict — goes to **stderr** as JSONL, and **stdout** is reserved for a
command's actual data so `reach ... | jq` keeps working. The old script writes
its OK line to stdout. So the two cannot be byte-identical on the same stream
without either abandoning the streaming model or special-casing this one
command, and neither is worth it.

What is enforced instead, and is strictly stronger where it counts:

  1. EXIT CODES MATCH EXACTLY, on every path. This is what the hook gates on,
     and it is the only part a caller can act on programmatically.
  2. EVERY FACT the old output carries appears in the new output — both version
     numbers on drift, the failing path when a file is missing. A migration that
     drops a detail from a failure message makes the failure harder to fix.
  3. THE NEW OUTPUT NAMES A REMEDY on failure, which D-263 requires and the old
     script only partly does.

In text mode the OK line is in fact byte-identical; only the stream differs.

Run: python3 tooling/test_check_parity.py
"""

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

REPO_ROOT = Path(__file__).resolve().parent.parent
OLD_SCRIPT_NAME = "check-client-version"

# (label, yaml version, godot version or None to omit the file entirely)
CASES = [
    ("ok", "0.9.1", "0.9.1"),
    ("drift", "0.9.1", "0.8.0"),
    ("missing-godot-file", "0.9.1", None),
]


def _build_fixture(tmp: Path, yaml_version: str, godot_version: str | None) -> Path:
    """A miniature repo root: sentinel, client file, and a copy of the old script.

    The old script resolves its root as `Path(__file__).parent.parent`, so
    copying it into `<fixture>/tooling/` is what re-points it — there is no
    override to pass it. The new command takes SR_REPO_ROOT. That asymmetry is
    itself part of why the port is worth doing.
    """
    root = tmp / "fixture"
    (root / "tooling").mkdir(parents=True)
    (root / "client").mkdir()

    # project.yaml carries commentary quoting OTHER version numbers, mirroring
    # the real file — a regex not anchored to line start would match those.
    (root / "project.yaml").write_text(
        "# history: 0.1.0, 0.2.0, 0.3.0 were pre-cascade\n"
        f"version: {yaml_version}\n"
        "name: fixture\n",
        encoding="utf-8",
    )
    if godot_version is not None:
        (root / "client" / "project.godot").write_text(
            "[application]\n\n"
            'config/name="Fixture"\n'
            f'config/version="{godot_version}"\n',
            encoding="utf-8",
        )

    shutil.copy2(REPO_ROOT / "tooling" / OLD_SCRIPT_NAME, root / "tooling" / OLD_SCRIPT_NAME)
    return root


def _run_old(root: Path) -> tuple[int, str]:
    result = subprocess.run(
        [sys.executable, str(root / "tooling" / OLD_SCRIPT_NAME)],
        capture_output=True,
        text=True,
    )
    return result.returncode, result.stdout + result.stderr


def _run_new(root: Path) -> tuple[int, str, str]:
    """Invoke the shipped command by BARE NAME — never a path (T-1261)."""
    env = {**os.environ, "SR_REPO_ROOT": str(root), "SR_OUTPUT_FORMAT": "text"}
    result = subprocess.run(
        ["reach", "check", "client-version"],
        capture_output=True,
        text=True,
        env=env,
    )
    json_env = {**os.environ, "SR_REPO_ROOT": str(root), "SR_OUTPUT_FORMAT": "json"}
    as_json = subprocess.run(
        ["reach", "check", "client-version"],
        capture_output=True,
        text=True,
        env=json_env,
    )
    return result.returncode, result.stdout + result.stderr, as_json.stderr


FACT_CASES = [
    # (label, catalog fact_ids, referenced fact_ids, expect non-zero)
    ("fact-ok", ["alpha", "beta"], ["alpha"], False),
    ("fact-unknown", ["alpha"], ["alpha", "ghost"], True),
    ("fact-advisory", [], ["alpha"], False),  # no catalogs: advisory, exit 0
]


def _build_fact_fixture(tmp: Path, catalog: list[str], referenced: list[str]) -> Path:
    """A miniature repo with knowledge catalogs and campaign content."""
    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")
    (root / "tooling").mkdir()

    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",
        )
    # Excluded by name in both implementations — if either starts counting it,
    # the canonical totals diverge and this case catches it.
    (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"
    )

    shutil.copy2(REPO_ROOT / "tooling" / "check-fact-ids", root / "tooling" / "check-fact-ids")
    (root / "tooling" / "check-fact-ids").chmod(0o755)
    return root


def check_fact_ids(failures: list[str]) -> None:
    """Parity for the first bash-to-Python rewrite (D-263).

    Worth more scrutiny than a Python move: nothing here was relocated, it was
    reimplemented, so the two could agree on the happy path and diverge on
    exactly the inputs the grep chain handled by accident — quoted values,
    trailing comments, the excluded catalog.
    """
    for label, catalog, referenced, expect_failure in FACT_CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _build_fact_fixture(Path(tmpdir), catalog, referenced)

            old = subprocess.run(
                [str(root / "tooling" / "check-fact-ids")],
                capture_output=True,
                text=True,
            )
            env = {**os.environ, "SR_REPO_ROOT": str(root), "SR_OUTPUT_FORMAT": "text"}
            new = subprocess.run(
                ["reach", "check", "fact-ids"],
                capture_output=True,
                text=True,
                env=env,
            )

            if (old.returncode != 0) != expect_failure:
                failures.append(
                    f"[{label}] the OLD script exited {old.returncode}; the fixture "
                    "does not set up the case it claims to"
                )
            if old.returncode != new.returncode:
                failures.append(
                    f"[{label}] exit differs: old={old.returncode} new={new.returncode}"
                )

            combined = new.stdout + new.stderr
            if expect_failure and "ghost" not in combined:
                failures.append(
                    f"[{label}] the new output does not name the unknown fact_id, so a "
                    "failure says something is wrong without saying what"
                )
            if "attribute_not_a_fact" in combined:
                failures.append(
                    f"[{label}] entity-attributes.yaml was counted as a catalog — its "
                    "schema is attribute keys, and including it inflates the canonical set"
                )


# Ranges from the repo's own history, chosen so both implementations see
# identical input with nothing mutated. A fixture repo would not work here: the
# canvas registry globs from a __file__-derived root, so under SR_REPO_ROOT the
# new service would diff the fixture while globbing the real tree. Real history
# sidesteps that asymmetry entirely — and the failure cases below are genuine
# instances of the regression this gate exists to catch, not constructed ones.
#
# SHAs are hardcoded because history is immutable. If a rebase ever invalidates
# one, the test says so by name rather than silently checking nothing.
CANVAS_CASES = [
    # (label, base, head, expect non-zero)
    ("canvas-needs-bump", "4e503c356~1", "4e503c356", True),  # T-1237: bumped later
    ("canvas-needs-bump-2", "566b56651~1", "566b56651", True),  # T-1194: same
    ("canvas-bumped", "9b146f9e1~1", "9b146f9e1", False),  # changed AND bumped
    ("canvas-clean", "6b31111cd~1", "6b31111cd", False),  # governance only
    ("canvas-no-base", "no-such-ref-anywhere", "HEAD", False),  # skip, not fail
]


def check_canvas_version(failures: list[str]) -> None:
    """Parity for the gate with the worst track record — five shipped regressions.

    Its failure path is the one that matters, so two of the five cases are real
    commits that changed canvas generation without bumping the version. Both
    were bumped after the fact, which is precisely the history that produced
    this check.
    """
    for label, base, head, expect_failure in CANVAS_CASES:
        old = subprocess.run(
            [
                sys.executable,
                str(REPO_ROOT / "tooling" / "check-canvas-version"),
                "--base",
                base,
                "--head",
                head,
            ],
            capture_output=True,
            text=True,
            cwd=REPO_ROOT,
        )
        new = subprocess.run(
            ["reach", "check", "canvas-version", "--base", base, "--head", head],
            capture_output=True,
            text=True,
            cwd=REPO_ROOT,
            env={**os.environ, "SR_OUTPUT_FORMAT": "text"},
        )

        if (old.returncode != 0) != expect_failure:
            failures.append(
                f"[{label}] the OLD script exited {old.returncode}; this range no "
                "longer sets up the case it claims to — history may have been rewritten"
            )
        if old.returncode != new.returncode:
            failures.append(
                f"[{label}] exit differs: old={old.returncode} new={new.returncode}"
            )

        if expect_failure:
            # The changed files are the actionable half of this failure; a
            # message that says "something changed" without saying what leaves
            # the reader to re-derive the intersection by hand.
            for stream in (old.stdout + old.stderr,):
                for line in stream.splitlines():
                    token = line.strip()
                    if token.endswith((".rs", ".gd")) and token not in (
                        new.stdout + new.stderr
                    ):
                        failures.append(
                            f"[{label}] new output omits a changed file the old one "
                            f"named: {token}"
                        )


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

    failures: list[str] = []

    for label, yaml_version, godot_version in CASES:
        with tempfile.TemporaryDirectory() as tmpdir:
            root = _build_fixture(Path(tmpdir), yaml_version, godot_version)
            old_code, old_text = _run_old(root)
            new_code, new_text, new_json = _run_new(root)

            # 1. Exit codes must match exactly — the hook gates on this.
            if old_code != new_code:
                failures.append(
                    f"[{label}] exit code differs: old={old_code} new={new_code}"
                )

            # A case that cannot fail proves nothing: drift and missing-file must
            # actually be non-zero, or "they matched" would be vacuous.
            if label != "ok" and new_code == 0:
                failures.append(
                    f"[{label}] expected a NON-ZERO exit — a gate that only ever "
                    "passes has never been tested"
                )
            if label == "ok" and new_code != 0:
                failures.append(f"[{label}] expected exit 0, got {new_code}")

            # 2. Every fact the old message carried must survive the port.
            #
            # Derived from what the old output ACTUALLY contains, not from a
            # hardcoded list — the first version of this asserted the yaml
            # version on every failing path, which the old script does not
            # report when the client file is missing. The rule is "nothing the
            # old message said is lost", so the old message has to define it.
            candidates = {yaml_version, godot_version, str(root / "client" / "project.godot")}
            expected_facts = [f for f in candidates if f and f in old_text]
            for fact in expected_facts:
                if fact not in new_text:
                    failures.append(
                        f"[{label}] new output drops a fact the old one carried: {fact!r}\n"
                        f"    old: {old_text.strip()!r}\n"
                        f"    new: {new_text.strip()!r}"
                    )

            # 3. Failures must name a remedy (D-263), structurally not just in prose.
            if label != "ok":
                verdicts = [
                    json.loads(line)
                    for line in new_json.splitlines()
                    if line.strip().startswith("{")
                ]
                verdict = next((v for v in verdicts if v.get("kind") == "verdict"), None)
                if verdict is None:
                    failures.append(f"[{label}] no verdict event on the stream")
                elif not verdict.get("fix"):
                    failures.append(
                        f"[{label}] verdict carries no `fix` — D-263 requires a failure "
                        "to name the command that resolves it"
                    )

            # The OK line is byte-identical in text mode; only the stream differs.
            if label == "ok" and old_text.strip() != new_text.strip():
                failures.append(
                    f"[ok] text differs\n    old: {old_text.strip()!r}\n"
                    f"    new: {new_text.strip()!r}"
                )

    check_fact_ids(failures)
    check_canvas_version(failures)

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

    print(
        f"test_check_parity: OK — client-version {len(CASES)} paths, "
        f"fact-ids {len(FACT_CASES)} paths, canvas-version {len(CANVAS_CASES)} paths, "
        "all agreeing on exit code and content"
    )
    return 0


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