174 lines
6.9 KiB
Python
174 lines
6.9 KiB
Python
# ============================================================
|
|
# 工序供应类型 + 多层 BOM 分解黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.aps_domain.mrp import decompose_orders, list_mrp
|
|
from server.aps_domain.sourcing import annotate_world_sourcing, build_process_tree, infer_op_sourcing
|
|
from server.state.seed import seed_world
|
|
|
|
|
|
def _next_id_factory(world):
|
|
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_wz_code_is_outsource():
|
|
assert infer_op_sourcing("WZ0006", "调质") == "OUTSOURCE"
|
|
assert infer_op_sourcing("NZ0027", "精车") == "MAKE"
|
|
|
|
|
|
def test_annotate_marks_wz_steps_external():
|
|
world = seed_world()
|
|
# 给已有步骤挂上 WZ 工序
|
|
world["operations"].append({
|
|
"id": 9001, "code": "WZ0006", "name": "调质", "type": "INTERNAL", "standardTime": 1,
|
|
})
|
|
world["routingSteps"].append({
|
|
"id": 9001, "routingId": 1, "operationId": 9001,
|
|
"sequenceNo": 99, "prevStepId": None,
|
|
"setupTime": 10, "runTimePerUnit": 1, "waitTime": 0, "transferTime": 0,
|
|
"isExternal": False,
|
|
})
|
|
annotate_world_sourcing(world)
|
|
step = next(s for s in world["routingSteps"] if s["id"] == 9001)
|
|
assert step["isExternal"] is True
|
|
op = next(o for o in world["operations"] if o["id"] == 9001)
|
|
assert op["type"] == "EXTERNAL"
|
|
|
|
|
|
def test_decompose_generates_outsource_from_wz():
|
|
world = seed_world()
|
|
world["operations"].append({
|
|
"id": 9001, "code": "WZ0006", "name": "调质", "type": "INTERNAL", "standardTime": 1,
|
|
})
|
|
world["routingSteps"].append({
|
|
"id": 9001, "routingId": 1, "operationId": 9001,
|
|
"sequenceNo": 99, "prevStepId": None,
|
|
"setupTime": 10, "runTimePerUnit": 1, "waitTime": 0, "transferTime": 0,
|
|
"isExternal": False,
|
|
})
|
|
result = decompose_orders(world, _next_id_factory(world))
|
|
outs = [o for o in result["outsource"] if o.get("operationCode") == "WZ0006" or "调质" in o.get("operationName", "")]
|
|
assert outs, "WZ 调质工序应分解出委外建议"
|
|
|
|
|
|
def test_process_tree_has_material_and_ops():
|
|
world = seed_world()
|
|
# 演示厂 CTRL-A / 产品 1
|
|
mat = next(m for m in world["materials"] if m["type"] == "FINISHED_PRODUCT")
|
|
tree = build_process_tree(world, mat["code"])
|
|
assert tree["kind"] == "material"
|
|
assert tree["code"] == mat["code"]
|
|
|
|
|
|
def test_multilevel_bom_semifinished_without_routing_stays_make_and_list_mrp_preserves_it():
|
|
"""A BOM-bearing semi-finished item stays MAKE unless explicit BUY overrides it."""
|
|
finished_id = 1001
|
|
semi_id = 1002
|
|
raw_id = 1003
|
|
explicit_buy_semi_id = 1004
|
|
buy_child_id = 1005
|
|
|
|
world = seed_world()
|
|
world.update({
|
|
"materials": [
|
|
{"id": finished_id, "code": "FG-MULTI", "name": "Multi-level finished good", "type": "FINISHED_PRODUCT", "unit": "PCS"},
|
|
{
|
|
"id": semi_id,
|
|
"code": "SFG-MAKE",
|
|
"name": "Semi-finished without routing",
|
|
"type": "SEMI_FINISHED",
|
|
"unit": "PCS",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
},
|
|
{
|
|
"id": raw_id,
|
|
"code": "RM-BUY",
|
|
"name": "Raw child",
|
|
"type": "RAW_MATERIAL",
|
|
"unit": "KG",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
"procurementLeadTime": 0,
|
|
},
|
|
{
|
|
"id": explicit_buy_semi_id,
|
|
"code": "SFG-EXPLICIT-BUY",
|
|
"name": "Explicitly purchased semi-finished item",
|
|
"type": "SEMI_FINISHED",
|
|
"procurementType": "BUY",
|
|
"unit": "PCS",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
"procurementLeadTime": 0,
|
|
},
|
|
{
|
|
"id": buy_child_id,
|
|
"code": "RM-BUY-CHILD",
|
|
"name": "Child of explicitly purchased semi-finished item",
|
|
"type": "RAW_MATERIAL",
|
|
"unit": "KG",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
"procurementLeadTime": 0,
|
|
},
|
|
],
|
|
"boms": [
|
|
{"id": 2001, "productId": finished_id, "isDefault": True},
|
|
{"id": 2002, "productId": semi_id, "isDefault": True},
|
|
{"id": 2003, "productId": explicit_buy_semi_id, "isDefault": True},
|
|
],
|
|
"bomItems": [
|
|
{"id": 3001, "bomId": 2001, "materialId": semi_id, "quantity": 2, "isKeyMaterial": False},
|
|
{"id": 3002, "bomId": 2001, "materialId": explicit_buy_semi_id, "quantity": 1, "isKeyMaterial": False},
|
|
{"id": 3003, "bomId": 2002, "materialId": raw_id, "quantity": 3, "isKeyMaterial": False},
|
|
{"id": 3004, "bomId": 2003, "materialId": buy_child_id, "quantity": 7, "isKeyMaterial": False},
|
|
],
|
|
"salesOrders": [{
|
|
"id": 4001,
|
|
"orderNo": "SO-MULTI-001",
|
|
"status": "CONFIRMED",
|
|
"deliveryDate": "2026-08-10",
|
|
"items": [{
|
|
"id": 5001,
|
|
"productId": finished_id,
|
|
"productName": "Multi-level finished good",
|
|
"quantity": 2,
|
|
"unit": "PCS",
|
|
"status": "CONFIRMED",
|
|
}],
|
|
}],
|
|
"routings": [],
|
|
"routingSteps": [],
|
|
"operations": [],
|
|
"purchaseOrders": [],
|
|
"outsourceOrders": [],
|
|
"flexBom": [],
|
|
"flexOrders": [],
|
|
})
|
|
|
|
result = decompose_orders(world, _next_id_factory(world), "SO-MULTI-001")
|
|
make_ids = {row["productId"] for row in result["make"]}
|
|
purchase_ids = {row["materialId"] for row in result["purchase"]}
|
|
|
|
assert semi_id in make_ids, "SEMI_FINISHED with its own BOM must be MAKE even without routing"
|
|
assert raw_id in purchase_ids, "MAKE semi-finished item must be exploded so its raw child is purchased"
|
|
assert semi_id not in purchase_ids, "Structurally MAKE semi-finished item must never be purchased"
|
|
assert explicit_buy_semi_id in purchase_ids, "procurementType=BUY must override the item's own BOM structure"
|
|
assert explicit_buy_semi_id not in make_ids, "Explicit BUY semi-finished item must not also appear in MAKE"
|
|
assert buy_child_id not in purchase_ids, "Explicit BUY item is purchased as a whole and its BOM is not exploded"
|
|
|
|
listed_make_ids = {row["productId"] for row in list_mrp(world)["make"]}
|
|
assert semi_id in listed_make_ids, "list_mrp must preserve the semi-finished MAKE suggestion from decompose_orders"
|