72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
|
# ============================================================
|
||
|
|
# 工序供应类型 + 多层 BOM 分解黄金测试
|
||
|
|
# ============================================================
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from server.aps_domain.mrp import decompose_orders
|
||
|
|
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"]
|