121 lines
5.3 KiB
Python
121 lines
5.3 KiB
Python
# 审批委托（delegation）黄金测试（plan.md §3.3 / 矩阵 P3）
|
||
# ============================================================
|
||
# 覆盖：delegate 可审批 P2、非 delegate 拒绝、owner 仍可、P3 双人分离不受影响、
|
||
# 双后端（文件/DB）record 持久化 delegateUserId。
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from server.agent_core import harness
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
|
||
|
||
def _identity(user_id: int, tenant: str = "platform") -> IdentityContext:
|
||
return IdentityContext(user_id, f"user-{user_id}", f"User {user_id}", tenant,
|
||
roles=("planner", "approver", "admin"))
|
||
|
||
|
||
def _as(identity: IdentityContext, callback):
|
||
token = bind_identity(identity)
|
||
try:
|
||
return callback()
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def _stage(owner: IdentityContext, action: str = "schedule.publish",
|
||
delegate_user_id: int | None = None) -> str:
|
||
params = {"versionId": 91}
|
||
token = bind_identity(owner)
|
||
try:
|
||
block = harness.stage_confirmation(
|
||
"delegate-test", action, params, title="委托测试", summary_lines=["t"],
|
||
delegate_user_id=delegate_user_id,
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
return str(block.props["confirmId"])
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_pending():
|
||
# 文件后端在测试前后清空；DB 后端由 database_backend fixture 管理（避免 reset 后跨库 clear）
|
||
if getattr(harness._approval_store, "backend", "file") == "file":
|
||
harness._approval_store.clear()
|
||
yield
|
||
if getattr(harness._approval_store, "backend", "file") == "file":
|
||
harness._approval_store.clear()
|
||
|
||
|
||
@pytest.fixture
|
||
def database_backend(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""共享 database 审批后端（SQLite 测试介质，configure 时自动迁移建表）。"""
|
||
from server.db.database import reset_engine
|
||
monkeypatch.setenv("APS_APPROVAL_BACKEND", "database")
|
||
monkeypatch.delenv("APS_APPROVAL_PATH", raising=False)
|
||
db_path = (tmp_path / "approval.db").as_posix()
|
||
monkeypatch.setenv("APS_DATABASE_URL", f"sqlite:///{db_path}")
|
||
monkeypatch.setenv("APS_APPROVAL_DATABASE_ALLOW_SQLITE", "1")
|
||
reset_engine()
|
||
store = harness.configure_approval_store() # 内部校验/迁移共享审批表
|
||
yield store
|
||
store.clear()
|
||
reset_engine()
|
||
|
||
|
||
def test_delegate_can_approve_p2(tmp_path: Path):
|
||
"""P2 委托：被委托人可审批通过。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner, delegate = _identity(4001), _identity(4002)
|
||
confirm_id = _stage(owner, delegate_user_id=4002)
|
||
result = _as(delegate, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert result is not None
|
||
assert result["needsSecondConfirm"] is False
|
||
assert str(result["approvals"][-1]["userId"]) == "4002"
|
||
|
||
|
||
def test_non_delegate_rejected(tmp_path: Path):
|
||
"""非 owner 非 delegate 的第三人不可审批。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner, stranger = _identity(4001), _identity(4003)
|
||
confirm_id = _stage(owner, delegate_user_id=4002)
|
||
result = _as(stranger, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert result is None
|
||
|
||
|
||
def test_owner_still_can_approve_when_delegated(tmp_path: Path):
|
||
"""委托后 owner 仍可审批（委托是扩展，不是转移）。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner = _identity(4001)
|
||
confirm_id = _stage(owner, delegate_user_id=4002)
|
||
result = _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert result is not None and result["needsSecondConfirm"] is False
|
||
|
||
|
||
def test_p3_separation_of_duties_unchanged_by_delegate(tmp_path: Path):
|
||
"""P3 双人分离不受委托影响：同一用户（含被委托）不能完成两次批准。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner, delegate = _identity(4001), _identity(4002)
|
||
confirm_id = _stage(owner, action="mes.dispatch", delegate_user_id=4002)
|
||
# 第一次：owner 批准（P3 首重）
|
||
first = _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert first is not None and first["needsSecondConfirm"] is True
|
||
# 第二次仍由 owner（同人）→ 拒绝（SOD）
|
||
denied = _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert denied is not None and denied.get("separationRequired") is True
|
||
# 第二次由 delegate（不同人）→ 通过并签发 grant
|
||
second = _as(delegate, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert second is not None and second["needsSecondConfirm"] is False
|
||
assert second.get("executionGrant")
|
||
|
||
|
||
def test_database_backend_persists_delegate(database_backend):
|
||
"""共享 database 后端：delegateUserId 随 record 持久化，delegate 可审批。"""
|
||
owner, delegate = _identity(7101), _identity(7102)
|
||
confirm_id = _as(owner, lambda: _stage(owner, delegate_user_id=7102))
|
||
approved = _as(delegate, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert approved is not None and approved["needsSecondConfirm"] is False
|
||
assert str(approved["approvals"][-1]["userId"]) == "7102"
|