aps-agent/tests/golden/test_kangni_flex.py

# ============================================================
# 康尼柔性场景黄金测试：交期三区间 / 三模式 / 故障 / 模具寿命
# ============================================================
from __future__ import annotations

from server.aps_domain.flex import (
    apply_equipment_fault, compare_sort_modes, insert_rush_order, simulate_due,
)
from server.engines import PoolEngine
from server.state.seed import seed_world


class _MemStore:
    def __init__(self, data):
        self.data = data

    def next_id(self, kind: str) -> int:
        key = f"_c_{kind}"
        self.data[key] = self.data.get(key, 100) + 1
        return self.data[key]

    def save(self):
        pass


def test_simulate_due_has_three_bands_and_gaps():
    store = _MemStore(seed_world())
    r = simulate_due(store, "PDU-UNIT", 500)
    assert r["optimisticFinish"]
    assert r["expectedFinish"]
    assert r["pessimisticFinish"]
    assert r["gaps"]
    assert any(g.get("kind") for g in r["gaps"])


def test_compare_modes_five_rows():
    import copy

    world = seed_world()
    eligible = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
    assert len(eligible) == 10
    before = copy.deepcopy(world)
    store = _MemStore(world)
    cmp = compare_sort_modes(store, compress_due=True)
    assert len(cmp["rows"]) == 5
    modes = {r["sortMode"] for r in cmp["rows"]}
    assert modes == {"ASC", "DESC", "BOTTLENECK", "KITTING_FIRST", "SKILL_FIRST"}
    labels = {r["label"] for r in cmp["rows"]}
    assert labels == {"正排", "倒排", "瓶颈锚", "齐套优先", "技能优先"}
    for row in cmp["rows"]:
        assert {"sortMode", "label", "totalTardiness", "onTimeCount", "orderCount",
                "makespan", "avgUtilization", "conflictCount", "vlCount"} <= set(row)
        assert row["orderCount"] == 10
    assert store.data == before

    quality = cmp["dataQuality"]
    assert quality["eligibleOrderCount"] == 10
    assert quality["kitStatusKnownCount"] == 0
    assert quality["skillLevelKnownCount"] == 0
    rows = {row["sortMode"]: row for row in cmp["rows"]}
    assert rows["KITTING_FIRST"]["strategyStatus"] == "DEGRADED"
    assert rows["KITTING_FIRST"]["strategyEvidence"][0]["code"] == "MISSING_KIT_STATUS"
    assert rows["SKILL_FIRST"]["strategyStatus"] == "DEGRADED"
    assert rows["SKILL_FIRST"]["strategyEvidence"][0]["code"] == "MISSING_SKILL_LEVEL"
    assert "当前数据证据不足" in cmp["hint"]


def test_capacity_forecast_horizons():
    from server.aps_domain.flex import capacity_analysis, run_flex_schedule
    store = _MemStore(seed_world())
    run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
    cap = capacity_analysis(store.data)
    assert cap["horizons"]
    assert {h["key"] for h in cap["horizons"]} == {"h4", "d1", "w1"}
    assert cap["pools"]
    p0 = cap["pools"][0]
    assert "forecast" in p0
    assert set(p0["forecast"].keys()) >= {"h4", "d1", "w1"}
    assert 0 <= p0["forecast"]["d1"]["utilization"]
    assert p0["forecast"]["d1"]["alert"] in ("ok", "warn", "crit")


def test_mold_life_consumed_and_can_lock():
    world = seed_world()
    mold = next(m for m in world["flexMolds"] if m["code"] == "MOLD-HV-01")
    before = mold["lifeUsed"]
    mold["lifeTotal"] = before + 50  # 排一单就会到限
    counters: dict[str, int] = {}

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

    PoolEngine().solve(world, nid, sort_mode="BOTTLENECK")
    assert mold["lifeUsed"] > before
    assert mold["status"] == "LOCKED" or mold["lifeUsed"] >= mold["lifeTotal"]


def test_fault_weld_can_raise_no_capability():
    store = _MemStore(seed_world())
    # 先正常排一轮，再焊机故障
    from server.aps_domain.flex import run_flex_schedule
    run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
    r = apply_equipment_fault(store, "WELD-01", status="MAINTENANCE", reschedule=True, actor="test")
    assert r["equipment"]["afterStatus"] == "MAINTENANCE"
    # 焊接单点：无备机 → L4 缩池重排
    assert r.get("responseLevel") == "L4"
    types = {c.get("conflictType") for c in r.get("conflicts") or []}
    assert "NO_CAPABILITY" in types or r["schedule"]["conflictCount"] >= 0


