aps-agent/tests/golden/test_ruiyang_demo_flow.py

120 lines
4.4 KiB
Python
Raw Permalink Normal View History

from __future__ import annotations
import hashlib
import json
from pathlib import Path
import pytest
from scripts.ruiyang_demo import DEFAULT_SOURCE_DIR, MIN_CANONICAL, run_demo
pytestmark = pytest.mark.skipif(
not DEFAULT_SOURCE_DIR.exists(),
reason=f"真实锐扬案例目录不存在: {DEFAULT_SOURCE_DIR}",
)
@pytest.fixture(scope="module")
def ruiyang_demo(tmp_path_factory: pytest.TempPathFactory) -> tuple[dict, Path]:
output_dir = tmp_path_factory.mktemp("ruiyang-demo")
summary = run_demo(
source_dir=DEFAULT_SOURCE_DIR,
output_dir=output_dir,
business_date="2026-08-05",
schedule_start="2026-08-06",
due_date="2026-09-04",
planning_horizon_days=30,
assume_kitted=True,
run_strict_check=True,
)
return summary, output_dir
def _load(output_dir: Path, name: str) -> dict:
return json.loads((output_dir / name).read_text(encoding="utf-8"))
def test_ruiyang_real_folder_imports_dual_projection_and_mrp(ruiyang_demo):
summary, output_dir = ruiyang_demo
inventory = _load(output_dir, "00-source-inventory.json")
projection = _load(output_dir, "02-import-projection.json")["projectionGate"]
mrp = _load(output_dir, "03-mrp.json")
strict = _load(output_dir, "04-strict-readiness.json")
assert inventory["workbook"]["name"] == "湖南锐扬MOM主数据收集表.xlsx"
assert inventory["workbook"]["sheetCount"] == 9
assert len(inventory["dxf"]) == 3
assert projection["passed"] is True
canonical = projection["counts"]["canonical"]
flex = projection["counts"]["flex"]
for key, minimum in MIN_CANONICAL.items():
assert canonical[key] >= minimum
assert canonical["bomItems"] == flex["bom"]
assert canonical["routingSteps"] == flex["routings"]
assert canonical["materials"] == flex["materials"]
assert canonical["salesOrders"] == flex["orders"]
assert mrp["counts"] == {"orders": 1, "make": 106, "purchase": 124, "outsource": 0}
assert strict["solveStatus"] == "BLOCKED"
assert strict["blockerCounts"]["TEMPLATE_ROUTING_UNCONFIRMED"] >= 1
assert strict["blockerCounts"]["UNTRUSTED_DRAFT_SUPPLY"] >= 1
assert summary["dates"] == {
"businessDate": "2026-08-05",
"scheduleStart": "2026-08-06",
"dueDate": "2026-09-04",
}
def test_ruiyang_demo_adapter_creates_production_order_and_work_orders(ruiyang_demo):
summary, output_dir = ruiyang_demo
adapter = _load(output_dir, "05-demo-adapter.json")
schedule = _load(output_dir, "06-schedule.json")
panel = _load(output_dir, "07-order-panel.json")
manifest = _load(output_dir, "manifest.json")
assert [row["operationCode"] for row in adapter["selectedEquipment"]] == [
"CUT", "BEND", "WELD", "PAINT", "ASM",
]
assert all(row["equipmentCode"] for row in adapter["selectedEquipment"])
assert adapter["assumeKitted"] is True
assert schedule["result"]["poCount"] == 1
assert schedule["result"]["woCount"] == 5
assert schedule["result"]["conflictCount"] == 0
assert len(schedule["productionOrders"]) == 1
assert len(schedule["workOrders"]) == 5
assert len(panel["make"]) == 106
assert len(panel["purchaseOrders"]) == 124
assert len(panel["productionOrders"]) == 1
assert len(panel["workOrders"]) == 5
assert summary["orderPanel"] == {
"salesOrders": 1,
"make": 106,
"purchaseOrders": 124,
"outsourceOrders": 0,
"productionOrders": 1,
"workOrders": 5,
}
for item in manifest["artifacts"]:
path = output_dir / item["name"]
assert path.exists()
assert path.stat().st_size == item["bytes"]
assert hashlib.sha256(path.read_bytes()).hexdigest() == item["sha256"]
def test_ruiyang_compact_payload_and_standalone_verifier(ruiyang_demo):
_, output_dir = ruiyang_demo
from scripts.verify_ruiyang_demo import verify
compact = _load(output_dir, "compact-ruiyang-demo.json")
result = verify(output_dir, DEFAULT_SOURCE_DIR, "2026-08-05")
assert compact["schemaVersion"] == "ruiyang-demo/1.0"
assert compact["source"]["sourceDir"] == str(DEFAULT_SOURCE_DIR.resolve())
assert compact["mrp"]["counts"] == {"orders": 1, "make": 106, "purchase": 124, "outsource": 0}
assert len(compact["schedule"]["productionOrders"]) == 1
assert len(compact["schedule"]["workOrders"]) == 5
assert result["passed"] is True