112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
# ============================================================
|
|
# 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
|