group_fqcd_jr/tests/unit/api/test_permission_enforcement.py

"""端点权限判定的 HTTP 层用例（不连数据库）。

## 为什么单开这一组

既有用例（含 `tests/integration/test_customer_service_handover_admin_mysql.py`）都通过
`app.dependency_overrides[build_request_context]` 注入**已经带好权限**的上下文，
所以只覆盖了"有权限能通"，**覆盖不到"缺权限必须被拒"**。

而 2026-09-12 出过的那类事故（客服二期三个权限码 `memory:candidate:confirm` /
`memory:candidate:review` / `handover:read` 没随代码合并进环境）恰好只会在这一层暴露：

    权限判定发生在身份解析之后（`app/api/dependencies/auth.py` → `IdentityService.resolve`），
    而服务层测试是自己构造 `RequestContext` 的、权限字段由测试塞进去，
    所以"权限码在库里根本不存在"这类问题，单元/集成测试全绿也照样漏。

## 替换了什么、保留了哪些真实部分

- **保留**：真实路由、真实 `build_request_context` 依赖注入位、真实
  `AuthorizationService.require` 判定与 `ForbiddenAgentError` → 403 信封。
- **替换**（只替换两处边界）：
  1. `build_request_context` → 直接返回指定权限集的上下文（跳过 JWT 与身份库查询）；
  2. `AuthorizationService` 的审计落库 → 内存替身（拒绝时会写一条 `permission.denied`）。

## 为什么正反两个方向都要断言

- **反向**（权限集里没有该权限 → 必须 403）：防"端点忘了做权限校验"。
- **正向**（把该权限放进去 → **不能**再是 403）：把"端点要求的权限码"钉住 ——
  若有人改了端点要的权限码而没同步这里，正向会立刻变红。
  正向之后的下游（MySQL / Redis）在单元环境不可用，返回 5xx 属正常；
  本用例只关心"不再因权限被拒"，因此断言 `status_code != 403` 而不是 `== 200`。
"""

from __future__ import annotations

from typing import Any

import pytest
from fastapi.testclient import TestClient

from app.api.dependencies.auth import build_request_context
from app.core.contracts import RequestContext
from app.main import create_app
from app.service import authorization_service

PERMISSION_DENIED_CODE = "AGENT_PERMISSION_DENIED"

#: (HTTP 方法, 路径, 该端点要求的权限码, 角色, 请求体)
CASES: tuple[tuple[str, str, str, tuple[str, ...], dict[str, Any] | None], ...] = (
    (
        "GET",
        "/api/v1/admin/customer-service/handover-tickets",
        "handover:read",
        ("admin",),
        None,
    ),
    (
        "GET",
        "/api/v1/admin/customer-service/handover-tickets/T-0001",
        "handover:read",
        ("admin",),
        None,
    ),
    (
        "GET",
        "/api/v1/admin/customer-profile-candidates",
        "memory:candidate:review",
        ("admin",),
        None,
    ),
    (
        "POST",
        "/api/v1/admin/customer-profile-candidates/1/reviews",
        "memory:candidate:review",
        ("admin",),
        {"decision": "approved"},
    ),
    (
        "POST",
        "/api/v1/users/me/memory-candidates/1/decisions",
        "memory:candidate:confirm",
        ("customer",),
        {"decision": "confirmed"},
    ),
)


class _FakeSession:
    """只承载"拒绝时写一条审计"这一步，不碰数据库。"""

    def add(self, _instance: object) -> None:
        return None

    def begin(self) -> _FakeSession:
        return self

    async def __aenter__(self) -> _FakeSession:
        return self

    async def __aexit__(self, *_exc: object) -> bool:
        return False


class _FakeSessionFactory:
    def __call__(self) -> _FakeSession:
        return _FakeSession()


def _contains_code(payload: Any, code: str) -> bool:
    """在响应 JSON 里递归找错误码，不依赖信封的具体层级。"""
    if isinstance(payload, dict):
        return any(
            (isinstance(value, str) and value == code) or _contains_code(value, code)
            for value in payload.values()
        )
    if isinstance(payload, list):
        return any(_contains_code(item, code) for item in payload)
    return False


def _request(
    monkeypatch: pytest.MonkeyPatch,
    *,
    method: str,
    path: str,
    permissions: tuple[str, ...],
    roles: tuple[str, ...],
    body: dict[str, Any] | None,
) -> Any:
    app = create_app()

    async def override_context() -> RequestContext:
        return RequestContext(
            user_id="9003",
            trace_id="permission-enforcement-trace",
            roles=roles,
            permissions=permissions,
        )

    app.dependency_overrides[build_request_context] = override_context
    monkeypatch.setattr(authorization_service, "SessionFactory", _FakeSessionFactory())
    try:
        # 正向用例会走到下游（MySQL / Redis），单元环境不可用会抛异常；
        # 这里让它变成 5xx 响应，用例只关心"不再因权限被拒"。
        with TestClient(app, raise_server_exceptions=False) as client:
            return client.request(method, path, json=body)
    finally:
        app.dependency_overrides.clear()


@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES)
def test_endpoint_denies_request_without_required_permission(
    monkeypatch: pytest.MonkeyPatch,
    method: str,
    path: str,
    permission: str,
    roles: tuple[str, ...],
    body: dict[str, Any] | None,
) -> None:
    """反向：权限集里没有该权限时，必须 403 且错误码是 `AGENT_PERMISSION_DENIED`。"""
    response = _request(
        monkeypatch,
        method=method,
        path=path,
        permissions=(),
        roles=roles,
        body=body,
    )

    assert response.status_code == 403, (
        f"{method} {path} 缺少 {permission} 时预期 403，实际 {response.status_code}："
        f"{response.text[:200]}"
    )
    assert _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text


@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES)
def test_endpoint_accepts_request_with_required_permission(
    monkeypatch: pytest.MonkeyPatch,
    method: str,
    path: str,
    permission: str,
    roles: tuple[str, ...],
    body: dict[str, Any] | None,
) -> None:
    """正向：放入该权限后不能再是 403 —— 同时把端点要求的权限码钉住。"""
    response = _request(
        monkeypatch,
        method=method,
        path=path,
        permissions=(permission,),
        roles=roles,
        body=body,
    )

    assert response.status_code != 403, (
        f"{method} {path} 带上 {permission} 后仍被拒；"
        f"端点要求的权限码可能已改动：{response.text[:200]}"
    )
    assert not _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text


def test_cases_cover_every_phase2_permission_code() -> None:
    """这组用例必须覆盖客服二期新增的三个权限码，缺一个就失去意义。"""
    covered = {case[2] for case in CASES}
    assert covered == {
        "handover:read",
        "memory:candidate:confirm",
        "memory:candidate:review",
    }