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"
                )


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)

    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, all agreeing on exit code and content"
    )
    return 0


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