aps-agent/tests/golden/test_approval_delegate.py

# 审批委托（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"