252 lines
12 KiB
Python
252 lines
12 KiB
Python
# ============================================================
|
||
# round-38 方向 N：WMS 缺料事件闭环黄金测试
|
||
# 固化：幂等去重 / 乱序不重复触发 / 库存版本进证据链 / 闭环全链
|
||
# 事件 → 影响半径 → Explore 方案卡 → 人工确认 → MES 下发 → 回执
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from server.agent_core import harness
|
||
from server.aps_domain import wms_events
|
||
from server.aps_domain.workflow import execute_confirmed, stage_schedule_publish
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
from server.integrations.mes_stub import reset_mes_client
|
||
from server.integrations.wms_stub import reset_wms_client
|
||
from server.state.checkpoints import CheckpointStore
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
class _MemStore:
|
||
def __init__(self, data, checkpoint_path: Path):
|
||
self.data = data
|
||
self.checkpoints = CheckpointStore(str(checkpoint_path))
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = f"_c_{kind}"
|
||
self.data[key] = self.data.get(key, 4000) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def _baseline(store):
|
||
"""先跑一版柔性排产作为基线（缺料事件前的已发布/DRAFT 版本）。"""
|
||
from server.aps_domain.flex import run_flex_schedule
|
||
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
||
return store.data["flexScheduleVersions"][-1]
|
||
|
||
|
||
def _seed_ledger(client):
|
||
client.upsert_inventory([
|
||
{"materialCode": "WIRE-HV", "name": "高压线材", "unit": "米",
|
||
"stock": 12000, "inTransit": 0, "safetyStock": 1000},
|
||
{"materialCode": "TERM-HV", "name": "高压端子", "unit": "个",
|
||
"stock": 6000, "inTransit": 2000, "safetyStock": 500},
|
||
{"materialCode": "BUSBAR", "name": "铜排", "unit": "根",
|
||
"stock": 1500, "inTransit": 400, "safetyStock": 200},
|
||
])
|
||
|
||
|
||
def _shortage_event(client, material_code: str, new_stock: float, shortage_qty: float,
|
||
occurred_at: str):
|
||
return client.emit_shortage(material_code, new_stock=new_stock,
|
||
shortage_qty=shortage_qty, occurred_at=occurred_at)
|
||
|
||
|
||
def _approve_p3(confirm_id: str) -> str:
|
||
"""P3 双人职责分离确认：默认身份第一重 + approver-2 第二重 → executionGrant。"""
|
||
first = harness.take_confirmation(confirm_id, approve=True)
|
||
assert first and first["needsSecondConfirm"] is True
|
||
token = bind_identity(IdentityContext(
|
||
2002, "approver-2", "Approver 2", "platform", roles=("planner",),
|
||
))
|
||
try:
|
||
second = harness.take_confirmation(confirm_id, approve=True)
|
||
finally:
|
||
reset_identity(token)
|
||
assert second and second.get("executionGrant")
|
||
return str(second["executionGrant"])
|
||
|
||
|
||
# ------------------------------------------------------------
|
||
# ① 幂等去重 + 乱序容忍：不重复触发
|
||
# ------------------------------------------------------------
|
||
def test_consume_dedup_and_out_of_order_no_retrigger(tmp_path: Path):
|
||
reset_wms_client(tmp_path / "wms_mirror.json")
|
||
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
||
_baseline(store)
|
||
client = reset_wms_client(tmp_path / "wms_mirror.json")
|
||
_seed_ledger(client)
|
||
ev1 = _shortage_event(client, "WIRE-HV", 500, 220, "2026-08-02 09:00:00")
|
||
ev2 = _shortage_event(client, "TERM-HV", 3000, 0, "2026-08-02 10:00:00")
|
||
ev3 = _shortage_event(client, "BUSBAR", 200, 180, "2026-08-02 11:00:00")
|
||
|
||
# 乱序投递：seq3 先到 → 缓冲不消费；seq1 到 → 排空 1；seq2 到 → 排空 2、3
|
||
r3 = wms_events.consume_event(store, ev3, actor="test")
|
||
assert r3["consumed"] is False and r3["duplicate"] is False
|
||
assert store.data["inventoryVersion"] == 0 # 未排空 → 版本不动
|
||
r1 = wms_events.consume_event(store, ev1, actor="test")
|
||
assert r1["consumed"] is True
|
||
assert [a["inventoryVersion"] for a in r1["applied"]] == [1]
|
||
r2 = wms_events.consume_event(store, ev2, actor="test")
|
||
assert r2["consumed"] is True
|
||
assert [a["inventoryVersion"] for a in r2["applied"]] == [2, 3]
|
||
assert store.data["inventoryVersion"] == 3
|
||
|
||
# 重复投递 → eventId 幂等，不重复消费/触发
|
||
dup = wms_events.consume_event(store, ev1, actor="test")
|
||
assert dup["duplicate"] is True and dup["consumed"] is False
|
||
assert len(store.data["wmsConsumed"]) == 3
|
||
|
||
# 证据链：每次消费唯一 inventory-version 且带 wms-event 引用
|
||
refs = [e["evidenceRefs"] for e in store.data["auditEvents"]
|
||
if e["action"] == "wms.event.consume"]
|
||
assert len(refs) == 3
|
||
versions = sorted({r[1] for r in refs if len(r) >= 2 and r[1].startswith("inventory-version:")})
|
||
assert versions == ["inventory-version:1", "inventory-version:2", "inventory-version:3"]
|
||
by_ev = {ev["eventId"]: refs[i] for i, ev in enumerate((ev1, ev2, ev3))}
|
||
assert all(f"wms-event:{eid}" in by_ev[eid] for eid in by_ev)
|
||
|
||
# 触发护栏：同一缺料事件只出一张方案卡；重复/重放不重复触发
|
||
s1 = wms_events.stage_shortage_solution(store, ev1, session_id="s1", actor="test")
|
||
assert s1["staged"] is True and s1["confirmId"]
|
||
s2 = wms_events.stage_shortage_solution(store, ev1, session_id="s1", actor="test")
|
||
assert s2["staged"] is False and s2["reason"] == "already-triggered"
|
||
assert s2["confirmId"] == s1["confirmId"]
|
||
replay_dup = wms_events.consume_shortage_event(store, ev1, session_id="s1", actor="test")
|
||
assert replay_dup["duplicate"] is True
|
||
cards = [e for e in store.data["auditEvents"] if e["action"] == "wms.event.stage"]
|
||
assert len(cards) == 1 # 只出一张方案卡
|
||
|
||
|
||
# ------------------------------------------------------------
|
||
# ③ + ④ 闭环全链：事件 → 方案卡 → 确认 → MES 下发 → 回执
|
||
# ------------------------------------------------------------
|
||
def test_shortage_closed_loop_event_to_mes_receipt(tmp_path: Path):
|
||
reset_mes_client(tmp_path / "mes_mirror.json") # MES 桩独立镜像，防幂等键串扰
|
||
reset_wms_client(tmp_path / "wms_mirror.json")
|
||
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
||
_baseline(store)
|
||
client = reset_wms_client(tmp_path / "wms_mirror.json")
|
||
_seed_ledger(client)
|
||
ev = _shortage_event(client, "WIRE-HV", 500, 220, "2026-08-02 09:00:00")
|
||
|
||
# 1) 事件上报 → 消费 + 库存版本 + 影响半径 + Explore 方案卡（P2）
|
||
r = wms_events.consume_shortage_event(store, ev, session_id="s-wms", actor="test")
|
||
assert r["duplicate"] is False
|
||
assert r["staged"] is True
|
||
inv_version = r["inventoryVersion"]
|
||
assert inv_version >= 1
|
||
assert r["card"]["type"] == "confirm-card"
|
||
assert r["card"]["props"]["action"] == "flex.reschedule"
|
||
assert r["card"]["props"]["power"] == "P2"
|
||
assert r["impact"]["affectedOrderCount"] >= 1
|
||
assert all(o["orderNo"] for o in r["impact"]["affectedOrders"])
|
||
assert r["impact"]["readOnly"] is True
|
||
confirm_id = r["confirmId"]
|
||
|
||
# 库存版本进入证据链（wms-event + inventory-version）
|
||
consume_audit = next(e for e in store.data["auditEvents"]
|
||
if e["action"] == "wms.event.consume")
|
||
assert f"wms-event:{ev['eventId']}" in consume_audit["evidenceRefs"]
|
||
assert f"inventory-version:{inv_version}" in consume_audit["evidenceRefs"]
|
||
stage_audit = next(e for e in store.data["auditEvents"]
|
||
if e["action"] == "wms.event.stage")
|
||
assert f"inventory-version:{inv_version}" in stage_audit["evidenceRefs"]
|
||
# 影响半径沙盒只读：主干订单/版本数不变
|
||
assert len(store.data["flexOrders"]) == 10
|
||
|
||
# 2) 人工确认（P2）→ L4 全量重排出 DRAFT 版本
|
||
versions_before = len(store.data["flexScheduleVersions"])
|
||
msg = execute_confirmed(store, confirm_id, approve=True, actor="tester")
|
||
assert "重排已执行" in msg
|
||
assert len(store.data["flexScheduleVersions"]) == versions_before + 1
|
||
new_version = store.data["flexScheduleVersions"][-1]
|
||
assert new_version["status"] == "DRAFT"
|
||
approve_audit = next(e for e in store.data["auditEvents"]
|
||
if e["action"] == "flex.reschedule.approve")
|
||
assert f"inventory-version:{inv_version}" in approve_audit["evidenceRefs"]
|
||
assert f"wms-event:{ev['eventId']}" in approve_audit["evidenceRefs"]
|
||
|
||
# 3) P2 publish establishes the execution baseline; it does not call MES.
|
||
publish_reply = stage_schedule_publish(
|
||
store,
|
||
session_id="s-wms",
|
||
actor="tester",
|
||
track="flex",
|
||
version_id=new_version["id"],
|
||
)
|
||
publish_confirm_id = next(
|
||
block.props["confirmId"]
|
||
for block in publish_reply.blocks
|
||
if block.type == "confirm-card"
|
||
)
|
||
publish_msg = execute_confirmed(
|
||
store,
|
||
publish_confirm_id,
|
||
approve=True,
|
||
actor="tester",
|
||
)
|
||
assert "已发布" in publish_msg
|
||
assert new_version["status"] == "PUBLISHED"
|
||
assert store.data.get("mesLinks") in (None, [])
|
||
|
||
# 3) MES 下发（复用 mes.dispatch，P3 门禁：双人确认）
|
||
stage = wms_events.stage_dispatch(store, ev["eventId"], session_id="s-wms", actor="test")
|
||
assert stage["staged"] is True
|
||
d_confirm_id = stage["block"].props["confirmId"]
|
||
grant = _approve_p3(d_confirm_id)
|
||
checkpoint = store.checkpoints.create(store.data, label="MES 下发前基线",
|
||
reason="auto:mes.dispatch")
|
||
res = wms_events.execute_dispatch_receipt(
|
||
store, event_id=ev["eventId"], confirm_id=d_confirm_id, execution_grant=grant,
|
||
version_id=new_version["id"], before_snapshot=str(checkpoint["pairId"]),
|
||
checkpoint_store=store.checkpoints, actor="tester")
|
||
assert len(res["created"]) >= 1
|
||
assert res["receipt"]["eventId"] == ev["eventId"]
|
||
assert res["receipt"]["scheduleVersionId"] == new_version["id"]
|
||
assert res["receipt"]["inventoryVersion"] == inv_version
|
||
|
||
# 4) 回执进入审计（全链证据：wms-event + inventory-version + schedule-version）
|
||
receipt_audits = [e for e in store.data["auditEvents"]
|
||
if e["action"] == "wms.mes.receipt"]
|
||
assert len(receipt_audits) == 1
|
||
refs = receipt_audits[0]["evidenceRefs"]
|
||
assert f"wms-event:{ev['eventId']}" in refs
|
||
assert f"inventory-version:{inv_version}" in refs
|
||
assert f"schedule-version:{new_version['id']}" in refs
|
||
assert len(receipt_audits[0]["rationale"]["externalWoIds"]) >= 1
|
||
assert len(store.data["wmsReceipts"]) == 1
|
||
|
||
# 全链审计可复现：WMS 缺料 → 影响半径 → 方案卡 → 人工确认 → MES 下发 → 回执
|
||
actions = [e["action"] for e in store.data["auditEvents"]]
|
||
for expected in ("wms.event.consume", "wms.event.stage",
|
||
"flex.reschedule.approve", "mes.dispatch", "wms.mes.receipt"):
|
||
assert expected in actions
|
||
|
||
|
||
# ------------------------------------------------------------
|
||
# 驳回分支：不重排、不留副作用
|
||
# ------------------------------------------------------------
|
||
def test_shortage_card_reject_does_not_reschedule(tmp_path: Path):
|
||
reset_wms_client(tmp_path / "wms_mirror.json")
|
||
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
||
_baseline(store)
|
||
client = reset_wms_client(tmp_path / "wms_mirror.json")
|
||
_seed_ledger(client)
|
||
ev = _shortage_event(client, "WIRE-HV", 500, 220, "2026-08-02 09:00:00")
|
||
r = wms_events.consume_shortage_event(store, ev, session_id="s-wms", actor="test")
|
||
assert r["staged"] is True
|
||
|
||
versions_before = len(store.data["flexScheduleVersions"])
|
||
msg = execute_confirmed(store, r["confirmId"], approve=False, actor="tester")
|
||
assert "已驳回" in msg
|
||
assert len(store.data["flexScheduleVersions"]) == versions_before
|
||
actions = [e["action"] for e in store.data["auditEvents"]]
|
||
assert "flex.reschedule.reject" in actions
|
||
assert all(e["result"] == "DENIED" for e in store.data["auditEvents"]
|
||
if e["action"] == "flex.reschedule.reject")
|
||
# 事件已消费但未触发副作用（无 MES 下发、无回执）
|
||
assert not store.data.get("wmsReceipts") |