127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify generated Ruiyang demo artifacts without importing the APS server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_OUTPUT_DIR = REPO_ROOT / "outputs" / "ruiyang-demo"
|
|
DEFAULT_SOURCE_DIR = Path(
|
|
os.environ.get("RUIYANG_DEMO_DIR")
|
|
or (REPO_ROOT / "demo-data" / "ruiyang-source")
|
|
)
|
|
MIN_CANONICAL = {"boms": 3, "bomItems": 42, "routings": 3, "routingSteps": 15}
|
|
EXPECTED_MRP = {"orders": 1, "make": 106, "purchase": 124, "outsource": 0}
|
|
EXPECTED_PANEL = {
|
|
"salesOrders": 1,
|
|
"make": 106,
|
|
"purchaseOrders": 124,
|
|
"outsourceOrders": 0,
|
|
"productionOrders": 1,
|
|
"workOrders": 5,
|
|
}
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
if not path.is_file():
|
|
raise FileNotFoundError(path)
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def verify(output_dir: Path, expected_source_dir: Path, business_date: str) -> dict[str, Any]:
|
|
output_dir = output_dir.resolve()
|
|
summary = load_json(output_dir / "summary.json")
|
|
manifest = load_json(output_dir / "manifest.json")
|
|
compact = load_json(output_dir / "compact-ruiyang-demo.json")
|
|
|
|
failures: list[str] = []
|
|
if Path(summary.get("sourceDir") or "").resolve() != expected_source_dir.resolve():
|
|
failures.append(f"sourceDir={summary.get('sourceDir')!r}")
|
|
if (summary.get("dates") or {}).get("businessDate") != business_date:
|
|
failures.append(f"businessDate={(summary.get('dates') or {}).get('businessDate')!r}")
|
|
if summary.get("mrp") != EXPECTED_MRP:
|
|
failures.append(f"mrp={summary.get('mrp')!r}")
|
|
if summary.get("orderPanel") != EXPECTED_PANEL:
|
|
failures.append(f"orderPanel={summary.get('orderPanel')!r}")
|
|
|
|
projection = summary.get("projection") or {}
|
|
canonical = ((projection.get("counts") or {}).get("canonical") or {})
|
|
flex = ((projection.get("counts") or {}).get("flex") or {})
|
|
if not projection.get("passed"):
|
|
failures.append("projection gate not passed")
|
|
for key, minimum in MIN_CANONICAL.items():
|
|
if int(canonical.get(key) or 0) < minimum:
|
|
failures.append(f"canonical.{key}={canonical.get(key)!r} < {minimum}")
|
|
if canonical.get("bomItems") != flex.get("bom"):
|
|
failures.append("canonical.bomItems != flex.bom")
|
|
if canonical.get("routingSteps") != flex.get("routings"):
|
|
failures.append("canonical.routingSteps != flex.routings")
|
|
|
|
schedule = summary.get("schedule") or {}
|
|
if schedule.get("poCount") != 1 or schedule.get("woCount") != 5:
|
|
failures.append(f"schedule counts={schedule.get('poCount')}/{schedule.get('woCount')}")
|
|
if schedule.get("conflictCount") != 0:
|
|
failures.append(f"schedule conflictCount={schedule.get('conflictCount')}")
|
|
if compact.get("schemaVersion") != "ruiyang-demo/1.0":
|
|
failures.append("compact schemaVersion mismatch")
|
|
|
|
artifact_count = 0
|
|
for item in manifest.get("artifacts") or []:
|
|
artifact_count += 1
|
|
path = output_dir / item["name"]
|
|
if not path.is_file():
|
|
failures.append(f"missing artifact {item['name']}")
|
|
continue
|
|
if path.stat().st_size != item["bytes"]:
|
|
failures.append(f"size mismatch {item['name']}")
|
|
if sha256(path) != item["sha256"]:
|
|
failures.append(f"sha256 mismatch {item['name']}")
|
|
|
|
result = {
|
|
"passed": not failures,
|
|
"outputDir": str(output_dir),
|
|
"sourceDir": str(expected_source_dir.resolve()),
|
|
"businessDate": business_date,
|
|
"artifactCount": artifact_count,
|
|
"canonical": canonical,
|
|
"mrp": summary.get("mrp"),
|
|
"orderPanel": summary.get("orderPanel"),
|
|
"schedule": {
|
|
"poCount": schedule.get("poCount"),
|
|
"woCount": schedule.get("woCount"),
|
|
"conflictCount": schedule.get("conflictCount"),
|
|
},
|
|
"failures": failures,
|
|
}
|
|
if failures:
|
|
raise RuntimeError("Ruiyang demo verification failed: " + "; ".join(failures))
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Verify Ruiyang demo outputs")
|
|
parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
|
|
parser.add_argument("--expected-source-dir", type=Path, default=DEFAULT_SOURCE_DIR)
|
|
parser.add_argument("--business-date", default="2026-08-05")
|
|
args = parser.parse_args()
|
|
print(json.dumps(
|
|
verify(args.output_dir, args.expected_source_dir, args.business_date),
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|