feat: GET /api/v1/me и POST /api/v1/dev/token, условный build_api_router (T17.4)
Dev-эмитент подключается только при AUTH_DEV_ISSUER и DEBUG; статический тест «нет кодов ролей» расширен на src/api/** и src/services/**. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
cd64087b0d
commit
f5a5bbf95f
70
api/src/api/endpoints/dev_auth.py
Normal file
70
api/src/api/endpoints/dev_auth.py
Normal file
@ -0,0 +1,70 @@
|
||||
"""`POST /api/v1/dev/token` — выпуск dev-токена по email (только разработка; spec authn-authz, D-5).
|
||||
|
||||
Маршрут существует лишь при `AUTH_DEV_ISSUER=true` **и** `DEBUG=true`: роутер собирается фабрикой
|
||||
:func:`build_dev_router`, которую `build_api_router` вызывает только при обоих флагах. При любом другом сочетании
|
||||
маршрута нет вовсе (404 «Not Found» от самого приложения).
|
||||
"""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.core.auth import DevHs256Verifier
|
||||
from src.core.config import Settings
|
||||
from src.db.db import get_async_session
|
||||
from src.repositories.users import AmbiguousUserError, load_principal_data
|
||||
|
||||
USER_NOT_FOUND = "пользователь не найден или отключён"
|
||||
USER_AMBIGUOUS = "неоднозначная учётная запись"
|
||||
|
||||
|
||||
class DevTokenRequest(BaseModel):
|
||||
"""Тело запроса: email существующего активного пользователя (регистр и пробелы по краям не важны, R-17).
|
||||
|
||||
Строка, а не `EmailStr`: email-validator отвергает зарезервированный домен `.local` dev-пользователей.
|
||||
"""
|
||||
|
||||
email: Annotated[str, Field(min_length=1)]
|
||||
|
||||
@field_validator("email")
|
||||
@classmethod
|
||||
def _normalize(cls, value: str) -> str:
|
||||
value = value.strip().lower()
|
||||
if not value:
|
||||
raise ValueError("email не может быть пустым")
|
||||
return value
|
||||
|
||||
|
||||
class DevTokenResponse(BaseModel):
|
||||
"""Ответ: JWT для заголовка `Authorization: Bearer`."""
|
||||
|
||||
access_token: str
|
||||
token_type: str = "bearer" # noqa: S105 - тип схемы авторизации OAuth2, не секрет
|
||||
|
||||
|
||||
def build_dev_router(settings: Settings) -> APIRouter:
|
||||
"""Роутер dev-эмитента. Токен подписывается тем же секретом и пишет email в тот же claim, что проверяет
|
||||
dev-проверяющий приложения (`build_verifier`): оба строятся из одних настроек (`AUTH_DEV_SECRET`,
|
||||
`AUTH_EMAIL_CLAIM`).
|
||||
"""
|
||||
if not settings.AUTH_DEV_SECRET:
|
||||
raise RuntimeError("AUTH_DEV_ISSUER=true и DEBUG=true, но AUTH_DEV_SECRET не задан")
|
||||
issuer = DevHs256Verifier(settings.AUTH_DEV_SECRET, email_claim=settings.AUTH_EMAIL_CLAIM)
|
||||
router = APIRouter(tags=["dev"])
|
||||
|
||||
@router.post("/dev/token", response_model=DevTokenResponse, summary="Dev-токен по email (только DEBUG)")
|
||||
async def issue_dev_token(
|
||||
body: DevTokenRequest, session: Annotated[AsyncSession, Depends(get_async_session)]
|
||||
) -> DevTokenResponse:
|
||||
"""HS256 JWT на 8 часов для существующего активного пользователя."""
|
||||
try:
|
||||
principal = await load_principal_data(session, body.email)
|
||||
except AmbiguousUserError as error:
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, detail=USER_AMBIGUOUS) from error
|
||||
if principal is None or not principal.user.is_active:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=USER_NOT_FOUND)
|
||||
return DevTokenResponse(access_token=issuer.issue_token(body.email))
|
||||
|
||||
return router
|
||||
31
api/src/api/endpoints/me.py
Normal file
31
api/src/api/endpoints/me.py
Normal file
@ -0,0 +1,31 @@
|
||||
"""`GET /api/v1/me` — профиль, роли и эффективные права текущего пользователя (spec authn-authz).
|
||||
|
||||
Единственный маршрут Среза 0 без `require(...)`: нужен любой действующий пользователь (`current_user`), права
|
||||
не проверяются — фронтенд по ответу решает, что показывать.
|
||||
"""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from src.api.dependencies.auth import Principal, current_user
|
||||
from src.schemas.me import MeResponse, RegionalOfficeOut
|
||||
|
||||
router = APIRouter(tags=["me"])
|
||||
|
||||
|
||||
@router.get("/me", response_model=MeResponse, summary="Текущий пользователь, его роли и права")
|
||||
async def get_me(principal: Annotated[Principal, Depends(current_user)]) -> MeResponse:
|
||||
"""Профиль, филиал, коды ролей и эффективные права (коды, по алфавиту)."""
|
||||
user = principal.user
|
||||
office = principal.regional_office
|
||||
return MeResponse(
|
||||
id=user.id,
|
||||
# пользователь найден по email, поэтому он заполнен; пустая строка — лишь для типов (колонка nullable)
|
||||
email=user.email or "",
|
||||
full_name=user.full_name,
|
||||
regional_office=RegionalOfficeOut(id=office.id, name=office.name, is_head_office=office.is_head_office),
|
||||
business_segment=user.business_segment,
|
||||
roles=sorted(principal.roles),
|
||||
permissions=sorted(principal.permissions),
|
||||
)
|
||||
@ -1,3 +1,25 @@
|
||||
"""Корневой роутер API `/api/v1` (R-6: эндпоинты — `src/api/endpoints/`).
|
||||
|
||||
`build_api_router(settings)` собирает роутер под настройки: `/me` — всегда, dev-эмитент токенов — только при
|
||||
`AUTH_DEV_ISSUER and DEBUG`. Модульный `router` собран по настройкам процесса и подключается в `src/app.py`.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter(prefix="/api/v1")
|
||||
from src.api.endpoints import me
|
||||
from src.api.endpoints.dev_auth import build_dev_router
|
||||
from src.core.config import Settings, get_settings
|
||||
|
||||
API_PREFIX = "/api/v1"
|
||||
|
||||
|
||||
def build_api_router(settings: Settings) -> APIRouter:
|
||||
"""Роутер `/api/v1` под настройки; dev-маршруты при любом другом сочетании флагов отсутствуют."""
|
||||
api_router = APIRouter(prefix=API_PREFIX)
|
||||
api_router.include_router(me.router)
|
||||
if settings.AUTH_DEV_ISSUER and settings.DEBUG:
|
||||
api_router.include_router(build_dev_router(settings))
|
||||
return api_router
|
||||
|
||||
|
||||
router = build_api_router(get_settings())
|
||||
|
||||
@ -214,17 +214,42 @@ def test_get_token_verifier_builds_lazily_and_caches(monkeypatch: pytest.MonkeyP
|
||||
# --- статическая проверка: код роли в логике не используется ------------------------------------------------------
|
||||
|
||||
|
||||
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:
|
||||
"""В `dependencies/auth.py` и `repositories/users.py` нет строковых констант — кодов ролей (права — данные)."""
|
||||
"""В `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
|
||||
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)}"
|
||||
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 -------------------------------------------------------------------------------
|
||||
|
||||
364
api/tests/test_me_and_dev_token.py
Normal file
364
api/tests/test_me_and_dev_token.py
Normal file
@ -0,0 +1,364 @@
|
||||
"""`GET /api/v1/me` и `POST /api/v1/dev/token` (spec authn-authz, задача 18).
|
||||
|
||||
Мини-приложения собираются в тесте на `build_api_router(settings)` под разные сочетания `DEBUG` / `AUTH_DEV_ISSUER`
|
||||
(без lifespan, сессия — `db_session` с откатом); отдельный тест проходит через полный `src.app.app`.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
from collections.abc import AsyncIterator
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import jwt
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from scripts.seed import run_seed
|
||||
from src.api.dependencies.auth import current_user
|
||||
from src.api.router import build_api_router, router
|
||||
from src.core.auth import DevHs256Verifier, build_verifier
|
||||
from src.core.config import Settings, get_settings
|
||||
from src.core.exception_handlers import register_exception_handlers
|
||||
from src.db.db import get_async_session
|
||||
from src.models.access import AppUser
|
||||
from src.schemas.me import MeResponse
|
||||
|
||||
HTTP_OK = 200
|
||||
HTTP_UNAUTHORIZED = 401
|
||||
HTTP_FORBIDDEN = 403
|
||||
HTTP_NOT_FOUND = 404
|
||||
HTTP_UNPROCESSABLE = 422
|
||||
USER_NOT_FOUND = {"detail": "пользователь не найден или отключён"}
|
||||
|
||||
# Матрица §10 (docs/context/05-domain-rules.md), записанная независимо от seed: коды прав по ролям.
|
||||
KA_RF_PERMISSIONS = sorted(
|
||||
{
|
||||
"DIRECTORY_VIEW",
|
||||
"DIRECTORY_EDIT",
|
||||
"DOCUMENT_VIEW",
|
||||
"DOCUMENT_EDIT",
|
||||
"CONSOLIDATION_VIEW",
|
||||
"CONSOLIDATION_EDIT",
|
||||
"KFO_VIEW",
|
||||
"KFO_EXPORT",
|
||||
"EX_VIEW",
|
||||
"EXPORT",
|
||||
}
|
||||
)
|
||||
ADMIN_PERMISSIONS = sorted(
|
||||
{"DIRECTORY_VIEW", "DIRECTORY_EDIT", "DIRECTORY_DELETE", "DIRECTORY_ADMIN", "EX_VIEW", "USERS_ADMIN", "AUDIT_VIEW"}
|
||||
)
|
||||
assert len(KA_RF_PERMISSIONS) == 10 # noqa: PLR2004,S101 - матрица §10 для KA_RF
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _quiet_auth_log(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Предупреждение «включён dev-эмитент» из `build_verifier` в выводе тестов не нужно."""
|
||||
monkeypatch.setattr("src.core.auth.logger", SimpleNamespace(warning=lambda *args, **kwargs: None))
|
||||
|
||||
|
||||
def _settings(*, debug: bool, dev_issuer: bool) -> Settings:
|
||||
"""Настройки из `.env.test` с переопределением флагов (секрет dev-эмитента — из окружения теста)."""
|
||||
return get_settings().model_copy(update={"DEBUG": debug, "AUTH_DEV_ISSUER": dev_issuer})
|
||||
|
||||
|
||||
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 _build_app(session: AsyncSession, settings: Settings) -> FastAPI:
|
||||
"""Мини-приложение на маршрутах `build_api_router(settings)`; проверяющий токены — по тем же настройкам."""
|
||||
app = FastAPI()
|
||||
register_exception_handlers(app)
|
||||
app.include_router(build_api_router(settings))
|
||||
|
||||
async def override_session() -> AsyncIterator[AsyncSession]:
|
||||
yield session
|
||||
|
||||
app.dependency_overrides[get_async_session] = override_session
|
||||
app.state.token_verifier = build_verifier(settings)
|
||||
return app
|
||||
|
||||
|
||||
@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 dev_app(seeded: AsyncSession) -> FastAPI:
|
||||
return _build_app(seeded, _settings(debug=True, dev_issuer=True))
|
||||
|
||||
|
||||
async def _dev_token(client: httpx.AsyncClient, email: str) -> str:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": email})
|
||||
assert response.status_code == HTTP_OK, response.text
|
||||
return str(response.json()["access_token"])
|
||||
|
||||
|
||||
# --- GET /me ------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_for_ka_rf_matches_matrix(dev_app: FastAPI, seeded: AsyncSession) -> None:
|
||||
"""Сценарий «Ответ для dev-пользователя»: roles = ["KA_RF"], permissions = права KA_RF по §10, отсортированы."""
|
||||
async with _client(dev_app) as client:
|
||||
token = await _dev_token(client, "ka_rf@armka.local")
|
||||
response = await client.get("/api/v1/me", headers=_bearer(token))
|
||||
assert response.status_code == HTTP_OK, response.text
|
||||
body = response.json()
|
||||
assert body["roles"] == ["KA_RF"]
|
||||
assert body["permissions"] == KA_RF_PERMISSIONS
|
||||
assert body["permissions"] == sorted(body["permissions"])
|
||||
user = (await seeded.execute(select(AppUser).where(AppUser.email == "ka_rf@armka.local"))).scalar_one()
|
||||
assert set(body) == {"id", "email", "full_name", "regional_office", "business_segment", "roles", "permissions"}
|
||||
assert body["id"] == str(user.id)
|
||||
assert body["email"] == "ka_rf@armka.local"
|
||||
assert body["full_name"] == user.full_name
|
||||
assert set(body["regional_office"]) == {"id", "name", "is_head_office"}
|
||||
assert body["regional_office"]["id"] == str(user.regional_office_id)
|
||||
assert isinstance(body["regional_office"]["name"], str)
|
||||
assert isinstance(body["regional_office"]["is_head_office"], bool)
|
||||
expected_segment = user.business_segment.value if user.business_segment is not None else None
|
||||
assert body["business_segment"] == expected_segment
|
||||
# ответ укладывается в схему
|
||||
MeResponse.model_validate(body)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_for_admin(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
token = await _dev_token(client, "admin@armka.local")
|
||||
response = await client.get("/api/v1/me", headers=_bearer(token))
|
||||
assert response.status_code == HTTP_OK
|
||||
body = response.json()
|
||||
assert body["roles"] == ["ADMIN"]
|
||||
assert body["permissions"] == ADMIN_PERMISSIONS
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_head_office_flag(dev_app: FastAPI) -> None:
|
||||
"""У пользователя головного офиса `is_head_office` = true, у филиала — false (по данным seed)."""
|
||||
async with _client(dev_app) as client:
|
||||
flags = {}
|
||||
for email in ("ka_go@armka.local", "ka_rf@armka.local"):
|
||||
token = await _dev_token(client, email)
|
||||
body = (await client.get("/api/v1/me", headers=_bearer(token))).json()
|
||||
flags[email] = body["regional_office"]["is_head_office"]
|
||||
assert flags["ka_go@armka.local"] is True
|
||||
assert flags["ka_rf@armka.local"] is False
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_accepts_token_in_any_email_case(dev_app: FastAPI) -> None:
|
||||
verifier = DevHs256Verifier(get_settings().AUTH_DEV_SECRET)
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.get("/api/v1/me", headers=_bearer(verifier.issue_token("KA_RF@Armka.Local")))
|
||||
assert response.status_code == HTTP_OK
|
||||
assert response.json()["email"] == "ka_rf@armka.local"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_without_token_is_401(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.get("/api/v1/me")
|
||||
assert response.status_code == HTTP_UNAUTHORIZED
|
||||
assert response.headers["www-authenticate"] == "Bearer"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_unknown_user_is_403(dev_app: FastAPI) -> None:
|
||||
verifier = DevHs256Verifier(get_settings().AUTH_DEV_SECRET)
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.get("/api/v1/me", headers=_bearer(verifier.issue_token("nobody@armka.local")))
|
||||
assert response.status_code == HTTP_FORBIDDEN
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_me_user_without_roles_has_empty_lists(dev_app: FastAPI, seeded: AsyncSession) -> None:
|
||||
office_id = (await seeded.execute(select(AppUser.regional_office_id).limit(1))).scalar_one()
|
||||
seeded.add(AppUser(email="norole@armka.local", full_name="Без Ролей", regional_office_id=office_id))
|
||||
await seeded.flush()
|
||||
verifier = DevHs256Verifier(get_settings().AUTH_DEV_SECRET)
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.get("/api/v1/me", headers=_bearer(verifier.issue_token("norole@armka.local")))
|
||||
assert response.status_code == HTTP_OK
|
||||
assert response.json()["roles"] == []
|
||||
assert response.json()["permissions"] == []
|
||||
|
||||
|
||||
def test_me_route_has_no_permission_requirement() -> None:
|
||||
"""`/me` — единственный маршрут Среза 0 без `require`: в зависимостях только `current_user`."""
|
||||
router = build_api_router(_settings(debug=False, dev_issuer=False))
|
||||
route = next(r for r in router.routes if getattr(r, "path", "") == "/api/v1/me")
|
||||
deps = [d.call for d in route.dependant.dependencies] # type: ignore[attr-defined]
|
||||
assert deps == [current_user]
|
||||
|
||||
|
||||
# --- POST /dev/token ----------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_for_admin(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": "admin@armka.local"})
|
||||
assert response.status_code == HTTP_OK
|
||||
body = response.json()
|
||||
assert set(body) == {"access_token", "token_type"}
|
||||
assert body["token_type"] == "bearer"
|
||||
me = await client.get("/api/v1/me", headers=_bearer(body["access_token"]))
|
||||
assert me.status_code == HTTP_OK
|
||||
assert me.json()["roles"] == ["ADMIN"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_is_hs256_for_8_hours(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
token = await _dev_token(client, "ka_rf@armka.local")
|
||||
header = jwt.get_unverified_header(token)
|
||||
payload = jwt.decode(token, options={"verify_signature": False})
|
||||
assert header["alg"] == "HS256"
|
||||
assert payload["email"] == "ka_rf@armka.local"
|
||||
assert payload["exp"] - payload["iat"] == 8 * 3600
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_email_case_insensitive(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
token = await _dev_token(client, " KA_RF@Armka.Local ")
|
||||
me = await client.get("/api/v1/me", headers=_bearer(token))
|
||||
assert me.status_code == HTTP_OK
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_unknown_user_is_404(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": "nobody@armka.local"})
|
||||
assert response.status_code == HTTP_NOT_FOUND
|
||||
assert response.json() == USER_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_inactive_user_is_404(dev_app: FastAPI, seeded: AsyncSession) -> None:
|
||||
user = (await seeded.execute(select(AppUser).where(AppUser.email == "expert@armka.local"))).scalar_one()
|
||||
user.is_active = False
|
||||
await seeded.flush()
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": "expert@armka.local"})
|
||||
assert response.status_code == HTTP_NOT_FOUND
|
||||
assert response.json() == USER_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_ambiguous_email_is_error(dev_app: FastAPI, seeded: AsyncSession) -> None:
|
||||
"""Два пользователя с email, различающимся регистром, — отказ (fail-closed), токен не выпускается."""
|
||||
office_id = (await seeded.execute(select(AppUser.regional_office_id).limit(1))).scalar_one()
|
||||
seeded.add(AppUser(email="KA_RF@armka.local", full_name="Двойник", regional_office_id=office_id))
|
||||
await seeded.flush()
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": "ka_rf@armka.local"})
|
||||
assert response.status_code == 409 # noqa: PLR2004
|
||||
assert response.json() == {"detail": "неоднозначная учётная запись"}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(
|
||||
"body", [{}, {"email": ""}, {"email": " "}, {"email": 5}, {"email": None}, {"mail": "a@b.c"}]
|
||||
)
|
||||
async def test_dev_token_invalid_body_is_422(dev_app: FastAPI, body: dict[str, Any]) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json=body)
|
||||
assert response.status_code == HTTP_UNPROCESSABLE
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dev_token_not_json_is_422(dev_app: FastAPI) -> None:
|
||||
async with _client(dev_app) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/dev/token", content=b"not-json", headers={"content-type": "application/json"}
|
||||
)
|
||||
assert response.status_code == HTTP_UNPROCESSABLE
|
||||
|
||||
|
||||
# --- прод-конфигурация --------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(
|
||||
("debug", "dev_issuer"),
|
||||
[(False, False), (False, True), (True, False)],
|
||||
ids=["prod", "issuer-without-debug", "debug-without-issuer"],
|
||||
)
|
||||
async def test_dev_token_route_absent_unless_both_flags(seeded: AsyncSession, *, debug: bool, dev_issuer: bool) -> None:
|
||||
app = _build_app(seeded, _settings(debug=debug, dev_issuer=dev_issuer))
|
||||
async with _client(app) as client:
|
||||
response = await client.post("/api/v1/dev/token", json={"email": "ka_rf@armka.local"})
|
||||
# маршрут именно отсутствует (а не отвечает 404 по своей логике)
|
||||
assert response.json() == {"detail": "Not Found"}
|
||||
assert response.status_code == HTTP_NOT_FOUND
|
||||
assert "/api/v1/dev/token" not in {getattr(r, "path", None) for r in app.routes}
|
||||
assert "/api/v1/me" in {getattr(r, "path", None) for r in app.routes}
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize(("debug", "dev_issuer"), [(False, False), (False, True), (True, False)])
|
||||
async def test_dev_secret_token_rejected_unless_both_flags(
|
||||
seeded: AsyncSession, *, debug: bool, dev_issuer: bool
|
||||
) -> None:
|
||||
"""Токен, подписанный dev-секретом, при любом сочетании флагов, кроме «оба», → 401 (проверяющего нет/JWKS)."""
|
||||
settings = _settings(debug=debug, dev_issuer=dev_issuer)
|
||||
token = DevHs256Verifier(settings.AUTH_DEV_SECRET).issue_token("ka_rf@armka.local")
|
||||
app = _build_app(seeded, settings)
|
||||
async with _client(app) as client:
|
||||
response = await client.get("/api/v1/me", headers=_bearer(token))
|
||||
assert response.status_code == HTTP_UNAUTHORIZED
|
||||
|
||||
|
||||
# --- полное приложение --------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_full_app_me_dev_token_and_openapi(seeded: AsyncSession) -> None:
|
||||
app_module = importlib.import_module("src.app")
|
||||
full_app: FastAPI = app_module.app
|
||||
|
||||
async def override_session() -> AsyncIterator[AsyncSession]:
|
||||
yield seeded
|
||||
|
||||
full_app.dependency_overrides[get_async_session] = override_session
|
||||
# проверяющий приложения строится лениво по `.env.test`; сбрасываем возможный кэш предыдущих тестов
|
||||
if hasattr(full_app.state, "token_verifier"):
|
||||
del full_app.state.token_verifier
|
||||
try:
|
||||
async with _client(full_app) as client:
|
||||
token = await _dev_token(client, "ka_rf@armka.local")
|
||||
me = await client.get("/api/v1/me", headers=_bearer(token))
|
||||
health = await client.get("/healthcheck")
|
||||
openapi = (await client.get("/openapi.json")).json()
|
||||
finally:
|
||||
full_app.dependency_overrides.pop(get_async_session, None)
|
||||
if hasattr(full_app.state, "token_verifier"):
|
||||
del full_app.state.token_verifier
|
||||
assert me.status_code == HTTP_OK
|
||||
assert me.json()["roles"] == ["KA_RF"]
|
||||
assert me.json()["permissions"] == KA_RF_PERMISSIONS
|
||||
assert health.status_code == HTTP_OK
|
||||
assert "uptime" in health.json()
|
||||
assert "/api/v1/me" in openapi["paths"]
|
||||
assert "/api/v1/dev/token" in openapi["paths"]
|
||||
assert "/healthcheck" in openapi["paths"]
|
||||
|
||||
|
||||
def test_module_router_matches_settings() -> None:
|
||||
"""Модульный `router` собран по настройкам процесса (`.env.test`: DEBUG и AUTH_DEV_ISSUER включены)."""
|
||||
paths = {getattr(r, "path", None) for r in router.routes}
|
||||
assert {"/api/v1/me", "/api/v1/dev/token"} <= paths
|
||||
Loading…
x
Reference in New Issue
Block a user