张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
"""`AuthorizationService.require` 契约测试（替身 session，不连数据库）。
|
||
|
||
它是所有受保护操作的公共闸门，覆盖四点：
|
||
1. 有权限时**直接返回**，不写审计、更不碰数据库——用"创建会话即报错"的替身证明；
|
||
2. 拒绝时**开新事务**写 `permission.denied` 审计，并抛 `ForbiddenAgentError`；
|
||
3. `admin=True` 时除了权限码还必须具备 admin/super_admin 角色；
|
||
4. 审计里必须留下 permission 与 trace_id，供事后追责。
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import ForbiddenAgentError
|
||
from app.service.authorization_service import AuthorizationService
|
||
|
||
BASE = RequestContext(user_id="9001", trace_id="trace-9")
|
||
|
||
|
||
def context_with(*permissions: str, roles: tuple[str, ...] = ()) -> RequestContext:
|
||
return RequestContext(
|
||
user_id="9001", trace_id="trace-9", permissions=tuple(permissions), roles=roles
|
||
)
|
||
|
||
|
||
class _AsyncContext:
|
||
async def __aenter__(self) -> None:
|
||
return None
|
||
|
||
async def __aexit__(self, *exc: object) -> bool:
|
||
return False
|
||
|
||
|
||
class FakeSession:
|
||
def __init__(self, sink: list[Any]) -> None:
|
||
self.sink = sink
|
||
|
||
def add(self, value: Any) -> None:
|
||
self.sink.append(value)
|
||
|
||
def begin(self) -> Any:
|
||
return _AsyncContext()
|
||
|
||
|
||
def patch_session(monkeypatch: pytest.MonkeyPatch, sink: list[Any]) -> None:
|
||
class FakeFactory:
|
||
async def __aenter__(self) -> FakeSession:
|
||
return FakeSession(sink)
|
||
|
||
async def __aexit__(self, *exc: object) -> bool:
|
||
return False
|
||
|
||
monkeypatch.setattr("app.service.authorization_service.SessionFactory", FakeFactory)
|
||
|
||
|
||
def explode_on_session() -> None:
|
||
raise AssertionError("有权限时不应创建数据库会话")
|
||
|
||
|
||
async def test_allowed_permission_returns_without_database_access(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr("app.service.authorization_service.SessionFactory", explode_on_session)
|
||
|
||
await AuthorizationService.require(
|
||
context_with("conversation:feedback"), "conversation:feedback"
|
||
)
|
||
|
||
|
||
async def test_denied_permission_writes_audit_and_raises(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
sink: list[Any] = []
|
||
patch_session(monkeypatch, sink)
|
||
|
||
with pytest.raises(ForbiddenAgentError):
|
||
await AuthorizationService.require(context_with("agent:run"), "config:write")
|
||
|
||
assert len(sink) == 1
|
||
audit = sink[0]
|
||
assert audit.action_type == "permission.denied"
|
||
assert audit.actor_id == 9001
|
||
assert audit.actor_type == "user"
|
||
assert audit.detail == {"permission": "config:write", "trace_id": "trace-9"}
|
||
|
||
|
||
async def test_admin_flag_requires_admin_role_even_with_permission(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""权限码齐备但角色不是 admin：必须拒绝，否则普通角色能进管理面。"""
|
||
sink: list[Any] = []
|
||
patch_session(monkeypatch, sink)
|
||
|
||
with pytest.raises(ForbiddenAgentError):
|
||
await AuthorizationService.require(
|
||
context_with("config:read", roles=("customer",)), "config:read", admin=True
|
||
)
|
||
|
||
assert len(sink) == 1
|
||
|
||
|
||
async def test_admin_flag_passes_for_admin_role(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr("app.service.authorization_service.SessionFactory", explode_on_session)
|
||
|
||
await AuthorizationService.require(
|
||
context_with("config:read", roles=("admin",)), "config:read", admin=True
|
||
)
|
||
|
||
|
||
async def test_admin_flag_still_requires_the_permission_code(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""反之亦然：是 admin 角色但没有该权限码，同样拒绝。"""
|
||
sink: list[Any] = []
|
||
patch_session(monkeypatch, sink)
|
||
|
||
with pytest.raises(ForbiddenAgentError):
|
||
await AuthorizationService.require(
|
||
context_with(roles=("admin",)), "config:read", admin=True
|
||
)
|
||
|
||
assert len(sink) == 1
|