def test_fault_press_prefers_l1_swap():
    store = _MemStore(seed_world())
    from server.aps_domain.flex import run_flex_schedule
    run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
    before_ver = len(store.data["flexScheduleVersions"])
    r = apply_equipment_fault(store, "PRESS-01", status="MAINTENANCE", reschedule=True, actor="test")
    assert r["equipment"]["afterStatus"] == "MAINTENANCE"
    # 压接池有冗余：优先 L1，不新建版本
    if r.get("responseLevel") == "L1":
        assert r.get("swap", {}).get("swapped")
        assert r.get("schedule") is None
        assert len(store.data["flexScheduleVersions"]) == before_ver
        eq = next(e for e in store.data["flexEquipment"] if e["code"] == "PRESS-01")
        assert eq["status"] == "MAINTENANCE"
    else:
        # 极端拥挤时退化为 L4 亦可接受
        assert r.get("responseLevel") == "L4"
        assert r.get("schedule")


def test_rush_insert_order():
    store = _MemStore(seed_world())
    before = len(store.data["flexOrders"])
    r = insert_rush_order(store, "PDU-UNIT", 80, actor="test")
    assert len(store.data["flexOrders"]) == before + 1
    assert r["order"]["orderNo"].startswith("RUSH-")


def test_pool_engine_kitting_and_skill_sort_modes():
    import copy

    base = seed_world()
    orders = [o for o in base.get("flexOrders", []) if o["status"] in ("RELEASED", "CREATED")]
    assert orders
    for i, o in enumerate(orders):
        o["kitStatus"] = "未完成" if i % 2 else "已完成"
        o["requiredSkillLevel"] = "L2"
    orders[1]["requiredSkillLevel"] = "L4"

    w1 = copy.deepcopy(base)
    r1 = PoolEngine().solve(w1, _MemStore(w1).next_id, sort_mode="KITTING_FIRST")
    assert r1["orderCount"] == len(orders)
    ready = {o["orderNo"] for o in orders if o["kitStatus"] == "已完成"}
    assert w1["flexVirtualLines"][0]["orderNo"] in ready

    w2 = copy.deepcopy(base)
    r2 = PoolEngine().solve(w2, _MemStore(w2).next_id, sort_mode="SKILL_FIRST")
    assert r2["orderCount"] == len(orders)
    high = {o["orderNo"] for o in orders if o["requiredSkillLevel"] == "L4"}
    assert w2["flexVirtualLines"][0]["orderNo"] in high


def test_compare_modes_uses_kitting_and_skill_when_evidence_complete():
    world = seed_world()
    orders = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
    assert len(orders) == 10
    for i, order in enumerate(orders):
        order["kitStatus"] = "已完成" if i % 2 == 0 else "未完成"
        order["requiredSkillLevel"] = "L4" if i % 3 == 0 else "L2"

    cmp = compare_sort_modes(_MemStore(world), compress_due=False)
    rows = {row["sortMode"]: row for row in cmp["rows"]}
    assert rows["KITTING_FIRST"]["strategyStatus"] == "READY"
    assert rows["KITTING_FIRST"]["strategyEvidence"] == []
    assert rows["SKILL_FIRST"]["strategyStatus"] == "READY"
    assert rows["SKILL_FIRST"]["strategyEvidence"] == []


def test_compare_modes_returns_missing_capability_conflict_evidence():
    world = seed_world()
    for equipment in world["flexEquipment"]:
        equipment["capabilities"] = [
            code for code in equipment.get("capabilities", []) if code != "OP-WELD"
        ]

    cmp = compare_sort_modes(_MemStore(world), compress_due=False)
    assert "OP-WELD" in cmp["dataQuality"]["uncoveredOperationCodes"]
    assert any(w["code"] == "UNCOVERED_OPERATION_CAPABILITY" for w in cmp["warnings"])
    for row in cmp["rows"]:
        assert row["resultStatus"] == "PARTIAL"
        assert "NO_CAPABILITY" in row["conflictTypes"]
        assert any(c["conflictType"] == "NO_CAPABILITY" for c in row["conflictEvidence"])


