487 lines
17 KiB
Python
487 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
REPORT_JSON = "gate-run-report.json"
|
|
REPORT_MD = "gate-run-report.md"
|
|
|
|
|
|
class ManifestError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class Gate:
|
|
gate_id: str
|
|
command: str | list[str]
|
|
working_directory: pathlib.Path
|
|
expected_exit_code: int
|
|
log_file: pathlib.Path
|
|
timeout_seconds: float
|
|
required_before_review: bool
|
|
|
|
|
|
def load_raw_manifest(path: pathlib.Path) -> Any:
|
|
text = path.read_text(encoding="utf-8")
|
|
suffix = path.suffix.lower()
|
|
if suffix == ".json":
|
|
return json.loads(text)
|
|
if suffix in {".yaml", ".yml"}:
|
|
return yaml.safe_load(text)
|
|
if suffix in {".md", ".markdown"}:
|
|
fenced = re.search(r"```(yaml|yml|json)\s*\n(.*?)\n```", text, flags=re.DOTALL | re.I)
|
|
if not fenced:
|
|
raise ManifestError("Markdown manifest must contain a fenced yaml, yml, or json block.")
|
|
language = fenced.group(1).casefold()
|
|
body = fenced.group(2)
|
|
return json.loads(body) if language == "json" else yaml.safe_load(body)
|
|
raise ManifestError("Gate manifest must be JSON, YAML, YML, or Markdown.")
|
|
|
|
|
|
def manifest_gates(raw: Any) -> list[Mapping[str, Any]]:
|
|
if isinstance(raw, Mapping):
|
|
gates = raw.get("gates")
|
|
else:
|
|
gates = raw
|
|
if not isinstance(gates, list):
|
|
raise ManifestError("Gate manifest must define a gates list.")
|
|
normalized: list[Mapping[str, Any]] = []
|
|
for index, item in enumerate(gates):
|
|
if not isinstance(item, Mapping):
|
|
raise ManifestError(f"Gate at index {index} must be an object.")
|
|
normalized.append(item)
|
|
return normalized
|
|
|
|
|
|
def bool_value(value: Any, default: bool) -> bool:
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
return value.strip().casefold() in {"1", "true", "yes", "y"}
|
|
return bool(value)
|
|
|
|
|
|
def validate_gate_id(value: Any) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ManifestError("Every gate must define a non-empty gate_id.")
|
|
gate_id = value.strip()
|
|
if any(separator in gate_id for separator in ("/", "\\", os.sep)):
|
|
raise ManifestError(f"gate_id must not contain path separators: {gate_id}")
|
|
return gate_id
|
|
|
|
|
|
def validate_command(value: Any, gate_id: str) -> str | list[str]:
|
|
if isinstance(value, str) and value.strip():
|
|
return value
|
|
if isinstance(value, list) and value and all(isinstance(item, str) and item for item in value):
|
|
return value
|
|
raise ManifestError(f"Gate {gate_id} must define command as a non-empty string or string list.")
|
|
|
|
|
|
def resolve_under_root(base: pathlib.Path, raw_path: Any, default: str, label: str) -> pathlib.Path:
|
|
candidate_text = default if raw_path in {None, ""} else str(raw_path)
|
|
candidate = pathlib.Path(candidate_text)
|
|
if not candidate.is_absolute():
|
|
candidate = base / candidate
|
|
resolved = candidate.resolve()
|
|
try:
|
|
resolved.relative_to(base)
|
|
except ValueError as exc:
|
|
raise ManifestError(f"{label} must resolve under project_root: {candidate_text}") from exc
|
|
return resolved
|
|
|
|
|
|
def resolve_log_file(output_dir: pathlib.Path, raw_path: Any, gate_id: str) -> pathlib.Path:
|
|
candidate_text = f"logs/{gate_id}.log" if raw_path in {None, ""} else str(raw_path)
|
|
candidate = pathlib.Path(candidate_text)
|
|
if candidate.is_absolute():
|
|
resolved = candidate.resolve()
|
|
else:
|
|
resolved = (output_dir / candidate).resolve()
|
|
try:
|
|
resolved.relative_to(output_dir)
|
|
except ValueError as exc:
|
|
raise ManifestError(f"log_file must resolve under output_dir for gate {gate_id}: {candidate_text}") from exc
|
|
return resolved
|
|
|
|
|
|
def parse_gates(
|
|
raw_manifest: Any,
|
|
project_root: pathlib.Path,
|
|
output_dir: pathlib.Path,
|
|
) -> tuple[list[Gate], list[str], list[str]]:
|
|
gates: list[Gate] = []
|
|
errors: list[str] = []
|
|
declared: list[str] = []
|
|
for index, raw_gate in enumerate(manifest_gates(raw_manifest)):
|
|
try:
|
|
gate_id = validate_gate_id(raw_gate.get("gate_id"))
|
|
declared.append(gate_id)
|
|
command = validate_command(raw_gate.get("command"), gate_id)
|
|
working_directory = resolve_under_root(
|
|
project_root,
|
|
raw_gate.get("working_directory"),
|
|
".",
|
|
f"working_directory for gate {gate_id}",
|
|
)
|
|
expected_exit_code = int(raw_gate.get("expected_exit_code", 0))
|
|
timeout_seconds = float(raw_gate.get("timeout_seconds", 300))
|
|
if timeout_seconds <= 0:
|
|
raise ManifestError(f"timeout_seconds must be positive for gate {gate_id}.")
|
|
log_file = resolve_log_file(output_dir, raw_gate.get("log_file"), gate_id)
|
|
required = bool_value(raw_gate.get("required_before_review"), True)
|
|
gates.append(
|
|
Gate(
|
|
gate_id=gate_id,
|
|
command=command,
|
|
working_directory=working_directory,
|
|
expected_exit_code=expected_exit_code,
|
|
log_file=log_file,
|
|
timeout_seconds=timeout_seconds,
|
|
required_before_review=required,
|
|
)
|
|
)
|
|
except ManifestError as exc:
|
|
errors.append(f"gate[{index}]: {exc}")
|
|
return gates, errors, declared
|
|
|
|
|
|
def empty_report(project_root: pathlib.Path, manifest_path: pathlib.Path, mode: str) -> dict[str, Any]:
|
|
return {
|
|
"project_root": str(project_root),
|
|
"gate_manifest_path": str(manifest_path),
|
|
"mode": mode,
|
|
"commands_declared": [],
|
|
"commands_run": [],
|
|
"commands_skipped": [],
|
|
"exit_codes": {},
|
|
"pass_fail_status": "PENDING",
|
|
"duration": 0.0,
|
|
"log_paths": {},
|
|
"environment_notes": [
|
|
"Commands are declared by the manifest.",
|
|
"Passing gates are engineering evidence only; they do not imply product acceptance or lifecycle approval.",
|
|
],
|
|
"gate_results": [],
|
|
"manifest_errors": [],
|
|
"changed_files": [],
|
|
"machine_readable_summary": {},
|
|
}
|
|
|
|
|
|
def should_exclude(path: pathlib.Path, output_dir: pathlib.Path, project_root: pathlib.Path) -> bool:
|
|
try:
|
|
path.relative_to(output_dir)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
path.relative_to(project_root / ".git")
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def snapshot_files(project_root: pathlib.Path, output_dir: pathlib.Path) -> dict[str, tuple[int, int]]:
|
|
snapshot: dict[str, tuple[int, int]] = {}
|
|
for path in project_root.rglob("*"):
|
|
if not path.is_file() or should_exclude(path, output_dir, project_root):
|
|
continue
|
|
stat = path.stat()
|
|
snapshot[path.relative_to(project_root).as_posix()] = (stat.st_size, stat.st_mtime_ns)
|
|
return snapshot
|
|
|
|
|
|
def diff_snapshots(before: dict[str, tuple[int, int]], after: dict[str, tuple[int, int]]) -> list[str]:
|
|
changed = {
|
|
path
|
|
for path in set(before) | set(after)
|
|
if before.get(path) != after.get(path)
|
|
}
|
|
return sorted(changed)
|
|
|
|
|
|
def command_display(command: str | list[str]) -> str:
|
|
if isinstance(command, list):
|
|
return " ".join(command)
|
|
return command
|
|
|
|
|
|
def command_for_subprocess(command: str | list[str]) -> tuple[str | list[str], bool]:
|
|
if isinstance(command, list):
|
|
return command, False
|
|
return command, True
|
|
|
|
|
|
def safe_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, bytes):
|
|
return value.decode("utf-8", errors="replace")
|
|
return str(value)
|
|
|
|
|
|
def run_gate(
|
|
gate: Gate,
|
|
project_root: pathlib.Path,
|
|
output_dir: pathlib.Path,
|
|
) -> dict[str, Any]:
|
|
start = time.monotonic()
|
|
before = snapshot_files(project_root, output_dir)
|
|
command, shell = command_for_subprocess(gate.command)
|
|
status = "PENDING"
|
|
exit_code: int | None = None
|
|
stdout = ""
|
|
stderr = ""
|
|
timeout = False
|
|
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
cwd=str(gate.working_directory),
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=gate.timeout_seconds,
|
|
shell=shell,
|
|
check=False,
|
|
)
|
|
exit_code = completed.returncode
|
|
stdout = completed.stdout
|
|
stderr = completed.stderr
|
|
status = "PASS" if exit_code == gate.expected_exit_code else "FAIL"
|
|
except subprocess.TimeoutExpired as exc:
|
|
timeout = True
|
|
stdout = safe_text(exc.stdout or exc.output)
|
|
stderr = safe_text(exc.stderr)
|
|
status = "TIMEOUT"
|
|
duration = time.monotonic() - start
|
|
after = snapshot_files(project_root, output_dir)
|
|
changed_files = diff_snapshots(before, after)
|
|
|
|
gate.log_file.parent.mkdir(parents=True, exist_ok=True)
|
|
log_lines = [
|
|
f"gate_id: {gate.gate_id}",
|
|
f"working_directory: {gate.working_directory}",
|
|
f"command: {command_display(gate.command)}",
|
|
f"expected_exit_code: {gate.expected_exit_code}",
|
|
f"exit_code: {exit_code}",
|
|
f"status: {status}",
|
|
f"duration_seconds: {duration:.3f}",
|
|
"",
|
|
"## stdout",
|
|
stdout,
|
|
"",
|
|
"## stderr",
|
|
stderr,
|
|
]
|
|
if timeout:
|
|
log_lines.extend(["", f"Command timed out after {gate.timeout_seconds} seconds."])
|
|
gate.log_file.write_text("\n".join(log_lines), encoding="utf-8")
|
|
|
|
return {
|
|
"gate_id": gate.gate_id,
|
|
"command": command_display(gate.command),
|
|
"working_directory": str(gate.working_directory),
|
|
"expected_exit_code": gate.expected_exit_code,
|
|
"exit_code": exit_code,
|
|
"status": status,
|
|
"duration": duration,
|
|
"log_path": gate.log_file.relative_to(output_dir).as_posix(),
|
|
"required_before_review": gate.required_before_review,
|
|
"timeout": timeout,
|
|
"changed_files": changed_files,
|
|
}
|
|
|
|
|
|
def finalize_report(report: dict[str, Any]) -> None:
|
|
if report["manifest_errors"]:
|
|
status = "ERROR"
|
|
elif report["mode"] == "dry_run":
|
|
status = "DRY_RUN"
|
|
elif any(result["status"] in {"FAIL", "TIMEOUT"} for result in report["gate_results"]):
|
|
status = "FAIL"
|
|
else:
|
|
status = "PASS"
|
|
report["pass_fail_status"] = status
|
|
report["duration"] = round(float(report["duration"]), 3)
|
|
report["changed_files"] = sorted(
|
|
{
|
|
changed
|
|
for result in report["gate_results"]
|
|
for changed in result.get("changed_files", [])
|
|
}
|
|
)
|
|
report["machine_readable_summary"] = {
|
|
"status": status,
|
|
"commands_declared_count": len(report["commands_declared"]),
|
|
"commands_run_count": len(report["commands_run"]),
|
|
"commands_skipped_count": len(report["commands_skipped"]),
|
|
"manifest_error_count": len(report["manifest_errors"]),
|
|
"failed_count": sum(1 for result in report["gate_results"] if result["status"] == "FAIL"),
|
|
"timeout_count": sum(1 for result in report["gate_results"] if result["status"] == "TIMEOUT"),
|
|
"changed_file_count": len(report["changed_files"]),
|
|
}
|
|
|
|
|
|
def write_json_report(report: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
(output_dir / REPORT_JSON).write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def write_markdown_report(report: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
summary = report["machine_readable_summary"]
|
|
lines = [
|
|
"# Gate Run Report",
|
|
"",
|
|
"## Summary",
|
|
"",
|
|
f"- Status: `{summary['status']}`",
|
|
f"- Commands declared: {summary['commands_declared_count']}",
|
|
f"- Commands run: {summary['commands_run_count']}",
|
|
f"- Commands skipped: {summary['commands_skipped_count']}",
|
|
f"- Manifest errors: {summary['manifest_error_count']}",
|
|
f"- Changed files: {summary['changed_file_count']}",
|
|
"",
|
|
"## Gate Results",
|
|
"",
|
|
]
|
|
if report["gate_results"]:
|
|
for result in report["gate_results"]:
|
|
lines.append(
|
|
f"- `{result['gate_id']}`: `{result['status']}` exit `{result['exit_code']}` log `{result['log_path']}`"
|
|
)
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Skipped Gates", ""])
|
|
if report["commands_skipped"]:
|
|
for skipped in report["commands_skipped"]:
|
|
lines.append(f"- `{skipped['gate_id']}`: {skipped['reason']}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Manifest Errors", ""])
|
|
if report["manifest_errors"]:
|
|
for error in report["manifest_errors"]:
|
|
lines.append(f"- {error}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Changed Files", ""])
|
|
if report["changed_files"]:
|
|
for path in report["changed_files"]:
|
|
lines.append(f"- `{path}`")
|
|
else:
|
|
lines.append("- None")
|
|
(output_dir / REPORT_MD).write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def write_reports(report: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
write_json_report(report, output_dir)
|
|
write_markdown_report(report, output_dir)
|
|
|
|
|
|
def run(
|
|
project_root: pathlib.Path,
|
|
manifest_path: pathlib.Path,
|
|
output_dir: pathlib.Path,
|
|
mode: str,
|
|
include_optional: bool = False,
|
|
) -> int:
|
|
project_root = project_root.resolve()
|
|
manifest_path = manifest_path.resolve()
|
|
output_dir = output_dir.resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
report = empty_report(project_root, manifest_path, mode)
|
|
report["environment_notes"].extend(
|
|
[
|
|
f"python: {sys.version.split()[0]}",
|
|
f"platform: {platform.platform()}",
|
|
]
|
|
)
|
|
|
|
if not project_root.is_dir():
|
|
report["manifest_errors"].append(f"project_root is not a directory: {project_root}")
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
if not manifest_path.is_file():
|
|
report["manifest_errors"].append(f"gate_manifest_path does not exist: {manifest_path}")
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
|
|
try:
|
|
raw_manifest = load_raw_manifest(manifest_path)
|
|
gates, errors, declared = parse_gates(raw_manifest, project_root, output_dir)
|
|
report["commands_declared"] = declared
|
|
report["manifest_errors"].extend(errors)
|
|
except Exception as exc:
|
|
report["manifest_errors"].append(str(exc))
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
|
|
if report["manifest_errors"]:
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
|
|
run_start = time.monotonic()
|
|
for gate in gates:
|
|
if mode == "dry_run":
|
|
report["commands_skipped"].append({"gate_id": gate.gate_id, "reason": "dry_run"})
|
|
continue
|
|
if not gate.required_before_review and not include_optional:
|
|
report["commands_skipped"].append({"gate_id": gate.gate_id, "reason": "optional"})
|
|
continue
|
|
result = run_gate(gate, project_root, output_dir)
|
|
report["gate_results"].append(result)
|
|
report["commands_run"].append(gate.gate_id)
|
|
report["exit_codes"][gate.gate_id] = result["exit_code"]
|
|
report["log_paths"][gate.gate_id] = result["log_path"]
|
|
|
|
report["duration"] = time.monotonic() - run_start
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 0 if report["pass_fail_status"] in {"PASS", "DRY_RUN"} else 1
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Run or dry-run declared regression validation gates.")
|
|
parser.add_argument("--project-root", required=True, type=pathlib.Path)
|
|
parser.add_argument("--gate-manifest", required=True, type=pathlib.Path)
|
|
parser.add_argument("--output-dir", required=True, type=pathlib.Path)
|
|
parser.add_argument("--mode", choices=["dry_run", "run"], required=True)
|
|
parser.add_argument("--include-optional", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
return run(
|
|
project_root=args.project_root,
|
|
manifest_path=args.gate_manifest,
|
|
output_dir=args.output_dir,
|
|
mode=args.mode,
|
|
include_optional=args.include_optional,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|