250 lines
11 KiB
Python
250 lines
11 KiB
Python
# 证据链统一协议（§3.4 证据链 v1）黄金测试
|
||
# ============================================================
|
||
# 覆盖：stage 冻结证据（evidenceRefs/beforeSnapshot/beforeFingerprint）、
|
||
# execute_confirmed 统一强校验（缺证据拒绝、版本失效拒绝、版本替换拒绝）、
|
||
# 全部 P2/P3 写审计携带 beforeSnapshot/evidenceRefs、mes.dispatch 强绑定回归。
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.agent_core import harness
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
|
||
|
||
class _AuditStore:
|
||
# world_key 与 _PLANNER(user 2001) 的 _effective_world_key 派生一致，
|
||
# 使 resolve_confirmation_store 回退到本 store 而不创建真实 scoped store。
|
||
def __init__(self, world_key: str = "personal-2001", tenant_uuid: str = "platform"):
|
||
self.world_key = world_key
|
||
self.tenant_uuid = tenant_uuid
|
||
self.data: dict = {"auditEvents": [], "scheduleVersions": [], "flexScheduleVersions": []}
|
||
|
||
def next_id(self, _kind: str) -> int:
|
||
return len(self.data["auditEvents"]) + 1
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
class _CheckpointStore:
|
||
def __init__(self) -> None:
|
||
self.pairs: list[dict] = []
|
||
|
||
def create(self, *_args, **_kwargs) -> dict:
|
||
pair = {"pairId": f"pair-{len(self.pairs) + 1}"}
|
||
self.pairs.append(pair)
|
||
return pair
|
||
|
||
def get(self, pair_id: str) -> dict | None:
|
||
return next((p for p in self.pairs if p["pairId"] == pair_id), None)
|
||
|
||
|
||
class _ProjectStore:
|
||
def active_world_key(self) -> str:
|
||
return "default"
|
||
|
||
def require_active_write(self) -> None:
|
||
pass
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_pending():
|
||
harness._approval_store.clear()
|
||
yield
|
||
harness._approval_store.clear()
|
||
|
||
|
||
_PLANNER = IdentityContext(2001, "planner", "Planner", "platform",
|
||
roles=("planner", "approver", "admin"))
|
||
|
||
|
||
def _stage(action: str, *, extra: dict | None = None, evidence_refs: list[str] | None = None) -> str:
|
||
params = {"versionId": 1, **(extra or {})}
|
||
token = bind_identity(_PLANNER)
|
||
try:
|
||
block = harness.stage_confirmation(
|
||
"session-evidence",
|
||
action,
|
||
params,
|
||
title="测试审批",
|
||
summary_lines=["证据链统一协议测试"],
|
||
evidence_refs=evidence_refs,
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
return str(block.props["confirmId"])
|
||
|
||
|
||
def _approve(store, confirm_id: str, actor: str = "planner") -> str:
|
||
token = bind_identity(IdentityContext(2001, actor, "Planner", "platform",
|
||
roles=("planner", "approver", "admin")))
|
||
try:
|
||
return execute_confirmed(store, confirm_id, approve=True, actor=actor)
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_stage_freeze_records_evidence_contract():
|
||
"""stage_confirmation 冻结证据：显式 evidenceRefs 与 versionId 派生一致。"""
|
||
confirm_id = _stage("mes.dispatch", evidence_refs=["schedule-version:1"])
|
||
import time as _time
|
||
pending = harness._approval_store.pending_record(
|
||
confirm_id, allowed=lambda r: True, now_epoch=_time.time(),
|
||
)
|
||
assert pending is not None
|
||
assert pending["evidenceRefs"] == ["schedule-version:1"]
|
||
assert pending["params"]["versionId"] == 1
|
||
# 版本证据协议：versionId 派生
|
||
assert harness.evidence_refs_for({"versionId": 7}) == ["schedule-version:7"]
|
||
assert harness.evidence_refs_for({}) == []
|
||
|
||
|
||
def test_world_fingerprint_stable_across_append_only_journals():
|
||
"""世界指纹排除 append-only 日志/审计/同步流水；业务变化则改变指纹。"""
|
||
world = {"scheduleVersions": [{"id": 1}], "salesOrders": [], "auditEvents": [{"a": 1}]}
|
||
fp1 = harness.world_fingerprint(world)
|
||
world["auditEvents"].append({"a": 2}) # 审计增长不误报漂移
|
||
assert harness.world_fingerprint(world) == fp1
|
||
world["salesOrders"].append({"orderNo": "SO-1"}) # 业务变化触发漂移
|
||
assert harness.world_fingerprint(world) != fp1
|
||
|
||
|
||
def test_verify_pending_evidence_rejects_world_fingerprint_drift():
|
||
world = {"auditEvents": [], "logs": [], "materials": []}
|
||
pending = {
|
||
"params": {},
|
||
"evidenceRefs": [f"masterdata-world:{harness.world_fingerprint(world)}"],
|
||
}
|
||
world["materials"].append({"id": 999, "code": "DRIFT"})
|
||
with pytest.raises(PermissionError, match="输入世界已漂移"):
|
||
harness.verify_pending_evidence(pending, current_world=world)
|
||
|
||
|
||
def test_verify_pending_evidence_missing_version_evidence_rejected():
|
||
"""缺版本证据：versionId 存在但 evidenceRefs 不含 schedule-version:<id> → 拒绝。"""
|
||
pending = {
|
||
"params": {"versionId": 3},
|
||
"evidenceRefs": ["other:ref"],
|
||
}
|
||
with pytest.raises(PermissionError, match="缺少版本证据"):
|
||
harness.verify_pending_evidence(pending)
|
||
|
||
|
||
def test_verify_pending_evidence_missing_snapshot_rejected():
|
||
"""beforeSnapshot 指向的快照不存在 → 拒绝。"""
|
||
pending = {
|
||
"params": {"versionId": 1},
|
||
"evidenceRefs": ["schedule-version:1"],
|
||
"beforeSnapshot": "missing-pair",
|
||
}
|
||
cp_store = _CheckpointStore()
|
||
with pytest.raises(PermissionError, match="前置快照不存在"):
|
||
harness.verify_pending_evidence(pending, checkpoint_store=cp_store)
|
||
|
||
|
||
def test_verify_pending_evidence_version_unresolvable_rejected():
|
||
"""版本不可解析（世界带版本表但找不到 versionId）→ 拒绝。"""
|
||
pending = {"params": {"versionId": 99}, "evidenceRefs": ["schedule-version:99"]}
|
||
world = {"scheduleVersions": [{"id": 1}]}
|
||
with pytest.raises(PermissionError, match="版本不存在或已失效"):
|
||
harness.verify_pending_evidence(pending, version_lookup=lambda vid: any(v["id"] == vid for v in world["scheduleVersions"]))
|
||
|
||
|
||
def test_execute_confirmed_p2_audit_carries_before_snapshot_and_evidence():
|
||
"""P2 动作（schedule.publish）执行后审计携带 beforeSnapshot + evidenceRefs。"""
|
||
store = _AuditStore()
|
||
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
||
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
||
"totalTardiness": 0})
|
||
store.data["productionOrders"] = []
|
||
confirm_id = _stage("schedule.publish")
|
||
_cp = _CheckpointStore()
|
||
from unittest import mock
|
||
with mock.patch("server.aps_domain.workflow.get_checkpoints", lambda: _cp):
|
||
msg = _approve(store, confirm_id)
|
||
assert "已发布" in msg
|
||
event = store.data["auditEvents"][-1]
|
||
assert event["action"] == "schedule.publish"
|
||
assert event["beforeSnapshot"] is not None and event["beforeSnapshot"].startswith("pair-")
|
||
assert event["evidenceRefs"] == ["schedule-version:1"]
|
||
|
||
|
||
def test_execute_confirmed_rejects_missing_version_evidence():
|
||
"""统一强校验 fail closed：版本证据缺失 → 拒绝且无世界写入。"""
|
||
store = _AuditStore()
|
||
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
||
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
||
"totalTardiness": 0})
|
||
store.data["productionOrders"] = []
|
||
confirm_id = _stage("schedule.publish", evidence_refs=["wrong:ref"])
|
||
msg = _approve(store, confirm_id)
|
||
assert "证据校验未通过" in msg
|
||
assert all(e["action"] != "schedule.publish" or e["result"] == "DENIED"
|
||
for e in store.data["auditEvents"])
|
||
assert store.data["scheduleVersions"][0]["status"] == "DRAFT" # 未发布
|
||
|
||
|
||
def test_execute_confirmed_rejects_version_substitution():
|
||
"""版本替换：证据指向版本 1，但执行端版本已失效/被替换 → 拒绝。"""
|
||
store = _AuditStore()
|
||
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
||
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
||
"totalTardiness": 0})
|
||
store.data["productionOrders"] = []
|
||
confirm_id = _stage("schedule.publish", evidence_refs=["schedule-version:1"])
|
||
# 出卡后版本被替换：原版本 1 被删除（世界带版本表但不可解析）
|
||
store.data["scheduleVersions"] = [{"id": 2, "versionNo": "V2"}]
|
||
msg = _approve(store, confirm_id)
|
||
assert "证据校验未通过" in msg or "版本不存在" in msg
|
||
|
||
|
||
def test_flex_time_update_audit_carries_before_snapshot():
|
||
"""B1 回归：flex.time.update 写世界前必须有写前快照，审计携带 beforeSnapshot。"""
|
||
store = _AuditStore()
|
||
store.data["flexRoutings"] = [
|
||
{"operationCode": "OP-1", "productCode": "P-1", "stdTimePerUnit": 5.0, "stdTimeSource": "实测"},
|
||
]
|
||
store.data["flexOperations"] = [{"code": "OP-1", "stdTimeMin": 5.0}]
|
||
params = {"operationCode": "OP-1", "productCode": "P-1", "stdMin": 9.0, "source": "实测"}
|
||
token = bind_identity(_PLANNER)
|
||
try:
|
||
block = harness.stage_confirmation(
|
||
"session-evidence", "flex.time.update", params,
|
||
title="测试审批", summary_lines=["工时维护快照绑定"],
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
confirm_id = str(block.props["confirmId"])
|
||
_cp = _CheckpointStore()
|
||
from unittest import mock
|
||
with mock.patch("server.aps_domain.workflow.get_checkpoints", lambda: _cp):
|
||
msg = _approve(store, confirm_id)
|
||
assert "已更新" in msg
|
||
event = store.data["auditEvents"][-1]
|
||
assert event["action"] == "flex.time.update"
|
||
assert event["beforeSnapshot"] is not None and event["beforeSnapshot"].startswith("pair-")
|
||
assert event["evidenceRefs"] == []
|
||
assert store.data["flexRoutings"][0]["stdTimePerUnit"] == 9.0
|
||
|
||
|
||
def test_mes_dispatch_strong_binding_regression():
|
||
"""mes.dispatch 强绑定回归：证据与版本一致时走确认流程且无回归。"""
|
||
_AuditStore()
|
||
confirm_id = _stage("mes.dispatch", extra={"track": "flex"}, evidence_refs=["schedule-version:1"])
|
||
# P3 第一次批准不执行
|
||
from server.auth.context import bind_identity, reset_identity
|
||
token = bind_identity(IdentityContext(2001, "planner", "Planner", "platform", roles=("planner",)))
|
||
try:
|
||
first = harness.take_confirmation(confirm_id, approve=True)
|
||
finally:
|
||
reset_identity(token)
|
||
assert first is not None and first["needsSecondConfirm"] is True
|
||
# 第二次批准由不同用户完成
|
||
token2 = bind_identity(IdentityContext(2002, "peer", "Peer", "platform", roles=("planner",)))
|
||
try:
|
||
second = harness.take_confirmation(confirm_id, approve=True)
|
||
finally:
|
||
reset_identity(token2)
|
||
assert second is not None and second["needsSecondConfirm"] is False
|