"""Run the complete backend structural, migration, lint and type validation.""" from __future__ import annotations import subprocess import sys from collections.abc import Callable, Sequence from dataclasses import dataclass @dataclass(frozen=True) class ValidationCheck: name: str command: tuple[str, ...] # Keep all current backend validation in one explicit, reviewable list. The # workflow owns setup and credentials; this helper owns command/result handling. VALIDATION_CHECKS: tuple[ValidationCheck, ...] = ( ValidationCheck("alembic chain", ("uv", "run", "python", "scripts/check_alembic_chain.py")), ValidationCheck( "savepoint isolation", ("uv", "run", "python", "scripts/check_savepoint_isolation.py"), ), ValidationCheck("TiDB parity", ("uv", "run", "python", "scripts/check_tidb_parity.py")), ValidationCheck( "trigger creation with binlog", ( "mysql", "-h", "127.0.0.1", "-u", "root", "-prootpass", "-e", "SET GLOBAL log_bin_trust_function_creators = 1;", ), ), ValidationCheck("alembic upgrade", ("uv", "run", "alembic", "upgrade", "head")), ValidationCheck("uv lock", ("uv", "lock", "--check")), ValidationCheck("ruff", ("uv", "run", "ruff", "check", ".")), ValidationCheck("mypy baseline", ("uv", "run", "python", "scripts/check_mypy_baseline.py")), ValidationCheck( "DAX approval callback dispatch", ( "uv", "run", "pytest", "--noconftest", "-q", "tests/test_approval/test_data_access_callback_dispatch.py", ), ), ValidationCheck( "DAX apply config preservation", ( "uv", "run", "pytest", "--noconftest", "-q", "tests/test_data_access/test_service_account_apply_gate.py", ), ), ValidationCheck( "GitHub Actions trust boundary", ( "uv", "run", "pytest", "--noconftest", "-q", "tests/test_deploy/test_github_actions_security_2026_08_02.py", ), ), ) Runner = Callable[[Sequence[str]], int] def run_validation( checks: Sequence[ValidationCheck] = VALIDATION_CHECKS, *, runner: Runner | None = None, ) -> int: """Run every diagnostic; any failed command fails the job after collection.""" if runner is None: runner = _subprocess_runner failures: list[str] = [] for check in checks: print(f"::group::{check.name}", flush=True) try: code = runner(check.command) finally: print("::endgroup::", flush=True) if code: failures.append(f"{check.name} (exit {code})") if failures: print("backend validation failed: " + ", ".join(failures), file=sys.stderr) return 1 return 0 def _subprocess_runner(command: Sequence[str]) -> int: return subprocess.run(command, check=False).returncode if __name__ == "__main__": raise SystemExit(run_validation())