#!/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 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//, 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 main() -> int: failures: list[str] = [] check_transport_isolation(failures) check_single_output_path(failures) check_commands_decorated(failures) check_errors_name_a_remedy(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())