def test_compare_modes_marks_shared_placeholder_capability_as_partial():
    world = seed_world()
    required_ops = sorted({
        routing["operationCode"] for routing in world["flexRoutings"]
        if routing.get("operationCode")
    })
    for equipment in world["flexEquipment"]:
        equipment["capabilities"] = []
    world["flexEquipment"][0]["capabilities"] = required_ops

    cmp = compare_sort_modes(_MemStore(world), compress_due=False)
    quality = cmp["dataQuality"]
    assert quality["capabilityMappedEquipmentCount"] == 1
    assert quality["unmappedEquipmentCount"] == quality["equipmentCount"] - 1
    assert quality["uncoveredOperationCodes"] == []
    assert quality["status"] == "PARTIAL"
    assert any(w["code"] == "PARTIAL_EQUIPMENT_CAPABILITY" for w in cmp["warnings"])


def test_compare_modes_preserves_kangni_resource_gap_counts():
    import copy

    world = seed_world()
    required_ops = sorted({
        routing["operationCode"] for routing in world["flexRoutings"]
        if routing.get("operationCode")
    })
    shared_capabilities = required_ops + [
        f"KNI-OP-{i:02d}" for i in range(1, 27 - len(required_ops))
    ]
    shared = copy.deepcopy(world["flexEquipment"][0])
    shared.update({
        "id": 1000,
        "code": "EQ-SHARED-CG-01",
        "name": "共享占位设备",
        "capabilities": shared_capabilities,
        "opStdTime": {code: 1 for code in shared_capabilities},
        "adaptableMolds": [],
        "status": "RUNNING",
    })
    world["flexEquipment"] = [shared] + [{
        "id": 1000 + i,
        "code": f"EQ-KNI-{i:02d}",
        "name": f"待映射设备{i:02d}",
        "capabilities": [],
        "opStdTime": {},
        "adaptableMolds": [],
        "availabilityRate": 1,
        "status": "RUNNING",
    } for i in range(1, 41)]
    world["flexMolds"] = [{
        "id": i,
        "code": f"MOLD-KNI-{i:02d}",
        "name": f"待映射模具{i:02d}",
        "operationCode": None,
        "adaptableEquipment": [],
        "lifeTotal": None,
        "lifeUsed": 0,
        "status": "AVAILABLE",
    } for i in range(1, 41)]

    cmp = compare_sort_modes(_MemStore(world), compress_due=False)
    quality = cmp["dataQuality"]
    assert quality["equipmentCount"] == 41
    assert quality["capabilityMappedEquipmentCount"] == 1
    assert quality["capabilityMappedEquipmentCodes"] == ["EQ-SHARED-CG-01"]
    assert quality["unmappedEquipmentCount"] == 40
    assert quality["moldCount"] == 40
    assert quality["fullyMappedMoldCount"] == 0
    assert quality["status"] == "PARTIAL"
    assert quality["productionReady"] is False
    warning_codes = {warning["code"] for warning in cmp["warnings"]}
    assert {"PARTIAL_EQUIPMENT_CAPABILITY", "INCOMPLETE_MOLD_MAPPING"} <= warning_codes
    assert "设备能力仅覆盖 1/41 台设备" in cmp["hint"]
    assert "模具完整映射仅覆盖 0/40 套" in cmp["hint"]


def test_compare_modes_treats_unknown_kit_and_uniform_skill_as_degraded_evidence():
    world = seed_world()
    orders = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
    assert len(orders) == 10
    for order in orders:
        order["kitStatus"] = "未知"
        order["requiredSkillLevel"] = "L3"
    orders[0]["kitStatus"] = "已完成"

    cmp = compare_sort_modes(_MemStore(world), compress_due=False)
    quality = cmp["dataQuality"]
    assert quality["kitStatusKnownCount"] == 1
    assert quality["kitStatusDistinctCount"] == 1
    assert quality["skillLevelKnownCount"] == 10
    assert quality["skillLevelDistinctCount"] == 1
    rows = {row["sortMode"]: row for row in cmp["rows"]}
    assert rows["KITTING_FIRST"]["strategyStatus"] == "PARTIAL"
    assert rows["KITTING_FIRST"]["strategyEvidence"][0]["code"] == "PARTIAL_KIT_STATUS"
    assert rows["SKILL_FIRST"]["strategyStatus"] == "DEGRADED"
    assert rows["SKILL_FIRST"]["strategyEvidence"][0]["code"] == "NO_SKILL_LEVEL_VARIATION"
    warning_codes = {warning["code"] for warning in cmp["warnings"]}
    assert {"MISSING_KIT_STATUS", "NO_SKILL_LEVEL_VARIATION"} <= warning_codes