group_fqcd_jr/tests/unit/api/test_onboarding_gate.py

import pytest
from fastapi import Request
from fastapi.security import HTTPAuthorizationCredentials

from app.api.dependencies.auth import build_request_context
from app.core.contracts import RequestContext
from app.core.errors import OnboardingRequiredError


def request(path: str) -> Request:
    return Request({"type": "http", "method": "GET", "path": path, "headers": []})


@pytest.mark.asyncio
async def test_customer_without_assessment_is_gated_after_authentication(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    authenticated = RequestContext(user_id="7", trace_id="initial")
    resolved = authenticated.model_copy(update={"roles": ("customer",)})

    class Authenticator:
        def authenticate(self, _token: str) -> RequestContext:
            return authenticated

    async def resolve(_self: object, _context: RequestContext) -> RequestContext:
        return resolved

    async def is_required(_self: object, _context: RequestContext) -> bool:
        return True

    monkeypatch.setattr("app.api.dependencies.auth._authenticator", lambda: Authenticator())
    monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve)
    monkeypatch.setattr(
        "app.service.risk_questionnaire_service.RiskQuestionnaireService.is_required", is_required
    )
    with pytest.raises(OnboardingRequiredError):
        await build_request_context(
            request("/api/v1/agent-runs"),
            HTTPAuthorizationCredentials(scheme="Bearer", credentials="token"),
        )


@pytest.mark.asyncio
async def test_questionnaire_endpoint_is_exempt_from_the_gate(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    authenticated = RequestContext(user_id="7", trace_id="initial")
    resolved = authenticated.model_copy(update={"roles": ("customer",)})

    class Authenticator:
        def authenticate(self, _token: str) -> RequestContext:
            return authenticated

    async def resolve(_self: object, _context: RequestContext) -> RequestContext:
        return resolved

    async def unexpected_check(_self: object, _context: RequestContext) -> bool:
        raise AssertionError("问卷入口不应经过完成状态拦截")

    monkeypatch.setattr("app.api.dependencies.auth._authenticator", lambda: Authenticator())
    monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve)
    monkeypatch.setattr(
        "app.service.risk_questionnaire_service.RiskQuestionnaireService.is_required",
        unexpected_check,
    )
    context = await build_request_context(
        request("/api/v1/onboarding/risk-questionnaire"),
        HTTPAuthorizationCredentials(scheme="Bearer", credentials="token"),
    )
    assert context == resolved


@pytest.mark.asyncio
@pytest.mark.parametrize("role", ["risk_operator", "admin"])
async def test_staff_roles_never_enter_customer_onboarding_gate(
    monkeypatch: pytest.MonkeyPatch,
    role: str,
) -> None:
    authenticated = RequestContext(user_id="7", trace_id="initial")
    resolved = authenticated.model_copy(update={"roles": (role,)})

    class Authenticator:
        def authenticate(self, _token: str) -> RequestContext:
            return authenticated

    async def resolve(_self: object, _context: RequestContext) -> RequestContext:
        return resolved

    async def unexpected_check(_self: object, _context: RequestContext) -> bool:
        raise AssertionError("员工身份不应进入客户风险测评门禁")

    monkeypatch.setattr("app.api.dependencies.auth._authenticator", lambda: Authenticator())
    monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve)
    monkeypatch.setattr(
        "app.service.risk_questionnaire_service.RiskQuestionnaireService.is_required",
        unexpected_check,
    )
    context = await build_request_context(
        request("/api/v1/risk/overview"),
        HTTPAuthorizationCredentials(scheme="Bearer", credentials="token"),
    )
    assert context == resolved