Files
QuantEngineByItz/tools/validate_dotnet_read_model_contract_v1.py
kjh2064 d610ecb57c
CI Workflow Lint / validate-ci-workflow-lint (push) Failing after 13s
Validators (Pushes and Pull Requests) / validate-ui-and-storage (push) Successful in 23s
Validators (Pushes and Pull Requests) / validate-core (push) Successful in 2m11s
feat(dotnet): add read model contract harness
2026-07-13 00:41:30 +09:00

80 lines
2.9 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
import yaml
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 read model contract")
parser.add_argument("--contract", default="docs/WBS_10_DOTNET_READ_MODEL_CONTRACT.yaml")
args = parser.parse_args(argv)
contract_path = Path(args.contract).resolve()
payload: dict[str, Any] = {
"formula_id": "WBS_10_DOTNET_READ_MODEL_CONTRACT_V1",
"gate": "FAIL",
"missing": [],
"evidence": {"contract": str(contract_path)},
}
try:
data = load_yaml(contract_path)
except FileNotFoundError:
payload["missing"].append("contract missing")
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 1
if data.get("formula_id") != "WBS_10_DOTNET_READ_MODEL_CONTRACT_V1":
payload["missing"].append("formula_id")
if data.get("goal") != "운영 화면과 조회 API의 read model 경계를 분리한다.":
payload["missing"].append("goal")
models = data.get("read_models") or []
if len(models) < 3:
payload["missing"].append("read_models")
for model in models:
if not model.get("model_id"):
payload["missing"].append("model_id")
if not model.get("purpose"):
payload["missing"].append(f"{model.get('model_id', 'unknown')}.purpose")
if not model.get("source"):
payload["missing"].append(f"{model.get('model_id', 'unknown')}.source")
if not model.get("consumers"):
payload["missing"].append(f"{model.get('model_id', 'unknown')}.consumers")
if not model.get("fields"):
payload["missing"].append(f"{model.get('model_id', 'unknown')}.fields")
if not model.get("staleness_budget"):
payload["missing"].append(f"{model.get('model_id', 'unknown')}.staleness_budget")
rules = data.get("rules") or []
if len(rules) < 4:
payload["missing"].append("rules")
payload["gate"] = "PASS" if not payload["missing"] else "FAIL"
payload["message"] = (
"WBS-10 dotnet read model contract validation passed."
if payload["gate"] == "PASS"
else "WBS-10 dotnet read model contract validation failed."
)
out_path = contract_path.parent.parent / "Temp" / "wbs_10_dotnet_read_model_contract_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())