119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
# 审批转派（transfer）黄金测试（plan.md §3.3 / 矩阵 P3）
|
||
# ============================================================
|
||
# 覆盖：owner 转派、非 owner 拒绝、P3 拒绝、转派后目标可批原 owner 不能、DB 持久化。
|
||
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") -> str:
|
||
params = {"versionId": 91}
|
||
token = bind_identity(owner)
|
||
try:
|
||
block = harness.stage_confirmation(
|
||
"transfer-test", action, params, title="转派测试", summary_lines=["t"],
|
||
)
|
||
finally:
|
||
reset_identity(token)
|
||
return str(block.props["confirmId"])
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_pending():
|
||
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):
|
||
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_owner_transfers_to_target(tmp_path: Path):
|
||
"""owner 可转派；转派后目标用户可审批，原 owner 不能。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner, target = _identity(4001), _identity(4002)
|
||
confirm_id = _stage(owner)
|
||
transferred = _as(owner, lambda: harness.transfer_confirmation(confirm_id, 4002))
|
||
assert transferred is not None
|
||
assert transferred["ownerUserId"] == 4002
|
||
# 原 owner 不可审批
|
||
assert _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True)) is None
|
||
# 目标用户可审批
|
||
approved = _as(target, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert approved is not None and approved["needsSecondConfirm"] is False
|
||
# 历史含 TRANSFERRED（转派后 owner 是 target，用 target 身份查看）
|
||
history = _as(target, harness.list_approval_history)
|
||
assert any(h["status"] == "TRANSFERRED" for h in history)
|
||
|
||
|
||
def test_non_owner_cannot_transfer(tmp_path: Path):
|
||
"""非 owner 非 delegate 的第三人不可转派。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner, stranger = _identity(4001), _identity(4003)
|
||
confirm_id = _stage(owner)
|
||
result = _as(stranger, lambda: harness.transfer_confirmation(confirm_id, 4002))
|
||
assert result is None
|
||
|
||
|
||
def test_p3_transfer_rejected(tmp_path: Path):
|
||
"""P3 转派拒绝（保双人职责分离）。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner = _identity(4001)
|
||
confirm_id = _stage(owner, action="mes.dispatch")
|
||
result = _as(owner, lambda: harness.transfer_confirmation(confirm_id, 4002))
|
||
assert result is None
|
||
|
||
|
||
def test_transfer_to_self_is_noop(tmp_path: Path):
|
||
"""转派给自己 = no-op（仍可审批）。"""
|
||
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
||
owner = _identity(4001)
|
||
confirm_id = _stage(owner)
|
||
result = _as(owner, lambda: harness.transfer_confirmation(confirm_id, 4001))
|
||
assert result is not None
|
||
approved = _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert approved is not None
|
||
|
||
|
||
def test_database_backend_transfer_persists(database_backend):
|
||
"""共享 database 后端：转派持久化并生效。"""
|
||
owner, target = _identity(7201), _identity(7202)
|
||
confirm_id = _as(owner, lambda: _stage(owner))
|
||
transferred = _as(owner, lambda: harness.transfer_confirmation(confirm_id, 7202))
|
||
assert transferred is not None and transferred["ownerUserId"] == 7202
|
||
assert _as(owner, lambda: harness.take_confirmation(confirm_id, approve=True)) is None
|
||
approved = _as(target, lambda: harness.take_confirmation(confirm_id, approve=True))
|
||
assert approved is not None and approved["needsSecondConfirm"] is False
|