aps-agent/tests/golden/test_readiness.py

# ============================================================
# M-B 黄金测试：数据齐备度 + 工时来源 + TIME_* 冲突标注
# ============================================================
from __future__ import annotations

from server.aps_domain.readiness import (
    annotate_time_conflicts, check_readiness, readiness_text,
    resolve_step_time, time_matrix,
)
from server.state.seed import seed_world


def test_demo_world_readiness_mostly_ready():
    world = seed_world()
    report = check_readiness(world)
    s = report["summary"]
    assert s["total"] == len([o for o in world["flexOrders"] if o["status"] not in ("DONE", "CANCELLED")])
    # 演示厂路线无覆盖工时但设备 opStdTime 齐 → 不算待维护
    assert s["timePending"] == 0
    assert s["ready"] >= 1
    assert "数据齐备度" in readiness_text(report) or "我帮你看了看" in readiness_text(report)


def test_missing_equipment_time_flags_unmaintained():
    world = seed_world()
    # 摘掉 OP-ASSY 的设备默认工时 → 该步变「待维护」
    for e in world["flexEquipment"]:
        e.get("opStdTime", {}).pop("OP-ASSY", None)
    report = check_readiness(world)
    assert report["summary"]["timePending"] > 0
    assy_orders = [r for r in report["orders"] if r["productCode"] == "HV-HARNESS"]
    assert any(i["type"] == "TIME_UNMAINTAINED" for r in assy_orders for i in r["issues"])
    assert all(not r["ready"] for r in assy_orders)


def test_inferred_time_is_warning_not_blocker():
    world = seed_world()
    for r in world["flexRoutings"]:
        if r["productCode"] == "HV-CONN" and r["operationCode"] == "OP-ASSY":
            r["stdTimePerUnit"] = 30.0
            r["stdTimeSource"] = "推断"
    report = check_readiness(world)
    conn = [r for r in report["orders"] if r["productCode"] == "HV-CONN"]
    assert any(i["type"] == "TIME_INFERRED" and i["severity"] == "warn"
               for r in conn for i in r["issues"])
    assert all(r["ready"] for r in conn)          # 推断不阻塞，只告警


def test_resolve_step_time_priority():
    eq = [{"opStdTime": {"OP-X": 5.0}}]
    assert resolve_step_time({"operationCode": "OP-X", "stdTimePerUnit": 8.0}, eq) == \
        {"stdMin": 8.0, "source": "实测"}
    assert resolve_step_time({"operationCode": "OP-X"}, eq) == \
        {"stdMin": 5.0, "source": "设备默认"}
    assert resolve_step_time({"operationCode": "OP-Y"}, eq)["source"] == "待维护"


def test_time_matrix_has_sources():
    world = seed_world()
    rows = time_matrix(world)
    assert rows
    assert {"productCode", "operationCode", "stdMin", "source"} <= set(rows[0].keys())
    assert all(r["source"] in ("实测", "推断", "模板", "设备默认", "待维护") for r in rows)


def test_annotate_time_conflicts_after_schedule():
    world = seed_world()
    # 制造一个待维护步骤
    for e in world["flexEquipment"]:
        e.get("opStdTime", {}).pop("OP-PKG", None)
    counters: dict[str, int] = {}

    def nid(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    n = annotate_time_conflicts(world, version_id=99, next_id=nid, order_nos=None)
    assert n > 0
    cfs = [c for c in world["flexConflicts"] if c["versionId"] == 99]
    assert any(c["conflictType"] == "TIME_UNMAINTAINED" for c in cfs)
    assert all(c["severity"] in ("MAJOR", "MINOR") for c in cfs)


def test_analyze_project_data():
    from server.aps_domain.readiness import analyze_project_data
    world = seed_world()
    text = analyze_project_data(world, "分析项目数据")
    assert "项目数据概览" in text
    assert "待排订单" in text and "可直接排产" in text
    text2 = analyze_project_data(
        world,
        "【文件解析】\n文件：a.xlsx\n有效 2 行 · 错误 0 行（可入库）\n\n给我分析一下",
    )
    assert "附件解析结果" in text2
    assert "a.xlsx" in text2