settled-reach/tooling/test_conformance.py

#!/usr/bin/env python3
"""The D-263 invariants, as a test rather than a style guide (T-1270).

D-263 lists the rules that make the layering real and then says plainly that a
contract nothing checks is a style guide. This is the check.

It matters most *later*. With one domain, every rule here is obvious and nobody
would break one. Once T-1250 lands ~120 commands, nobody re-reads a decision
record before adding a verb — this is what tells them, at the moment it is cheap
to fix rather than after the pattern has been copied forty times.

Invariants:

  1. Transport stays out of the logic layers. No typer/click import in any
     service.py, schemas.py or helpers.py — ever. Transport lives in main.py,
     router.py, and core/cli.py (the single designated transport module).
  2. Nothing prints but console. No bare print()/sys.stdout.write outside
     core/console.py — two output paths drift, and the second is always the one
     that ends up unformatted on stdout inside a hook.
  3. Every registered command carries @command, so no command can have logging
     without error handling or vice versa.
  4. Every command has help at its own level.
  5. Every ReachError names a remedy — a `fix=` on every raise site. The hardest
     to enforce and the most valuable: an error that only says "no" is the thing
     D-263 exists to replace.

The import-graph invariant (nothing heavy reachable from main.py) lives in
test_lazy_domains.py rather than being duplicated here.

Run: python3 tooling/test_conformance.py
"""

import ast
import shutil
import subprocess
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
PACKAGE = REPO_ROOT / "tooling"

# Files permitted to import the CLI framework. See core/cli.py for why it is on
# the list: a shared group class is transport by definition, and the alternative
# is a copy of it in every router that drifts.
TRANSPORT_FILES = {"main.py", "router.py", "cli.py"}
LOGIC_FILES = {"service.py", "schemas.py", "helpers.py", "dependencies.py"}
FORBIDDEN_IMPORTS = {"typer", "click"}


# The invariants govern the PACKAGE, not the legacy tree. The ~123 loose scripts
# under tooling/ predate all of this and use bare print() throughout; holding
# them to a contract they were never written against would mean 500 failures on
# day one and a suite nobody runs.
#
# This is not a permanent carve-out. As T-1250 moves each script into
# domains/<name>/, it lands inside this scope and the invariants start applying
# automatically — so the test's reach grows with the migration rather than
# needing to be widened by hand.
PACKAGE_ROOTS = ("main.py", "__init__.py", "core", "domains")


def _package_files() -> list[Path]:
    """Every .py that is part of the reach package — not the legacy scripts."""
    files: list[Path] = []
    for entry in PACKAGE_ROOTS:
        target = PACKAGE / entry
        if target.is_dir():
            files.extend(target.rglob("*.py"))
        elif target.is_file():
            files.append(target)
    return sorted(path for path in files if not path.name.startswith("test_"))


def _imports(tree: ast.AST) -> set[str]:
    found: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            found.update(alias.name.split(".")[0] for alias in node.names)
        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
            found.add(node.module.split(".")[0])
    return found


def check_transport_isolation(failures: list[str]) -> None:
    """(1) No typer/click outside the designated transport files."""
    for path in _package_files():
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        offending = _imports(tree) & FORBIDDEN_IMPORTS
        if not offending:
            continue
        if path.name in TRANSPORT_FILES:
            continue
        rel = path.relative_to(REPO_ROOT)
        failures.append(
            f"[transport] {rel} imports {', '.join(sorted(offending))} — "
            f"only {', '.join(sorted(TRANSPORT_FILES))} may. "
            "A service must not know it was called from a CLI."
        )
        if path.name in LOGIC_FILES:
            failures.append(
                f"[transport] {rel} is a LOGIC file — this is the invariant that "
                "makes services callable from tests and from each other"
            )


