feat(wbs): add execution plan validator and wiring
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
REQUIRED_WP_IDS = {
|
||||
"WBS-10-WP1",
|
||||
"WBS-10-WP2",
|
||||
"WBS-10-WP3",
|
||||
"WBS-10-WP4",
|
||||
}
|
||||
|
||||
|
||||
def load_yaml(path: Path) -> dict[str, Any]:
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(path)
|
||||
return yaml.safe_load(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Validate WBS-10 dotnet migration execution plan")
|
||||
parser.add_argument("--plan", default="docs/WBS_10_DOTNET_MIGRATION_EXECUTION_PLAN.yaml")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
plan_path = Path(args.plan).resolve()
|
||||
payload: dict[str, Any] = {
|
||||
"formula_id": "WBS_10_DOTNET_MIGRATION_EXECUTION_PLAN_V1",
|
||||
"gate": "FAIL",
|
||||
"missing": [],
|
||||
"evidence": {"plan": str(plan_path)},
|
||||
}
|
||||
|
||||
try:
|
||||
data = load_yaml(plan_path)
|
||||
except FileNotFoundError:
|
||||
payload["missing"].append("plan missing")
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
return 1
|
||||
|
||||
if data.get("formula_id") != "WBS_10_DOTNET_MIGRATION_EXECUTION_PLAN_V1":
|
||||
payload["missing"].append("formula_id")
|
||||
if data.get("status") != "draft":
|
||||
payload["missing"].append("status")
|
||||
|
||||
work_packages = data.get("work_packages") or []
|
||||
wp_ids: set[str] = set()
|
||||
for wp in work_packages:
|
||||
wp_id = wp.get("wp_id", "")
|
||||
wp_ids.add(wp_id)
|
||||
if not wp.get("title"):
|
||||
payload["missing"].append(f"{wp_id}.title")
|
||||
if not wp.get("objective"):
|
||||
payload["missing"].append(f"{wp_id}.objective")
|
||||
if not wp.get("depends_on"):
|
||||
payload["missing"].append(f"{wp_id}.depends_on")
|
||||
if not wp.get("inputs"):
|
||||
payload["missing"].append(f"{wp_id}.inputs")
|
||||
if not wp.get("outputs"):
|
||||
payload["missing"].append(f"{wp_id}.outputs")
|
||||
success_data = wp.get("success_data") or {}
|
||||
if "schema" not in success_data:
|
||||
payload["missing"].append(f"{wp_id}.success_data.schema")
|
||||
if "fields" not in success_data:
|
||||
payload["missing"].append(f"{wp_id}.success_data.fields")
|
||||
if "pass_condition" not in success_data:
|
||||
payload["missing"].append(f"{wp_id}.success_data.pass_condition")
|
||||
|
||||
for required in sorted(REQUIRED_WP_IDS - wp_ids):
|
||||
payload["missing"].append(f"{required} missing")
|
||||
|
||||
execution_order = data.get("execution_order") or []
|
||||
if execution_order != ["WBS-10-WP1", "WBS-10-WP2", "WBS-10-WP3", "WBS-10-WP4"]:
|
||||
payload["missing"].append("execution_order")
|
||||
|
||||
payload["gate"] = "PASS" if not payload["missing"] else "FAIL"
|
||||
payload["message"] = (
|
||||
"WBS-10 dotnet migration execution plan validation passed."
|
||||
if payload["gate"] == "PASS"
|
||||
else "WBS-10 dotnet migration execution plan validation failed."
|
||||
)
|
||||
|
||||
out_path = plan_path.parent.parent / "Temp" / "wbs_10_dotnet_migration_execution_plan_v1.json"
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
return 0 if payload["gate"] == "PASS" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user