aps-agent/tests/golden/test_harness_p3.py

# ============================================================
# P3 外部副作用二次确认门禁
# ============================================================
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from server.agent_core import harness
from server.agent_core.audit import write_audit
from server.aps_domain.workflow import execute_confirmed
from server.auth.context import IdentityContext, bind_identity, reset_identity


class _AuditStore:
    def __init__(self, world_key: str = "default", tenant_uuid: str = "platform"):
        self.world_key = world_key
        self.tenant_uuid = tenant_uuid
        self.data: dict = {"auditEvents": []}

    def next_id(self, _kind: str) -> int:
        return len(self.data["auditEvents"]) + 1

    def save(self) -> None:
        pass


class _CheckpointStore:
    def create(self, *_args, **_kwargs) -> dict:
        return {"pairId": "test-pair"}


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()


def _stage(action: str) -> str:
    params = ({"track": "flex", "versionId": 1,
               "evidenceRefs": ["schedule-version:1"]}
              if action == "mes.dispatch" else {"versionId": 1})
    block = harness.stage_confirmation(
        "session-p3",
        action,
        params,
        title="测试审批",
        summary_lines=["不会在第一重批准后执行"],
    )
    return str(block.props["confirmId"])


def test_p3_requires_two_approvals_before_token_is_consumed():
    confirm_id = _stage("mes.dispatch")
    first = harness.take_confirmation(confirm_id, approve=True)
    assert first is not None
    assert first["needsSecondConfirm"] is True
    assert first["approvalStep"] == 1
    assert harness.is_confirmation_pending(confirm_id) 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 is not None
    assert second["needsSecondConfirm"] is False
    assert second["approvalStep"] == 2
    assert harness.is_confirmation_pending(confirm_id) is False
    assert harness.take_confirmation(confirm_id, approve=True) is None


def test_p3_second_approval_requires_a_different_user():
    confirm_id = _stage("mes.dispatch")
    first = harness.take_confirmation(confirm_id, approve=True)
    denied = harness.take_confirmation(confirm_id, approve=True)

    assert first and first["approvalStep"] == 1
    assert denied and denied["approvalDenied"] is True
    assert denied["separationRequired"] is True
    assert denied["approvalStep"] == 1
    assert harness.is_confirmation_pending(confirm_id) is True
    assert harness.list_approval_history()[-1]["status"] == "SOD_DENIED"


def test_p3_reject_at_second_step_consumes_token_without_execution():
    confirm_id = _stage("mes.dispatch")
    harness.take_confirmation(confirm_id, approve=True)
    rejected = harness.take_confirmation(confirm_id, approve=False)
    assert rejected is not None
    assert rejected["needsSecondConfirm"] is False
    assert harness.is_confirmation_pending(confirm_id) is False


def test_p3_reject_audit_keeps_p3_power_classification():
    store = _AuditStore()
    confirm_id = _stage("mes.dispatch")
    execute_confirmed(store, confirm_id, approve=True, actor="planner")
    message = execute_confirmed(store, confirm_id, approve=False, actor="planner")

    assert "驳回" in message
    rejected = store.data["auditEvents"][-1]
    assert rejected["action"] == "mes.dispatch.reject"
    assert rejected["power"] == "P3"
    assert rejected["result"] == "DENIED"


