Every non-zero exit names the command that would fix it, and still exits non-zero. Both halves matter; the second is the one that gets lost, because a tool that explains itself beautifully and exits 0 looks MORE correct while having silently disabled its own gate. core/errors.py holds ReachError(message, fix=) and @handle_errors. core/logging.py holds @logged, emitting through console rather than a second sink — one output path, so there is nothing to drift. core/command.py composes them, and the order is load-bearing: handle_errors wraps logged, so the logger sees the original exception. Inverted, every failure would be recorded as "SystemExit" and the log would say nothing about what went wrong while looking like it worked. core/ raises SystemExit, not typer.Exit. A service must be callable from a test, another service, or a future second front end, and an exception type that only makes sense inside a CLI leaks the transport into every layer. The check router is retrofitted off its hand-rolled verdict-and-exit pattern — exactly the boilerplate this removes — and test_check_parity.py passes unchanged across the retrofit. That test predates the decorators and pins exit codes against the old script, so it is independent evidence, not a test tuned to match new behaviour. Unknown domains and unknown verbs now enumerate what exists instead of only saying no. That needed a shared group class, which collided with "no typer outside main.py and router.py" — resolved by sharpening the invariant rather than breaking it, since its purpose is that a SERVICE never knows it was called from a CLI. Transport now lives in main.py, router.py and core/cli.py; never in service.py, schemas.py or helpers.py. The upside is that cli.domain() carries the settings that were previously per-router decisions, including the load-bearing rich_markup_mode=None that one forgetful domain could have undone. test_conformance.py makes five invariants executable, AST-based rather than grep. Scoped to the package, not the 123 legacy scripts — and deliberately so: as T-1250 moves each script into domains/, it lands inside the scope and the rules start applying automatically, so the test's reach grows with the migration. Proven to fail before being trusted: removing @command and removing a fix= each produced a failure naming the file, the line and the reason. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
218 lines
8.4 KiB
Python
218 lines
8.4 KiB
Python
#!/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/<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 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())
|