group_fqcd_jr/tests/unit/service/test_authorization_service.py

"""`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