feat: зависимости current_user и require(code), репозиторий пользователей, схема me
Пользователь, филиал, роли и права читаются одним запросом; роль is_active=false прав не даёт; email без учёта регистра, неоднозначность — 403 (R-17). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
5ed52a7239
commit
11941caa20
109
api/src/api/dependencies/auth.py
Normal file
109
api/src/api/dependencies/auth.py
Normal file
@ -0,0 +1,109 @@
|
||||
"""Зависимости FastAPI: личность из JWT, пользователь из БД, право `require(code)` (spec authn-authz, D-5, D-10).
|
||||
|
||||
Цепочка: заголовок `Authorization: Bearer` → проверяющий токены (`TokenVerifier`) → ОДИН запрос в БД
|
||||
(`load_principal_data`) → :class:`Principal`. Зависимости кэшируются FastAPI в пределах запроса, поэтому любое число
|
||||
`require(...)` и `current_user` на одном маршруте даёт один запрос к БД.
|
||||
|
||||
Ответы: 401 — нет/невалидный токен (с `WWW-Authenticate: Bearer`); 403 — пользователь не найден, неоднозначен,
|
||||
отключён либо нет права. Тело — `{"detail": "<причина по-русски>"}` (форматирует обработчик `HTTPException`).
|
||||
Права проверяются только по кодам прав: код роли в логике не используется.
|
||||
"""
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, Header, HTTPException, Request, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.core.auth import AuthError, Claims, TokenVerifier, build_verifier
|
||||
from src.core.config import get_settings
|
||||
from src.db.db import get_async_session
|
||||
from src.repositories.users import AmbiguousUserError, UserWithAccess, load_principal_data
|
||||
|
||||
# Текущий пользователь вместе с филиалом, ролями и эффективными правами (то, что нужно `GET /me` и `require`)
|
||||
Principal = UserWithAccess
|
||||
|
||||
_BEARER = "Bearer"
|
||||
_UNAUTHORIZED_HEADERS = {"WWW-Authenticate": _BEARER}
|
||||
_MISSING_HEADER = "требуется заголовок Authorization: Bearer <токен>"
|
||||
_BEARER_PARTS = 2
|
||||
|
||||
|
||||
def _unauthorized(detail: str) -> HTTPException:
|
||||
return HTTPException(status.HTTP_401_UNAUTHORIZED, detail=detail, headers=dict(_UNAUTHORIZED_HEADERS))
|
||||
|
||||
|
||||
def _forbidden(detail: str) -> HTTPException:
|
||||
return HTTPException(status.HTTP_403_FORBIDDEN, detail=detail)
|
||||
|
||||
|
||||
def parse_bearer(authorization: str | None) -> str:
|
||||
"""Токен из значения заголовка `Authorization: Bearer <jwt>`; иначе 401 (схема без учёта регистра)."""
|
||||
if authorization is not None:
|
||||
parts = authorization.split()
|
||||
if len(parts) == _BEARER_PARTS and parts[0].lower() == _BEARER.lower():
|
||||
return parts[1]
|
||||
raise _unauthorized(_MISSING_HEADER)
|
||||
|
||||
|
||||
def get_token_verifier(request: Request) -> TokenVerifier | None:
|
||||
"""Проверяющий токены приложения: строится лениво по настройкам и кэшируется в `app.state.token_verifier`.
|
||||
|
||||
`None` — проверять нечем (ничего не настроено). В тестах переопределяется через `app.dependency_overrides`.
|
||||
"""
|
||||
state = request.app.state
|
||||
if not hasattr(state, "token_verifier"):
|
||||
state.token_verifier = build_verifier(get_settings())
|
||||
verifier: TokenVerifier | None = state.token_verifier
|
||||
return verifier
|
||||
|
||||
|
||||
async def verified_claims(
|
||||
verifier: Annotated[TokenVerifier | None, Depends(get_token_verifier)],
|
||||
authorization: Annotated[str | None, Header()] = None,
|
||||
) -> Claims:
|
||||
"""Claims проверенного токена. Не обращается к БД: все 401 решаются до неё."""
|
||||
token = parse_bearer(authorization)
|
||||
if verifier is None:
|
||||
raise _unauthorized("аутентификация не настроена")
|
||||
try:
|
||||
return await verifier.verify(token)
|
||||
except AuthError as error:
|
||||
raise _unauthorized(str(error)) from error
|
||||
|
||||
|
||||
async def current_user(
|
||||
claims: Annotated[Claims, Depends(verified_claims)],
|
||||
session: Annotated[AsyncSession, Depends(get_async_session)],
|
||||
) -> Principal:
|
||||
"""Пользователь по email из токена с филиалом, ролями и эффективными правами (один запрос к БД).
|
||||
|
||||
`Claims.email` уже нормализован (`strip().lower()`), сравнение в БД — по `lower(app_user.email)` (R-17).
|
||||
403: пользователя нет в АРМ; учётная запись неоднозначна (несколько email, различающихся регистром) или отключена.
|
||||
"""
|
||||
try:
|
||||
principal = await load_principal_data(session, claims.email)
|
||||
except AmbiguousUserError as error:
|
||||
raise _forbidden("неоднозначная учётная запись") from error
|
||||
if principal is None:
|
||||
raise _forbidden("пользователь не зарегистрирован в АРМ")
|
||||
if not principal.user.is_active:
|
||||
raise _forbidden("учётная запись отключена")
|
||||
return principal
|
||||
|
||||
|
||||
def require(code: str) -> Callable[..., Awaitable[Principal]]:
|
||||
"""Фабрика зависимости: пропускает, только если среди эффективных прав есть `code`; иначе 403 с названием права.
|
||||
|
||||
Код права не сверяется с enum на этапе создания (проверяется лишь непустота): права — данные. Роли не
|
||||
используются. Возвращает :class:`Principal`, так что зависимость можно использовать и как `current_user`.
|
||||
"""
|
||||
if not code or not code.strip():
|
||||
raise ValueError("код права для require() не задан")
|
||||
|
||||
async def check(principal: Annotated[Principal, Depends(current_user)]) -> Principal:
|
||||
if code not in principal.permissions:
|
||||
raise _forbidden(f"недостаточно прав: требуется {code}")
|
||||
return principal
|
||||
|
||||
return check
|
||||
73
api/src/repositories/users.py
Normal file
73
api/src/repositories/users.py
Normal file
@ -0,0 +1,73 @@
|
||||
"""Репозиторий пользователей: личность и эффективные права одним запросом (spec authn-authz, D-5, R-17).
|
||||
|
||||
`effective_permissions(user) = ∪ permission.code по всем активным ролям пользователя` (CLAUDE.md: права — данные).
|
||||
Код роли в проверках не участвует: роли возвращаются только для показа (`GET /me`).
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from sqlalchemy import and_, func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import aliased
|
||||
|
||||
from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
|
||||
from src.models.directories import RegionalOffice
|
||||
|
||||
# Достаточно двух строк, чтобы заметить неоднозначность; больше пользователей читать незачем
|
||||
_AMBIGUITY_PROBE_LIMIT = 2
|
||||
|
||||
|
||||
class AmbiguousUserError(Exception):
|
||||
"""По email найдено больше одного пользователя (email в БД уникален с учётом регистра, сравнение — без него)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UserWithAccess:
|
||||
"""Пользователь, его филиал, коды активных ролей и эффективные права (коды).
|
||||
|
||||
`roles` — отсортированный кортеж кодов активных ролей; `permissions` — объединение прав этих ролей без дубликатов.
|
||||
"""
|
||||
|
||||
user: AppUser
|
||||
regional_office: RegionalOffice
|
||||
roles: tuple[str, ...]
|
||||
permissions: frozenset[str]
|
||||
|
||||
|
||||
async def load_principal_data(session: AsyncSession, email: str) -> UserWithAccess | None:
|
||||
"""Пользователь по email (без учёта регистра) вместе с филиалом, ролями и правами — ОДИН SQL-запрос.
|
||||
|
||||
Возвращает `None`, если пользователя нет; бросает :class:`AmbiguousUserError`, если подходит больше одного
|
||||
(fail-closed, R-17). Неактивный пользователь возвращается как есть (решение — за вызывающим). Роль с
|
||||
`is_active=false` не даёт прав и в `roles` не попадает.
|
||||
"""
|
||||
matched = (
|
||||
select(AppUser)
|
||||
.where(func.lower(AppUser.email) == email.lower())
|
||||
.order_by(AppUser.id)
|
||||
.limit(_AMBIGUITY_PROBE_LIMIT)
|
||||
.subquery()
|
||||
)
|
||||
user = aliased(AppUser, matched)
|
||||
statement = (
|
||||
select(user, RegionalOffice, Role.code, Permission.code)
|
||||
.join(RegionalOffice, RegionalOffice.id == user.regional_office_id)
|
||||
.outerjoin(UserRole, UserRole.user_id == user.id)
|
||||
.outerjoin(Role, and_(Role.id == UserRole.role_id, Role.is_active.is_(True)))
|
||||
.outerjoin(RolePermission, RolePermission.role_id == Role.id)
|
||||
.outerjoin(Permission, Permission.id == RolePermission.permission_id)
|
||||
)
|
||||
rows = (await session.execute(statement)).all()
|
||||
if not rows:
|
||||
return None
|
||||
users = {row[0].id for row in rows}
|
||||
if len(users) > 1:
|
||||
raise AmbiguousUserError(email)
|
||||
roles = {row[2] for row in rows if row[2] is not None}
|
||||
permissions = frozenset(row[3] for row in rows if row[3] is not None)
|
||||
return UserWithAccess(
|
||||
user=rows[0][0],
|
||||
regional_office=rows[0][1],
|
||||
roles=tuple(sorted(roles)),
|
||||
permissions=permissions,
|
||||
)
|
||||
29
api/src/schemas/me.py
Normal file
29
api/src/schemas/me.py
Normal file
@ -0,0 +1,29 @@
|
||||
"""Схема ответа `GET /api/v1/me` (spec authn-authz): профиль, роли и эффективные права текущего пользователя."""
|
||||
|
||||
import uuid
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from src.models.enums import BusinessSegment
|
||||
|
||||
|
||||
class RegionalOfficeOut(BaseModel):
|
||||
"""Филиал пользователя."""
|
||||
|
||||
id: uuid.UUID
|
||||
name: str
|
||||
is_head_office: bool
|
||||
|
||||
|
||||
class MeResponse(BaseModel):
|
||||
"""Текущий пользователь. `roles` и `permissions` — коды, отсортированные по алфавиту."""
|
||||
|
||||
model_config = ConfigDict(use_enum_values=True)
|
||||
|
||||
id: uuid.UUID
|
||||
email: str
|
||||
full_name: str
|
||||
regional_office: RegionalOfficeOut
|
||||
business_segment: BusinessSegment | None
|
||||
roles: list[str]
|
||||
permissions: list[str]
|
||||
479
api/tests/test_auth_deps.py
Normal file
479
api/tests/test_auth_deps.py
Normal file
@ -0,0 +1,479 @@
|
||||
"""Зависимости аутентификации и прав (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 test_role_codes_are_not_used_in_auth_logic() -> None:
|
||||
"""В `dependencies/auth.py` и `repositories/users.py` нет строковых констант — кодов ролей (права — данные)."""
|
||||
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
|
||||
for relative in ("api/dependencies/auth.py", "repositories/users.py"):
|
||||
tree = ast.parse((SRC / relative).read_text(encoding="utf-8"))
|
||||
literals = {
|
||||
node.value for node in ast.walk(tree) if isinstance(node, ast.Constant) and isinstance(node.value, str)
|
||||
}
|
||||
assert not literals & role_codes, f"{relative}: код роли в логике: {sorted(literals & role_codes)}"
|
||||
|
||||
|
||||
# --- интеграционные тесты на 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
|
||||
Loading…
x
Reference in New Issue
Block a user