def check_single_output_path(failures: list[str]) -> None:
    """(2) Nothing prints but console."""
    for path in _package_files():
        if path.name == "console.py":
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if isinstance(func, ast.Name) and func.id == "print":
                failures.append(
                    f"[output] {path.relative_to(REPO_ROOT)}:{node.lineno} calls print() — "
                    "core/console.py is the single output path (D-263)"
                )
            elif (
                isinstance(func, ast.Attribute)
                and func.attr == "write"
                and isinstance(func.value, ast.Attribute)
                and func.value.attr in {"stdout", "stderr"}
            ):
                failures.append(
                    f"[output] {path.relative_to(REPO_ROOT)}:{node.lineno} writes to "
                    "sys.stdout/stderr directly — go through core/console.py"
                )


def check_commands_decorated(failures: list[str]) -> None:
    """(3) and (4): every registered command carries @command and has help."""
    probe = """
import json, sys
from tooling.core.command import MARKER
from tooling.main import DOMAINS, _load_domain

report = []
for name in sorted(DOMAINS):
    group = _load_domain(name)
    ctx = None
    for verb in group.list_commands(ctx):
        cmd = group.get_command(ctx, verb)
        callback = getattr(cmd, "callback", None)
        report.append({
            "domain": name,
            "verb": verb,
            "decorated": bool(getattr(callback, MARKER, False)),
            "help": (cmd.help or cmd.short_help or "").strip(),
        })
print(json.dumps(report))
"""
    result = subprocess.run(
        [sys.executable, "-c", probe], capture_output=True, text=True, cwd=REPO_ROOT
    )
    if result.returncode != 0:
        failures.append(f"[commands] could not introspect the CLI:\n{result.stderr}")
        return

    import json

    report = json.loads(result.stdout)
    if not report:
        failures.append(
            "[commands] no commands found — every assertion here would pass vacuously"
        )
    for entry in report:
        where = f"{entry['domain']} {entry['verb']}"
        if not entry["decorated"]:
            failures.append(
                f"[commands] `reach {where}` is missing @command — it would run "
                "without the error contract or the invocation record"
            )
        if not entry["help"]:
            failures.append(f"[commands] `reach {where}` has no help text")


def check_errors_name_a_remedy(failures: list[str]) -> None:
    """(5) Every ReachError raise site passes fix=."""
    for path in _package_files():
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Raise) or not isinstance(node.exc, ast.Call):
                continue
            func = node.exc.func
            name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
            if name != "ReachError":
                continue
            if not any(kw.arg == "fix" for kw in node.exc.keywords):
                failures.append(
                    f"[remedy] {path.relative_to(REPO_ROOT)}:{node.lineno} raises "
                    "ReachError without fix= — an error that only says 'no' is "
                    "what D-263 exists to replace"
                )


def check_exit_codes(failures: list[str]) -> None:
    """(6) Discovery succeeds; being wrong fails.

    The distinction is easy to break in either direction, and both directions
    are bad in ways nothing else would catch. Make discovery a usage error and
    an agent reads its own onboarding as a failure. Make a wrong name succeed
    and a typo in a hook passes silently — the exit-0 trap D-263 opens with.
    """
    cases = [
        ([], 0, "bare `reach` is discovery, not a usage error"),
        (["check"], 0, "bare `reach <domain>` is discovery, not a usage error"),
        (["--help"], 0, "--help succeeds"),
        (["definitely-not-a-domain"], 2, "an unknown domain is a usage error"),
        (["check", "definitely-not-a-verb"], 2, "an unknown verb is a usage error"),
    ]
    for args, expected, why in cases:
        result = subprocess.run(
            ["reach", *args], capture_output=True, text=True, cwd=REPO_ROOT
        )
        if result.returncode != expected:
            failures.append(
                f"[exit] `reach {' '.join(args)}` exited {result.returncode}, "
                f"expected {expected} — {why}"
            )


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

    failures: list[str] = []
    check_transport_isolation(failures)
    check_single_output_path(failures)
    check_commands_decorated(failures)
    check_errors_name_a_remedy(failures)
    check_exit_codes(failures)

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

    print("test_conformance: OK — transport isolated, one output path, "
          "every command decorated and helped, every error names a remedy")
    return 0


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