def test_p3_confirm_api_executes_external_effect_only_after_second_approval(monkeypatch):
    import server.aps_domain.mes as mes_module
    import server.aps_domain.workflow as workflow_module
    import server.gateway.app as gateway_module
    import server.state.projects as projects_module
    from server.auth.context import bind_identity, reset_identity
    from tests.auth_provider import install_test_auth

    store = _AuditStore(world_key="personal-1001", tenant_uuid="tenant-p3-test")
    peer_store = _AuditStore(world_key="personal-1002", tenant_uuid="tenant-p3-test")
    calls: list[str] = []

    def fake_dispatch(
        _store,
        *,
        track: str,
        actor: str,
        confirm_id: str,
        execution_grant: str,
        version_id: int,
        before_snapshot: str,
        evidence_refs: list[str],
        checkpoint_store,
    ) -> dict:
        assert _store is store
        assert confirm_id == confirm_id_from_test
        assert execution_grant
        assert version_id == 1
        assert before_snapshot == "test-pair"
        assert evidence_refs == ["schedule-version:1"]
        assert isinstance(checkpoint_store, _CheckpointStore)
        calls.append(f"{track}:{actor}")
        return {"message": "MES 下发完成"}

    def current_store():
        from server.auth.context import get_identity

        return store if get_identity().user_id == 1001 else peer_store

    monkeypatch.setattr(gateway_module, "get_store", current_store)
    monkeypatch.setattr(
        "server.state.store._scoped_store",
        lambda tenant_uuid, world_key: (
            store
            if (tenant_uuid, world_key) == ("tenant-p3-test", "personal-1001")
            else peer_store
        ),
    )
    monkeypatch.setattr(workflow_module, "get_checkpoints", lambda: _CheckpointStore())
    monkeypatch.setattr(mes_module, "apply_dispatch", fake_dispatch)
    monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
    provider = install_test_auth(monkeypatch, "tenant-p3-test")
    identity_token = bind_identity(provider._identity("planner"))
    try:
        confirm_id = _stage("mes.dispatch")
    finally:
        reset_identity(identity_token)
    confirm_id_from_test = confirm_id
    client = TestClient(gateway_module.create_app())
    login = client.post("/api/auth/login", json={"username": "planner"})
    assert login.status_code == 200

    first = client.post(
        "/api/actions/confirm",
        json={"sessionId": "planner", "confirmId": confirm_id, "approve": True},
    )
    assert first.status_code == 200
    assert first.json()["secondConfirmRequired"] is True
    assert first.json()["refresh"] is False
    assert calls == []

    second = client.post(
        "/api/auth/login", json={"username": "collaborator"},
    )
    assert second.status_code == 200

    second = client.post(
        "/api/actions/confirm",
        json={"sessionId": "forged-admin", "confirmId": confirm_id, "approve": True},
    )
    assert second.status_code == 200
    assert second.json()["secondConfirmRequired"] is False
    assert second.json()["refresh"] is True
    assert calls == ["flex:collaborator"]

    duplicate = client.post(
        "/api/actions/confirm",
        json={"sessionId": "planner", "confirmId": confirm_id, "approve": True},
    )
    assert duplicate.status_code == 200
    assert duplicate.json()["refresh"] is False
    assert duplicate.json()["secondConfirmRequired"] is False
    assert calls == ["flex:collaborator"]


def test_p2_keeps_single_approval_semantics():
    confirm_id = _stage("schedule.publish")
    approved = harness.take_confirmation(confirm_id, approve=True)
    assert approved is not None
    assert approved["needsSecondConfirm"] is False
    assert approved["approvalStep"] == 1
    assert harness.is_confirmation_pending(confirm_id) is False


def test_pending_status_does_not_leak_across_tenants():
    owner = IdentityContext(1, "owner", "Owner", "tenant-a", roles=("planner",))
    outsider = IdentityContext(2, "outsider", "Outsider", "tenant-b")
    token = bind_identity(owner)
    try:
        confirm_id = _stage("mes.dispatch")
    finally:
        reset_identity(token)
    token = bind_identity(outsider)
    try:
        assert harness.is_confirmation_pending(confirm_id) is False
        assert harness.take_confirmation(confirm_id, approve=True) is None
    finally:
        reset_identity(token)


def test_audit_actor_is_derived_from_authenticated_identity():
    store = _AuditStore()
    token = bind_identity(IdentityContext(
        77,
        "actual-user",
        "Actual User",
        "platform",
        roles=("planner",),
    ))
    try:
        event = write_audit(
            store.data,
            store.next_id,
            actor="forged-admin",
            category="GATE",
            action="mes.dispatch.stage",
            target={"type": "MES", "id": "stage"},
            power="P3",
            rationale={},
        )
    finally:
        reset_identity(token)

    assert event["actor"] == "actual-user"
    assert event["actorId"] == "77"
    assert event["rationale"]["requestedActor"] == "forged-admin"


def test_permission_error_from_stage_is_returned_as_403(monkeypatch):
    import server.aps_domain.mes as mes_module
    import server.gateway.app as gateway_module
    from tests.auth_provider import install_test_auth

    def denied(*_args, **_kwargs):
        raise PermissionError("当前身份没有发起审批的角色权限")

    monkeypatch.setattr(mes_module, "stage_dispatch", denied)
    install_test_auth(monkeypatch, "tenant-stage-role")
    client = TestClient(gateway_module.create_app())
    assert client.post("/api/auth/login", json={"username": "planner"}).status_code == 200

    response = client.post(
        "/api/mes/dispatch/stage",
        json={"sessionId": "web", "track": "flex"},
    )
    assert response.status_code == 403
    assert response.json()["error"]["code"] == "FORBIDDEN"