aps-agent/tests/golden/test_rush.py

# ============================================================
# OR-04 紧急插单专用流黄金测试
# 固化：快评不碰主干；采用后生成 DRAFT 版本且可回滚至采用前
# ============================================================
from __future__ import annotations

import copy

from server.aps_domain.rush import apply_rush, evaluate_rush
from server.state.seed import seed_world
from server.timeutil import add_minutes, fmt_date, today0


def _next_id_factory(world):
    tables = {
        "audit": "auditEvents",
        "conflict": "conflicts",
        "log": "logs",
        "productionOrder": "productionOrders",
        "salesOrder": "salesOrders",
        "scheduleVersion": "scheduleVersions",
        "workOrder": "workOrders",
    }
    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        if kind not in counters:
            rows = world.get(tables.get(kind, kind + "s"), [])
            counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
        counters[kind] += 1
        return counters[kind]

    return next_id


def test_rush_evaluate_does_not_mutate_trunk():
    """插单快评全程沙盒：主干订单数/急单标记不变。"""
    world = seed_world()
    before = copy.deepcopy(world)
    n_orders = len(world["salesOrders"])
    impact = evaluate_rush(world, {
        "customerName": "快评客户",
        "customerLevel": "VIP",
        "productId": 1,
        "quantity": 200,
        "deliveryDate": fmt_date(add_minutes(today0(), 4 * 24 * 60)),
        "priority": 1,
        "strategy": "DELIVERY_FIRST",
    })
    assert impact["affectedOrderCount"] >= 0
    assert "delayDelta" in impact and "conflictDelta" in impact
    assert impact["after"]["orderCount"] == impact["baseline"]["orderCount"] + 1
    assert len(world["salesOrders"]) == n_orders
    assert world["salesOrders"] == before["salesOrders"]
    assert world["productionOrders"] == before["productionOrders"]
    assert world["scheduleVersions"] == before["scheduleVersions"]


def test_rush_apply_creates_draft_and_keeps_baseline_rollback_point():
    """采用插单：主干新增急单 + DRAFT 版本；采用前快照可还原。"""
    world = seed_world()
    baseline_versions = len(world["scheduleVersions"])
    snapshot = copy.deepcopy(world)

    impact = evaluate_rush(world, {
        "customerName": "采用客户",
        "customerLevel": "VIP",
        "productCode": "CTRL-A",
        "quantity": 150,
        "deliveryDate": fmt_date(add_minutes(today0(), 5 * 24 * 60)),
        "strategy": "DELIVERY_FIRST",
    })
    # 快评后主干仍干净
    assert len(world["salesOrders"]) == len(snapshot["salesOrders"])

    applied = apply_rush(world, _next_id_factory(world), {
        "payload": impact["payload"],
        "strategy": impact["strategy"],
        "evalId": impact["evalId"],
    })
    order = applied["order"]
    result = applied["result"]

    assert order["isRush"] is True
    assert order["status"] == "APPROVED"
    assert any(so["id"] == order["id"] for so in world["salesOrders"])
    assert len(world["scheduleVersions"]) == baseline_versions + 1
    assert world["scheduleVersions"][-1]["status"] == "DRAFT"
    assert result.versionNo == world["scheduleVersions"][-1]["versionNo"]
    assert result.orderCount == 8  # 种子 7 + 急单 1

    # 回滚到采用前快照（模拟检查点恢复）
    world.clear()
    world.update(copy.deepcopy(snapshot))
    assert len(world["salesOrders"]) == 7
    assert len(world["scheduleVersions"]) == baseline_versions
    assert all(not so.get("orderNo", "").endswith(str(order["id"])) or so["id"] != order["id"]
               for so in world["salesOrders"])


def test_rush_evaluate_existing_order():
    """对已有订单加急：受影响指标来自沙盒差分。"""
    world = seed_world()
    target = world["salesOrders"][3]  # 非急单
    assert not target.get("isRush")
    impact = evaluate_rush(world, {"orderNo": target["orderNo"], "strategy": "COMPREHENSIVE"})
    assert impact["mode"] == "existing"
    assert impact["rushOrder"]["orderNo"] == target["orderNo"]
    assert not target.get("isRush")  # 主干未改
    assert impact["baseline"]["orderCount"] == 7
    assert impact["after"]["orderCount"] == 7


