aps-agent/tests/golden/test_evidence_contract.py

# 证据链统一协议（§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