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
|