def _trial_flex_world() -> dict:
    """试排世界夹具：主干种子数据 + planningContext.trialOnly + 最小柔性主数据。

    模拟「文件夹快试导入」后的真实形态：排产只认 flex* 主数据，
    主干 RULE 引擎在此类世界上只能产出 0 工单的空壳版本。
    """
    world = seed_world()
    world["planningContext"] = {"trialOnly": True, "dataDate": "2026-08-04"}
    world["flexOperations"] = [{"id": 1, "code": "CUT", "name": "下料", "changeoverMin": 10}]
    world["flexTeams"] = [{
        "id": 1, "code": "TEAM-CUT", "name": "下料班", "memberCount": 4,
        "supportOps": ["CUT"],
    }]
    world["flexRoutings"] = [{
        "id": 1, "productCode": "CTRL-A", "productName": "控制柜 A", "seq": 1,
        "operationCode": "CUT", "operationName": "下料", "stdTimePerUnit": 5.0,
    }]
    world["flexMaterials"] = [{
        "id": 1, "code": "CTRL-A", "name": "控制柜 A", "type": "FINISHED_PRODUCT",
        "unit": "件", "stock": 0, "inTransit": 0, "safetyStock": 0,
        "procurementLeadTime": 0, "sourcingType": "MAKE",
    }]
    world["flexEquipment"] = [{
        "id": 1, "code": "EQ-1", "name": "切割机", "status": "RUNNING",
        "capabilities": ["CUT"], "availabilityRate": 1.0,
    }]
    return world


def test_rush_apply_in_trial_world_routes_to_flex_engine():
    """试排世界采用急单必须走柔性引擎（回归：急单入池却排不进/看不见）。

    锁定三点：急单桥接进 flexOrders（source=SALES/RELEASED）；生成 FV 柔性
    草稿版本且急单真实排出虚拟产线；主轨 scheduleVersions 不再新增空壳版本。
    """
    world = _trial_flex_world()
    n_trunk_versions = len(world["scheduleVersions"])
    n_flex_versions = len(world.get("flexScheduleVersions") or [])

    applied = apply_rush(world, _next_id_factory(world), {
        "payload": {
            "customerName": "试排急单客户",
            "customerLevel": "VIP",
            "productCode": "CTRL-A",
            "quantity": 3,
            "deliveryDate": fmt_date(add_minutes(today0(), 5 * 24 * 60)),
            "priority": 1,
        },
        "strategy": "DELIVERY_FIRST",
    })
    order = applied["order"]
    result = applied["result"]

    assert order["isRush"] is True
    assert order["status"] == "APPROVED"

    # 急单已桥接进柔性订单池
    flex_row = next(
        (o for o in world["flexOrders"] if o.get("orderNo") == order["orderNo"]),
        None,
    )
    assert flex_row is not None, "急单未桥接进 flexOrders，柔性排产必然看不到它"
    assert flex_row["source"] == "SALES"
    assert flex_row["status"] == "RELEASED"
    assert flex_row["isRush"] is True

    # 生成的是柔性 FV 草稿版本，而不是主轨 RULE 空壳版本
    assert str(result.versionNo).startswith("FV")
    assert len(world["scheduleVersions"]) == n_trunk_versions, "主轨不应新增空壳版本"
    assert len(world["flexScheduleVersions"]) == n_flex_versions + 1
    assert world["flexScheduleVersions"][-1]["versionNo"] == result.versionNo
    assert result.orderCount == len(world["flexOrders"])

    # 急单真实排出虚拟产线（修复前在此类世界为 0 工单）
    rush_vls = [
        v for v in world["flexVirtualLines"]
        if v.get("orderNo") == order["orderNo"]
        and v.get("versionId") == world["flexScheduleVersions"][-1]["id"]
    ]
    assert rush_vls, "急单在柔性版本中应有至少一条虚拟产线"