aps-agent/tests/golden/test_mrp.py

# ============================================================
# MRP 订单分解黄金测试（moduleId: golden-mrp, 发版门禁 §14.3）
# 固化「订单 → BOM/工艺分解 → 采购/委外建议」最小闭环：
#   R1 缺料物料生成采购建议，且建议下单日按前置期倒排
#   R2 库存充足的物料不生成采购建议
#   R3 外协工艺步骤生成委外建议；取消外协标记后建议消失（幂等重分解）
#   R4 指定订单分解只影响该订单的建议；订单号不存在报错
#   R5 BOM 用量编辑（master.bom.upsert）联动采购建议数量
# ============================================================
from __future__ import annotations

import copy

import pytest

from server.aps_domain.masterdata import apply_master_action
from server.aps_domain.mrp import decompose_orders, list_mrp
from server.state.seed import seed_world
from server.timeutil import add_minutes, fmt_date, parse_dt, today0


def _next_id_factory(world):
    """按当前世界最大 ID 起号。"""
    tables = {
        "purchaseOrder": "purchaseOrders",
        "outsourceOrder": "outsourceOrders",
    }
    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_shortage_generates_purchase_with_leadtime_backdating():
    """R1+R2：缺料生成采购建议且下单日=需求日-前置期-缓冲；充足物料零建议。"""
    world = seed_world()
    next_id = _next_id_factory(world)
    # 把 PCB-001（多数产品共用）调为零库存零在途 → 必缺料
    apply_master_action(world, next_id, "master.material.upsert",
                        {"id": 10, "stock": 0, "inTransit": 0, "safetyStock": 500, "procurementLeadTime": 7})
    result = decompose_orders(world, next_id)

    pcb = [p for p in result["purchase"] if p["materialId"] == 10]
    assert pcb, "零库存物料必须生成采购建议"
    row = pcb[0]
    # 建议下单日 = 需求日(交期) - 前置期7天 - 缓冲1天
    expected = fmt_date(add_minutes(parse_dt(row["requiredDate"] + " 08:00"), -8 * 24 * 60))
    assert row["suggestedOrderDate"] == expected
    assert row["status"] == "DRAFT"
    # 库存极充足的物料（标签 LABEL-001 id=15，库存 2500）不应出现在建议中
    assert not any(p["materialId"] == 15 for p in result["purchase"])


def test_external_step_roundtrip_outsource_suggestion():
    """R3：勾外协 → 出委外建议；取消外协 → 重分解后建议消失（DRAFT 幂等替换）。"""
    world = seed_world()
    next_id = _next_id_factory(world)
    # 把产品 1 工艺的第 2 步（DIP插件，stepId=2）标记外协
    apply_master_action(world, next_id, "master.routing.upsert",
                        {"stepId": 2, "isExternal": True})
    result = decompose_orders(world, next_id)
    outs = [o for o in result["outsource"] if o["productId"] == 1]
    assert outs, "外协步骤应生成委外建议"
    assert all(o["operationName"] == "DIP插件" for o in outs)

    # 取消外协后重分解：旧 DRAFT 建议被替换清空
    apply_master_action(world, next_id, "master.routing.upsert",
                        {"stepId": 2, "isExternal": False})
    decompose_orders(world, next_id)
    remaining = [o for o in list_mrp(world)["outsourceOrders"] if o["productId"] == 1]
    assert not remaining, "取消外协并重分解后不应残留 DRAFT 委外建议"


def test_single_order_scope_and_unknown_order():
    """R4：指定订单分解只清/生成该订单的建议；未知订单号报错。"""
    world = seed_world()
    next_id = _next_id_factory(world)
    apply_master_action(world, next_id, "master.material.upsert",
                        {"id": 10, "stock": 0, "inTransit": 0, "safetyStock": 500, "procurementLeadTime": 7})
    decompose_orders(world, next_id)                   # 全量分解
    all_count = len(list_mrp(world)["purchaseOrders"])
    assert all_count > 1

    target_no = world["salesOrders"][0]["orderNo"]
    result = decompose_orders(world, next_id, target_no)   # 只重分解第一张
    assert result["orders"] == [target_no]
    # 其它订单的建议未被清掉
    others = [p for p in list_mrp(world)["purchaseOrders"] if p["salesOrderNo"] != target_no]
    assert others, "指定订单分解不得清除其它订单的建议"

    with pytest.raises(ValueError):
        decompose_orders(world, next_id, "SO99999999999")


def test_bom_quantity_edit_changes_purchase_quantity():
    """R5：BOM 用量翻倍后，同一物料的采购建议数量增加（主数据→MRP 联动）。"""
    world = seed_world()
    next_id = _next_id_factory(world)
    apply_master_action(world, next_id, "master.material.upsert",
                        {"id": 10, "stock": 0, "inTransit": 0, "safetyStock": 500, "procurementLeadTime": 7})
    r1 = decompose_orders(world, next_id)
    qty_before = sum(p["quantity"] for p in r1["purchase"] if p["materialId"] == 10)

    # BOM 明细 1：产品 1 用 PCB-001，用量 1 → 2
    apply_master_action(world, next_id, "master.bom.upsert", {"itemId": 1, "quantity": 2})
    r2 = decompose_orders(world, next_id)
    qty_after = sum(p["quantity"] for p in r2["purchase"] if p["materialId"] == 10)

    assert qty_after > qty_before, "BOM 用量增加后采购建议数量应增加"

def test_decompose_persists_make_suggestions_without_faking_production_orders():
    """?????????????????????????????"""
    world = seed_world()
    next_id = _next_id_factory(world)
    production_before = copy.deepcopy(world.get("productionOrders") or [])

    result = decompose_orders(world, next_id)
    projection = list_mrp(world)

    assert result["make"]
    assert projection["make"] == result["make"]
    assert world["makeSuggestions"] == result["make"]
    assert world.get("productionOrders") == production_before
    assert set(projection) >= {
        "make", "purchaseOrders", "outsourceOrders", "productionOrders",
    }