93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_WORKFLOW = ROOT / ".gitea" / "workflows" / "ci.yml"
|
|
|
|
|
|
def _load_yaml(path: Path) -> dict:
|
|
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Lint the QuantEngine CI workflow for recurring service-binding mistakes.")
|
|
ap.add_argument("--workflow", default=str(DEFAULT_WORKFLOW))
|
|
args = ap.parse_args()
|
|
|
|
workflow_path = Path(args.workflow)
|
|
if not workflow_path.is_absolute():
|
|
workflow_path = (ROOT / workflow_path).resolve()
|
|
data = _load_yaml(workflow_path)
|
|
|
|
errors: list[str] = []
|
|
evidence: dict[str, object] = {
|
|
"workflow": str(workflow_path.relative_to(ROOT)),
|
|
"jobs": sorted((data.get("jobs") or {}).keys()),
|
|
}
|
|
|
|
jobs = data.get("jobs") or {}
|
|
core = jobs.get("validate-core") or {}
|
|
services = core.get("services") or {}
|
|
postgres = services.get("postgres") or {}
|
|
ci_text = workflow_path.read_text(encoding="utf-8")
|
|
|
|
ports = postgres.get("ports") or []
|
|
if any(str(port).strip() == "5432:5432" for port in ports):
|
|
errors.append("validate-core.services.postgres.ports contains fixed host mapping 5432:5432")
|
|
|
|
if "PGHOST: postgres" not in ci_text:
|
|
errors.append("workflow does not pin PGHOST=postgres for CI database steps")
|
|
|
|
if "QE_WBS_PG_DSN=host=postgres" not in ci_text:
|
|
errors.append("workflow does not publish QE_WBS_PG_DSN with service hostname")
|
|
|
|
spec_path = ROOT / "spec" / "60_quant_engine_wbs.yaml"
|
|
spec = _load_yaml(spec_path)
|
|
done_tasks: list[str] = []
|
|
for task_id, task in (spec.get("tasks") or {}).items():
|
|
if (task or {}).get("status") != "DONE":
|
|
continue
|
|
mode = ((task or {}).get("execution") or {}).get("mode")
|
|
if mode == "not_ci_reproducible":
|
|
continue
|
|
done_tasks.append(task_id)
|
|
done_tasks.sort()
|
|
|
|
match = re.search(
|
|
r"for task in (?P<tasks>.+?); do\s+python3 tools/verify_wbs_task_v1\.py --task \"?\$?task\"?\s+done",
|
|
ci_text,
|
|
re.S,
|
|
)
|
|
if not match:
|
|
errors.append("workflow missing Generate DONE WBS Verdicts task list")
|
|
else:
|
|
declared = re.findall(r"QE-[A-Z0-9-]+", match.group("tasks"))
|
|
missing = [task for task in done_tasks if task not in declared]
|
|
extra = [task for task in declared if task not in done_tasks]
|
|
if missing:
|
|
errors.append("workflow omits DONE verdict tasks: " + ", ".join(missing))
|
|
if extra:
|
|
errors.append("workflow includes non-DONE verdict tasks: " + ", ".join(extra))
|
|
|
|
result = {
|
|
"formula_id": "GITEA_CI_WORKFLOW_LINT_V1",
|
|
"gate": "PASS" if not errors else "FAIL",
|
|
"errors": errors,
|
|
"evidence": evidence,
|
|
}
|
|
out = ROOT / "Temp" / "gitea_ci_workflow_lint_v1.json"
|
|
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if not errors else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|