74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
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
|
|
|