286 lines
9.7 KiB
Python
286 lines
9.7 KiB
Python
# ============================================================
|
|
# 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"
|