152 lines
5.2 KiB
Python
152 lines
5.2 KiB
Python
# ============================================================
|
||
# OR-03 订单池 / 审核黄金测试
|
||
# 固化：未批准不进排产；submit → approve 后可排；编辑已批→CHANGED
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.aps_domain.orders import (
|
||
apply_order_action,
|
||
apply_order_submit,
|
||
is_schedulable,
|
||
pool_summary,
|
||
)
|
||
from server.engines import get_engine
|
||
from server.engines.base import EngineParams
|
||
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 _run(world, *, include_unapproved: bool = False):
|
||
params = EngineParams(
|
||
orderIds=[],
|
||
engineType="RULE",
|
||
strategyTemplate="COMPREHENSIVE",
|
||
planningHorizonDays=14,
|
||
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
||
includeUnapproved=include_unapproved,
|
||
)
|
||
return get_engine("RULE").solve(world, params, _next_id_factory(world))
|
||
|
||
|
||
def test_draft_excluded_from_schedule_run():
|
||
"""DRAFT 订单默认不进入 schedule.run；includeUnapproved 时可试排。"""
|
||
world = seed_world()
|
||
applied = apply_order_action(world, _next_id_factory(world), "order.upsert", {
|
||
"customerName": "待审客户",
|
||
"customerLevel": "A",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 9 * 24 * 60)),
|
||
"priority": 2,
|
||
"status": "DRAFT",
|
||
"productId": 1,
|
||
"quantity": 50,
|
||
})
|
||
draft = applied["order"]
|
||
assert draft["status"] == "DRAFT"
|
||
assert not is_schedulable(draft["status"])
|
||
|
||
result = _run(world)
|
||
assert result.orderCount == 7
|
||
assert all(po["salesOrderId"] != draft["id"] for po in world["productionOrders"])
|
||
|
||
result2 = _run(world, include_unapproved=True)
|
||
assert result2.orderCount == 8
|
||
assert any(po["salesOrderId"] == draft["id"] for po in world["productionOrders"])
|
||
|
||
|
||
def test_submit_approve_enters_scheduling():
|
||
"""DRAFT → submit → approve 后纳入正式排产。"""
|
||
world = seed_world()
|
||
applied = apply_order_action(world, _next_id_factory(world), "order.upsert", {
|
||
"customerName": "审核流客户",
|
||
"customerLevel": "VIP",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 10 * 24 * 60)),
|
||
"priority": 1,
|
||
"status": "DRAFT",
|
||
"productId": 1,
|
||
"quantity": 80,
|
||
})
|
||
order = applied["order"]
|
||
|
||
submitted = apply_order_submit(world, {"id": order["id"]})
|
||
assert submitted["order"]["status"] == "SUBMITTED"
|
||
assert _run(world).orderCount == 7
|
||
|
||
approved = apply_order_action(world, _next_id_factory(world), "order.approve", {
|
||
"orderIds": [order["id"]],
|
||
})
|
||
assert approved["order"]["status"] == "APPROVED"
|
||
assert approved["approvedCount"] == 1
|
||
|
||
result = _run(world)
|
||
assert result.orderCount == 8
|
||
assert any(po["salesOrderId"] == order["id"] for po in world["productionOrders"])
|
||
|
||
|
||
def test_edit_approved_becomes_changed():
|
||
"""编辑已批准订单内容 → CHANGED，暂时退出可排池。"""
|
||
world = seed_world()
|
||
target = world["salesOrders"][0]
|
||
assert target["status"] == "APPROVED"
|
||
|
||
apply_order_action(world, _next_id_factory(world), "order.upsert", {
|
||
"id": target["id"],
|
||
"customerName": target["customerName"],
|
||
"customerLevel": target["customerLevel"],
|
||
"deliveryDate": target["deliveryDate"],
|
||
"priority": target["priority"],
|
||
"productId": target["items"][0]["productId"],
|
||
"quantity": target["items"][0]["quantity"] + 10,
|
||
"status": "APPROVED",
|
||
})
|
||
assert target["status"] == "CHANGED"
|
||
assert not is_schedulable(target["status"])
|
||
assert pool_summary(world)["changed"] >= 1
|
||
assert _run(world).orderCount == 6
|
||
|
||
|
||
def test_batch_approve_pending():
|
||
"""orderIds=pending 批量批准所有 SUBMITTED。"""
|
||
world = seed_world()
|
||
for name in ("批甲", "批乙"):
|
||
o = apply_order_action(world, _next_id_factory(world), "order.upsert", {
|
||
"customerName": name,
|
||
"customerLevel": "B",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 11 * 24 * 60)),
|
||
"priority": 5,
|
||
"status": "DRAFT",
|
||
"productId": 2,
|
||
"quantity": 40,
|
||
})["order"]
|
||
apply_order_submit(world, {"id": o["id"]})
|
||
|
||
sm = pool_summary(world)
|
||
assert sm["pending"] == 2
|
||
applied = apply_order_action(world, _next_id_factory(world), "order.approve", {
|
||
"orderIds": "pending",
|
||
})
|
||
assert applied["approvedCount"] == 2
|
||
assert pool_summary(world)["pending"] == 0
|
||
assert _run(world).orderCount == 9
|