f5a5bbf95f
Dev-эмитент подключается только при AUTH_DEV_ISSUER и DEBUG; статический тест «нет кодов ролей» расширен на src/api/** и src/services/**. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
505 lines
22 KiB
Python
505 lines
22 KiB
Python
"""Зависимости аутентификации и прав (spec authn-authz, задача 17): `current_user`, `require(code)`, 401/403.
|
||
|
||
БД-тесты (`integration`) работают на `db_session` с seed dev-данных; мини-приложение собрано в тесте без lifespan,
|
||
верификатор токенов подменён на `DevHs256Verifier`. Быстрые тесты (без БД) проверяют разбор заголовка и то, что
|
||
401-ветки не обращаются к сессии.
|
||
"""
|
||
|
||
import ast
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Callable, Iterator
|
||
from datetime import timedelta
|
||
from pathlib import Path
|
||
from typing import Annotated, Any
|
||
|
||
import httpx
|
||
import pytest
|
||
import yaml
|
||
from fastapi import Depends, FastAPI, HTTPException
|
||
from sqlalchemy import event, select, update
|
||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
||
|
||
from scripts.seed import run_seed
|
||
from src.api.dependencies.auth import (
|
||
Principal,
|
||
current_user,
|
||
get_token_verifier,
|
||
parse_bearer,
|
||
require,
|
||
)
|
||
from src.core.auth import DevHs256Verifier
|
||
from src.core.exception_handlers import register_exception_handlers
|
||
from src.db.db import get_async_session
|
||
from src.models.access import AppUser, Role, UserRole
|
||
from src.models.directories import RegionalOffice
|
||
from src.models.enums import BusinessSegment
|
||
from src.repositories.users import AmbiguousUserError, load_principal_data
|
||
from src.schemas.me import MeResponse, RegionalOfficeOut
|
||
|
||
SECRET = "test-secret-for-auth-deps-0123456789abcdef"
|
||
OTHER_SECRET = "foreign-secret-for-auth-deps-0123456789ab"
|
||
SRC = Path(__file__).resolve().parent.parent / "src"
|
||
SEED_CORE = Path(__file__).resolve().parent.parent / "seed" / "core"
|
||
|
||
NOT_REGISTERED = "пользователь не зарегистрирован в АРМ"
|
||
DISABLED = "учётная запись отключена"
|
||
AMBIGUOUS = "неоднозначная учётная запись"
|
||
|
||
|
||
class _ForbiddenSession:
|
||
"""Заглушка сессии для 401-веток: любое обращение — ошибка теста."""
|
||
|
||
def __getattr__(self, name: str) -> Any:
|
||
raise AssertionError(f"обращение к БД при 401: session.{name}")
|
||
|
||
|
||
def _build_app(session_factory: Callable[[], AsyncIterator[Any]], verifier: DevHs256Verifier | None) -> FastAPI:
|
||
app = FastAPI()
|
||
register_exception_handlers(app)
|
||
|
||
@app.get("/whoami")
|
||
async def whoami(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, Any]:
|
||
return {
|
||
"email": principal.user.email,
|
||
"roles": list(principal.roles),
|
||
"permissions": sorted(principal.permissions),
|
||
"office": principal.regional_office.name,
|
||
}
|
||
|
||
@app.get("/kfo-verify", dependencies=[Depends(require("KFO_VERIFY"))])
|
||
async def kfo_verify() -> dict[str, bool]:
|
||
return {"ok": True}
|
||
|
||
@app.get("/doc-edit", dependencies=[Depends(require("DOCUMENT_EDIT"))])
|
||
async def doc_edit() -> dict[str, bool]:
|
||
return {"ok": True}
|
||
|
||
@app.get("/both", dependencies=[Depends(require("KFO_VERIFY")), Depends(require("DOCUMENT_EDIT"))])
|
||
async def both(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, bool]:
|
||
return {"ok": True}
|
||
|
||
app.dependency_overrides[get_async_session] = session_factory
|
||
app.dependency_overrides[get_token_verifier] = lambda: verifier
|
||
return app
|
||
|
||
|
||
def _client(app: FastAPI) -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test")
|
||
|
||
|
||
def _bearer(token: str) -> dict[str, str]:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
# --- быстрые тесты без БД ---------------------------------------------------------------------------------------
|
||
|
||
|
||
def test_parse_bearer_ok() -> None:
|
||
assert parse_bearer("Bearer abc.def.ghi") == "abc.def.ghi"
|
||
assert parse_bearer("bearer abc") == "abc"
|
||
|
||
|
||
@pytest.mark.parametrize("header", [None, "", "Bearer", "Bearer ", "Bearer ", "Basic abc", "abc.def", "Bearer a b"])
|
||
def test_parse_bearer_rejects(header: str | None) -> None:
|
||
with pytest.raises(HTTPException) as info:
|
||
parse_bearer(header)
|
||
assert info.value.status_code == 401 # noqa: PLR2004
|
||
assert info.value.headers == {"WWW-Authenticate": "Bearer"}
|
||
assert "Authorization: Bearer" in info.value.detail
|
||
|
||
|
||
def test_require_rejects_empty_code() -> None:
|
||
with pytest.raises(ValueError, match="код права"):
|
||
require("")
|
||
with pytest.raises(ValueError, match="код права"):
|
||
require(" ")
|
||
|
||
|
||
@pytest.fixture
|
||
def untouchable_app() -> tuple[FastAPI, list[str]]:
|
||
"""Приложение с заглушкой сессии; список `touched` пополняется, если зависимость сессии вообще вызвана."""
|
||
touched: list[str] = []
|
||
|
||
async def session() -> AsyncIterator[_ForbiddenSession]:
|
||
touched.append("session")
|
||
yield _ForbiddenSession()
|
||
|
||
verifier = DevHs256Verifier(SECRET)
|
||
return _build_app(session, verifier), touched
|
||
|
||
|
||
async def _expect_401(app: FastAPI, headers: dict[str, str]) -> httpx.Response:
|
||
async with _client(app) as client:
|
||
response = await client.get("/whoami", headers=headers)
|
||
assert response.status_code == 401 # noqa: PLR2004
|
||
assert response.headers["www-authenticate"] == "Bearer"
|
||
assert isinstance(response.json()["detail"], str)
|
||
return response
|
||
|
||
|
||
async def test_401_without_header_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
|
||
app, touched = untouchable_app
|
||
response = await _expect_401(app, {})
|
||
assert "Authorization: Bearer" in response.json()["detail"]
|
||
assert touched == []
|
||
|
||
|
||
async def test_401_not_bearer_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
|
||
app, touched = untouchable_app
|
||
await _expect_401(app, {"Authorization": "Basic dXNlcjpwYXNz"})
|
||
assert touched == []
|
||
|
||
|
||
async def test_401_garbage_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
|
||
app, touched = untouchable_app
|
||
response = await _expect_401(app, _bearer("not-a-jwt"))
|
||
assert response.json() == {"detail": "некорректный токен"}
|
||
assert touched == []
|
||
|
||
|
||
async def test_401_expired_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
|
||
app, touched = untouchable_app
|
||
token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local", ttl=timedelta(minutes=-5))
|
||
response = await _expect_401(app, _bearer(token))
|
||
assert response.json() == {"detail": "срок действия токена истёк"}
|
||
assert touched == []
|
||
|
||
|
||
async def test_401_foreign_signature_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
|
||
app, touched = untouchable_app
|
||
token = DevHs256Verifier(OTHER_SECRET).issue_token("ka_go@armka.local")
|
||
response = await _expect_401(app, _bearer(token))
|
||
assert response.json() == {"detail": "недействительная подпись токена"}
|
||
assert touched == []
|
||
|
||
|
||
async def test_401_when_verifier_not_configured() -> None:
|
||
touched: list[str] = []
|
||
|
||
async def session() -> AsyncIterator[_ForbiddenSession]:
|
||
touched.append("session")
|
||
yield _ForbiddenSession()
|
||
|
||
app = _build_app(session, None)
|
||
token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local")
|
||
response = await _expect_401(app, _bearer(token))
|
||
assert response.json() == {"detail": "аутентификация не настроена"}
|
||
assert touched == []
|
||
|
||
|
||
def test_get_token_verifier_builds_lazily_and_caches(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Проверяющий строится один раз на приложение (в том числе `None`) и кладётся в `app.state.token_verifier`."""
|
||
from types import SimpleNamespace # noqa: PLC0415
|
||
|
||
calls: list[object] = []
|
||
sentinel = DevHs256Verifier(SECRET)
|
||
|
||
def fake_build(settings: object) -> DevHs256Verifier:
|
||
calls.append(settings)
|
||
return sentinel
|
||
|
||
monkeypatch.setattr("src.api.dependencies.auth.build_verifier", fake_build)
|
||
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()))
|
||
assert get_token_verifier(request) is sentinel # type: ignore[arg-type]
|
||
assert get_token_verifier(request) is sentinel # type: ignore[arg-type]
|
||
assert len(calls) == 1
|
||
assert request.app.state.token_verifier is sentinel
|
||
|
||
monkeypatch.setattr("src.api.dependencies.auth.build_verifier", lambda settings: None)
|
||
request_none = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()))
|
||
assert get_token_verifier(request_none) is None # type: ignore[arg-type]
|
||
assert request_none.app.state.token_verifier is None
|
||
|
||
|
||
# --- статическая проверка: код роли в логике не используется ------------------------------------------------------
|
||
|
||
|
||
def _non_docstring_string_constants(tree: ast.AST) -> set[str]:
|
||
"""Строковые литералы модуля, кроме docstring'ов (модуля, классов, функций)."""
|
||
docstrings: set[int] = set()
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.Module | ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef):
|
||
first = node.body[0] if node.body else None
|
||
if isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant):
|
||
docstrings.add(id(first.value))
|
||
return {
|
||
node.value
|
||
for node in ast.walk(tree)
|
||
if isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in docstrings
|
||
}
|
||
|
||
|
||
def test_role_codes_are_not_used_in_auth_logic() -> None:
|
||
"""В `src/api/**` и `src/services/**` нет строковых констант — кодов ролей (права — данные; T17.4)."""
|
||
roles = yaml.safe_load((SEED_CORE / "roles.yaml").read_text(encoding="utf-8"))
|
||
role_codes = {item["code"] for item in roles}
|
||
assert {"KA_RF", "KA_GO", "EXPERT"} <= role_codes
|
||
files = [
|
||
*sorted((SRC / "api").rglob("*.py")),
|
||
*sorted((SRC / "services").rglob("*.py")),
|
||
SRC / "repositories" / "users.py",
|
||
]
|
||
assert SRC / "api" / "dependencies" / "auth.py" in files
|
||
assert SRC / "api" / "endpoints" / "me.py" in files
|
||
for path in files:
|
||
literals = _non_docstring_string_constants(ast.parse(path.read_text(encoding="utf-8")))
|
||
assert not literals & role_codes, f"{path.relative_to(SRC)}: код роли в логике: {sorted(literals & role_codes)}"
|
||
|
||
|
||
def test_role_code_scan_detects_literal() -> None:
|
||
"""Сканер видит код роли в обычной константе и пропускает docstring."""
|
||
tree = ast.parse('"""KA_RF в docstring модуля."""\n\ndef f():\n """KA_GO в docstring."""\n return "EXPERT"\n')
|
||
assert _non_docstring_string_constants(tree) == {"EXPERT"}
|
||
|
||
|
||
# --- интеграционные тесты на seed -------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
async def seeded(db_session: AsyncSession) -> AsyncSession:
|
||
await run_seed(db_session, dev=True)
|
||
await db_session.flush()
|
||
return db_session
|
||
|
||
|
||
@pytest.fixture
|
||
async def app(seeded: AsyncSession) -> FastAPI:
|
||
async def session() -> AsyncIterator[AsyncSession]:
|
||
yield seeded
|
||
|
||
return _build_app(session, DevHs256Verifier(SECRET))
|
||
|
||
|
||
@pytest.fixture
|
||
def token_for() -> Callable[[str], dict[str, str]]:
|
||
verifier = DevHs256Verifier(SECRET)
|
||
|
||
def make(email: str) -> dict[str, str]:
|
||
return _bearer(verifier.issue_token(email))
|
||
|
||
return make
|
||
|
||
|
||
async def _add_user(session: AsyncSession, email: str, *, active: bool = True, roles: tuple[str, ...] = ()) -> AppUser:
|
||
office_id = (await session.execute(select(RegionalOffice.id).limit(1))).scalar_one()
|
||
user = AppUser(email=email, full_name="Тест Тестов", regional_office_id=office_id, is_active=active)
|
||
session.add(user)
|
||
await session.flush()
|
||
for code in roles:
|
||
role_id = (await session.execute(select(Role.id).where(Role.code == code))).scalar_one()
|
||
session.add(UserRole(user_id=user.id, role_id=role_id))
|
||
await session.flush()
|
||
return user
|
||
|
||
|
||
def _seed_permissions(*role_codes: str) -> set[str]:
|
||
matrix = yaml.safe_load((SEED_CORE / "role_permissions.yaml").read_text(encoding="utf-8"))
|
||
return {code for role in role_codes for code in matrix[role]}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_permission_present(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
|
||
"""Сценарий «Право есть»: KA_GO, KFO_VERIFY."""
|
||
async with _client(app) as client:
|
||
response = await client.get("/kfo-verify", headers=token_for("ka_go@armka.local"))
|
||
assert response.status_code == 200 # noqa: PLR2004
|
||
assert response.json() == {"ok": True}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_permission_absent(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
|
||
"""Сценарий «Права нет»: EXPERT, DOCUMENT_EDIT → 403 с названием права."""
|
||
async with _client(app) as client:
|
||
response = await client.get("/doc-edit", headers=token_for("expert@armka.local"))
|
||
assert response.status_code == 403 # noqa: PLR2004
|
||
assert response.json() == {"detail": "недостаточно прав: требуется DOCUMENT_EDIT"}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_whoami_returns_user_office_roles_and_permissions(
|
||
app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/whoami", headers=token_for("ka_go@armka.local"))
|
||
assert response.status_code == 200 # noqa: PLR2004
|
||
body = response.json()
|
||
assert body["email"] == "ka_go@armka.local"
|
||
assert body["roles"] == ["KA_GO"]
|
||
assert body["permissions"] == sorted(_seed_permissions("KA_GO"))
|
||
assert body["office"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_two_roles_union_without_duplicates(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
"""Сценарий «Две роли»: KA_RF + CURATOR_GO → объединение прав без дубликатов."""
|
||
await _add_user(seeded, "two.roles@armka.local", roles=("KA_RF", "CURATOR_GO"))
|
||
ka_rf, curator = _seed_permissions("KA_RF"), _seed_permissions("CURATOR_GO")
|
||
assert ka_rf & curator, "тест осмыслен, только если у ролей есть общие права"
|
||
async with _client(app) as client:
|
||
response = await client.get("/whoami", headers=token_for("two.roles@armka.local"))
|
||
assert response.status_code == 200 # noqa: PLR2004
|
||
body = response.json()
|
||
assert body["roles"] == ["CURATOR_GO", "KA_RF"]
|
||
assert body["permissions"] == sorted(ka_rf | curator)
|
||
assert len(body["permissions"]) == len(set(body["permissions"]))
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_unknown_user_is_403(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/whoami", headers=token_for("nobody@armka.local"))
|
||
assert response.status_code == 403 # noqa: PLR2004
|
||
assert response.json() == {"detail": NOT_REGISTERED}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_inactive_user_is_403(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
await _add_user(seeded, "off@armka.local", active=False, roles=("KA_GO",))
|
||
async with _client(app) as client:
|
||
response = await client.get("/kfo-verify", headers=token_for("off@armka.local"))
|
||
assert response.status_code == 403 # noqa: PLR2004
|
||
assert response.json() == {"detail": DISABLED}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_email_case_variants_are_ambiguous_403(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
"""R-17: два пользователя с email, различающимися регистром → 403, а не «первый попавшийся»."""
|
||
await _add_user(seeded, "Ivan@x.local", roles=("KA_GO",))
|
||
await _add_user(seeded, "ivan@x.local", roles=("EXPERT",))
|
||
async with _client(app) as client:
|
||
response = await client.get("/whoami", headers=token_for("IVAN@x.local"))
|
||
assert response.status_code == 403 # noqa: PLR2004
|
||
assert response.json() == {"detail": AMBIGUOUS}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_token_email_case_insensitive(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/kfo-verify", headers=token_for("KA_GO@Armka.Local"))
|
||
assert response.status_code == 200 # noqa: PLR2004
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_user_email_stored_in_mixed_case_is_found(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
await _add_user(seeded, "Mixed.Case@Armka.Local", roles=("KA_GO",))
|
||
async with _client(app) as client:
|
||
response = await client.get("/kfo-verify", headers=token_for("mixed.case@armka.local"))
|
||
assert response.status_code == 200 # noqa: PLR2004
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_inactive_role_grants_nothing(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
await _add_user(seeded, "mix@armka.local", roles=("KA_GO", "EXPERT"))
|
||
await seeded.execute(update(Role).where(Role.code == "KA_GO").values(is_active=False))
|
||
async with _client(app) as client:
|
||
verify = await client.get("/kfo-verify", headers=token_for("mix@armka.local"))
|
||
who = await client.get("/whoami", headers=token_for("mix@armka.local"))
|
||
assert verify.status_code == 403 # noqa: PLR2004
|
||
assert who.json()["roles"] == ["EXPERT"]
|
||
assert who.json()["permissions"] == sorted(_seed_permissions("EXPERT"))
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_user_without_roles_has_no_permissions(
|
||
seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
|
||
) -> None:
|
||
await _add_user(seeded, "norole@armka.local")
|
||
async with _client(app) as client:
|
||
who = await client.get("/whoami", headers=token_for("norole@armka.local"))
|
||
verify = await client.get("/kfo-verify", headers=token_for("norole@armka.local"))
|
||
assert who.status_code == 200 # noqa: PLR2004
|
||
assert who.json()["roles"] == []
|
||
assert who.json()["permissions"] == []
|
||
assert verify.status_code == 403 # noqa: PLR2004
|
||
|
||
|
||
@pytest.fixture
|
||
def sql_log(db_engine: AsyncEngine) -> Iterator[list[str]]:
|
||
"""Все SQL-выражения, выполненные через движок тестовой БД с момента запроса фикстуры."""
|
||
statements: list[str] = []
|
||
|
||
def record(_conn: Any, _cursor: Any, statement: str, *_rest: Any) -> None:
|
||
statements.append(statement)
|
||
|
||
event.listen(db_engine.sync_engine, "before_cursor_execute", record)
|
||
try:
|
||
yield statements
|
||
finally:
|
||
event.remove(db_engine.sync_engine, "before_cursor_execute", record)
|
||
|
||
|
||
def _principal_queries(statements: list[str]) -> list[str]:
|
||
return [s for s in statements if "app_user" in s]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_db_queried_once_per_request(
|
||
app: FastAPI, token_for: Callable[[str], dict[str, str]], sql_log: list[str]
|
||
) -> None:
|
||
"""Два `require` и `current_user` на одном маршруте — один запрос к пользователю и правам."""
|
||
async with _client(app) as client:
|
||
single = await client.get("/whoami", headers=token_for("ka_go@armka.local"))
|
||
assert single.status_code == 200 # noqa: PLR2004
|
||
queries_single = len(_principal_queries(sql_log))
|
||
sql_log.clear()
|
||
combined = await client.get("/both", headers=token_for("ka_go@armka.local"))
|
||
assert combined.status_code == 200 # noqa: PLR2004
|
||
assert queries_single == 1
|
||
assert len(_principal_queries(sql_log)) == 1
|
||
assert len(sql_log) == 1, sql_log
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_both_requires_checked_on_one_route(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/both", headers=token_for("expert@armka.local"))
|
||
assert response.status_code == 403 # noqa: PLR2004
|
||
assert response.json() == {"detail": "недостаточно прав: требуется KFO_VERIFY"}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_repository_load_principal_data(seeded: AsyncSession) -> None:
|
||
assert await load_principal_data(seeded, "nobody@armka.local") is None
|
||
data = await load_principal_data(seeded, "ka_go@armka.local")
|
||
assert data is not None
|
||
assert data.user.email == "ka_go@armka.local"
|
||
assert data.roles == ("KA_GO",)
|
||
assert data.permissions == frozenset(_seed_permissions("KA_GO"))
|
||
assert data.regional_office.id == data.user.regional_office_id
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_repository_raises_on_ambiguity(seeded: AsyncSession) -> None:
|
||
await _add_user(seeded, "Dup@armka.local", roles=("EXPERT",))
|
||
await _add_user(seeded, "dup@armka.local", roles=("KA_GO",))
|
||
with pytest.raises(AmbiguousUserError):
|
||
await load_principal_data(seeded, "dup@armka.local")
|
||
|
||
|
||
def test_me_response_schema_serializes_enum_and_null_segment() -> None:
|
||
office = RegionalOfficeOut(id=uuid.uuid4(), name="ГО", is_head_office=True)
|
||
fields: dict[str, Any] = {
|
||
"id": uuid.uuid4(),
|
||
"email": "a@b.c",
|
||
"full_name": "А",
|
||
"regional_office": office,
|
||
"roles": ["KA_GO"],
|
||
"permissions": ["KFO_VIEW"],
|
||
}
|
||
empty = MeResponse(business_segment=None, **fields)
|
||
assert empty.model_dump(mode="json")["business_segment"] is None
|
||
segment = next(iter(BusinessSegment))
|
||
filled = MeResponse(business_segment=segment, **fields)
|
||
assert filled.model_dump()["business_segment"] == str(segment)
|
||
assert type(filled.model_dump()["business_segment"]) is str
|