Max Ronzhin
84b06f78f1
Покрытие валидации/сохранения contact_email, простановки Reply-To в письмах регистрации/приглашений/саммари при включённом и выключенном контактном адресе, и эндпоинта тестовой отправки (успех, дефолтный получатель, сбой транспорта без утечки логина/пароля).
469 lines
18 KiB
Python
469 lines
18 KiB
Python
"""Интеграционные тесты для `/api/v1/auth`: полный цикл регистрация → вход,
|
||
ротация/reuse refresh-токена, закрытые эндпоинты без JWT.
|
||
"""
|
||
|
||
import re
|
||
import uuid
|
||
from collections.abc import AsyncGenerator, Sequence
|
||
from typing import Annotated
|
||
|
||
import httpx
|
||
import pytest_asyncio
|
||
from fastapi import Depends, FastAPI
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from api.auth import get_auth_service
|
||
from core.db import get_session
|
||
from core.redis import redis_client
|
||
from models.team import Team
|
||
from models.user import User
|
||
from services.auth import AuthService
|
||
from services.email import EmailAttachment
|
||
from services.instance_settings import InstanceSettingsService, SettingsUpdateIn
|
||
|
||
REFRESH_COOKIE = "refresh_token"
|
||
|
||
|
||
@pytest_asyncio.fixture(autouse=True)
|
||
async def _reset_registration_gating(db_session: AsyncSession) -> None:
|
||
"""Сбросить перед каждым тестом настройки, влияющие на публичную регистрацию.
|
||
|
||
`registration_team_choice`/`registration_email_domain_enabled` — общая
|
||
dev-БД, строки могут быть легитимными данными разработчика (включены
|
||
вручную через админку) — тесты, которым это не важно, не должны зависеть
|
||
от текущего состояния общей БД. Тесты, которым важно обратное, переключают
|
||
настройку сами внутри тела теста уже после этого сброса (та же
|
||
транзакция — `db_session` никогда не коммитится в реальную БД, см.
|
||
`conftest.py`).
|
||
"""
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(registration_team_choice=False, registration_email_domain_enabled=False)
|
||
)
|
||
await db_session.commit()
|
||
|
||
|
||
class _CapturingEmailBackend:
|
||
"""Тестовый `EmailBackend`: сохраняет отправленные письма в памяти вместо лога."""
|
||
|
||
def __init__(self) -> None:
|
||
self.sent: list[tuple[str, str, str]] = []
|
||
self.reply_to: list[str | None] = []
|
||
|
||
async def send(
|
||
self,
|
||
*,
|
||
to: str,
|
||
subject: str,
|
||
body: str,
|
||
html_body: str | None = None,
|
||
attachments: Sequence[EmailAttachment] = (),
|
||
reply_to: str | None = None,
|
||
) -> None:
|
||
self.sent.append((to, subject, body))
|
||
self.reply_to.append(reply_to)
|
||
|
||
|
||
def _extract_verification_token(email_body: str) -> str:
|
||
match = re.search(r"token=([\w\-]+)", email_body)
|
||
assert match is not None, f"токен подтверждения не найден в письме: {email_body!r}"
|
||
return match.group(1)
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def email_backend(app: FastAPI) -> AsyncGenerator[_CapturingEmailBackend, None]:
|
||
"""Подменить email-бэкенд `AuthService` на перехватывающий (для чтения ссылки verify).
|
||
|
||
`app` — свежий экземпляр на каждый тест (фикстура function-scoped), поэтому
|
||
override не нужно откатывать вручную.
|
||
"""
|
||
backend = _CapturingEmailBackend()
|
||
|
||
async def _get_auth_service_override(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> AuthService:
|
||
return AuthService(session=session, redis=redis_client, email_backend=backend)
|
||
|
||
app.dependency_overrides[get_auth_service] = _get_auth_service_override
|
||
yield backend
|
||
|
||
|
||
async def _register_and_verify(
|
||
client: httpx.AsyncClient,
|
||
email_backend: _CapturingEmailBackend,
|
||
*,
|
||
email: str = "alice@example.com",
|
||
name_user: str = "Alice",
|
||
password: str = "supersecret1",
|
||
) -> None:
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": email, "name_user": name_user, "password": password},
|
||
)
|
||
assert response.status_code == 201, response.text
|
||
|
||
assert email_backend.sent, "письмо с подтверждением не было отправлено"
|
||
_, _, body = email_backend.sent[-1]
|
||
token = _extract_verification_token(body)
|
||
|
||
verify_response = await client.post("/api/v1/auth/verify-email", json={"token": token})
|
||
assert verify_response.status_code == 204, verify_response.text
|
||
|
||
|
||
async def test_full_cycle_register_verify_login_me(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
"""Регистрация → подтверждение email → вход → GET /users/me с access-токеном."""
|
||
await _register_and_verify(client, email_backend)
|
||
|
||
login_response = await client.post(
|
||
"/api/v1/auth/token",
|
||
data={"username": "alice@example.com", "password": "supersecret1"},
|
||
)
|
||
assert login_response.status_code == 200, login_response.text
|
||
body = login_response.json()
|
||
assert body["token_type"] == "bearer"
|
||
access_token = body["access_token"]
|
||
assert REFRESH_COOKIE in login_response.cookies
|
||
|
||
me_response = await client.get(
|
||
"/api/v1/users/me", headers={"Authorization": f"Bearer {access_token}"}
|
||
)
|
||
assert me_response.status_code == 200
|
||
me_body = me_response.json()
|
||
assert me_body["email"] == "alice@example.com"
|
||
assert me_body["name_user"] == "Alice"
|
||
assert me_body["role"] == "user"
|
||
|
||
|
||
async def test_register_duplicate_email_returns_409(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
await _register_and_verify(client, email_backend, email="dup@example.com")
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": "dup@example.com", "name_user": "Dup", "password": "supersecret1"},
|
||
)
|
||
assert response.status_code == 409
|
||
|
||
|
||
async def test_login_without_verification_returns_403(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
register_response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": "bob@example.com", "name_user": "Bob", "password": "supersecret1"},
|
||
)
|
||
assert register_response.status_code == 201
|
||
|
||
login_response = await client.post(
|
||
"/api/v1/auth/token",
|
||
data={"username": "bob@example.com", "password": "supersecret1"},
|
||
)
|
||
assert login_response.status_code == 403
|
||
|
||
|
||
async def test_verify_email_with_invalid_token_returns_400(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
response = await client.post("/api/v1/auth/verify-email", json={"token": "does-not-exist"})
|
||
assert response.status_code == 400
|
||
|
||
|
||
async def test_protected_endpoint_without_jwt_returns_401(client: httpx.AsyncClient) -> None:
|
||
response = await client.get("/api/v1/users/me")
|
||
assert response.status_code == 401
|
||
|
||
|
||
async def test_refresh_rotates_token_and_rejects_reuse(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
await _register_and_verify(client, email_backend, email="carol@example.com")
|
||
login_response = await client.post(
|
||
"/api/v1/auth/token",
|
||
data={"username": "carol@example.com", "password": "supersecret1"},
|
||
)
|
||
assert login_response.status_code == 200
|
||
old_refresh_token = client.cookies.get(REFRESH_COOKIE)
|
||
assert old_refresh_token is not None
|
||
|
||
refresh_response = await client.post("/api/v1/auth/refresh")
|
||
assert refresh_response.status_code == 200
|
||
new_refresh_token = client.cookies.get(REFRESH_COOKIE)
|
||
assert new_refresh_token is not None
|
||
assert new_refresh_token != old_refresh_token
|
||
|
||
# Повторное предъявление уже потраченного (старого) refresh-токена — reuse.
|
||
client.cookies.set(REFRESH_COOKIE, old_refresh_token)
|
||
reuse_response = await client.post("/api/v1/auth/refresh")
|
||
assert reuse_response.status_code == 401
|
||
|
||
|
||
async def test_refresh_without_cookie_returns_401(client: httpx.AsyncClient) -> None:
|
||
response = await client.post("/api/v1/auth/refresh")
|
||
assert response.status_code == 401
|
||
|
||
|
||
async def test_logout_revokes_refresh_token(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
await _register_and_verify(client, email_backend, email="dave@example.com")
|
||
login_response = await client.post(
|
||
"/api/v1/auth/token",
|
||
data={"username": "dave@example.com", "password": "supersecret1"},
|
||
)
|
||
assert login_response.status_code == 200
|
||
|
||
logout_response = await client.post("/api/v1/auth/logout")
|
||
assert logout_response.status_code == 204
|
||
|
||
refresh_response = await client.post("/api/v1/auth/refresh")
|
||
assert refresh_response.status_code == 401
|
||
|
||
|
||
# --- Выбор команды при регистрации ---------------------------------------------
|
||
|
||
|
||
async def test_registration_options_disabled_by_default(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Настройка `registration_team_choice` выключена (см. `_reset_registration_gating`)
|
||
— команды не отдаются."""
|
||
db_session.add(Team(name=f"Zeta Team {uuid.uuid4()}"))
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/auth/registration-options")
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["team_choice_enabled"] is False
|
||
assert body["teams"] == []
|
||
assert body["email_domain"] is None
|
||
|
||
|
||
async def test_registration_options_enabled_returns_teams_sorted_by_name(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Список команд отсортирован по названию; проверяем взаимный порядок своих команд
|
||
(не полное равенство списка — в общей dev-БД могут быть и другие команды)."""
|
||
suffix = uuid.uuid4().hex[:8]
|
||
name_alpha, name_zeta = f"Alpha-{suffix}", f"Zeta-{suffix}"
|
||
db_session.add_all([Team(name=name_zeta), Team(name=name_alpha)])
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(registration_team_choice=True)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/auth/registration-options")
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["team_choice_enabled"] is True
|
||
names = [team["name"] for team in body["teams"]]
|
||
assert names == sorted(names)
|
||
assert name_alpha in names and name_zeta in names
|
||
assert names.index(name_alpha) < names.index(name_zeta)
|
||
|
||
|
||
async def test_register_without_team_id_still_works(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": "no-team@example.com", "name_user": "No Team", "password": "supersecret1"},
|
||
)
|
||
assert response.status_code == 201, response.text
|
||
|
||
|
||
async def test_register_with_team_id_when_enabled_assigns_team(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
team = Team(name=f"Engineering {uuid.uuid4()}")
|
||
db_session.add(team)
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(registration_team_choice=True)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "with-team@example.com",
|
||
"name_user": "With Team",
|
||
"password": "supersecret1",
|
||
"team_id": str(team.id),
|
||
},
|
||
)
|
||
assert response.status_code == 201, response.text
|
||
|
||
result = await db_session.execute(select(User).where(User.email == "with-team@example.com"))
|
||
created = result.scalar_one()
|
||
assert created.team_id == team.id
|
||
|
||
|
||
async def test_register_with_team_id_when_disabled_returns_400(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
# `registration_team_choice` выключена `_reset_registration_gating`.
|
||
team = Team(name=f"Disabled Choice Team {uuid.uuid4()}")
|
||
db_session.add(team)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "team-disabled@example.com",
|
||
"name_user": "Team Disabled",
|
||
"password": "supersecret1",
|
||
"team_id": str(team.id),
|
||
},
|
||
)
|
||
assert response.status_code == 400
|
||
|
||
|
||
async def test_register_with_unknown_team_id_returns_400(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(registration_team_choice=True)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "unknown-team@example.com",
|
||
"name_user": "Unknown Team",
|
||
"password": "supersecret1",
|
||
"team_id": "00000000-0000-0000-0000-000000000000",
|
||
},
|
||
)
|
||
assert response.status_code == 400
|
||
|
||
|
||
# --- Верификация регистрирующихся по домену email -------------------------------------
|
||
|
||
|
||
async def test_registration_options_returns_email_domain_when_enabled(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""При включённой верификации домена `registration-options` отдаёт эталонный домен."""
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(
|
||
registration_email_domain_enabled=True, registration_email_domain="example.com"
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/auth/registration-options")
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["email_domain"] == "example.com"
|
||
|
||
|
||
async def test_registration_options_email_domain_null_when_disabled(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Выключенная верификация домена — `email_domain` всегда `null`, даже если домен сохранён."""
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(
|
||
registration_email_domain_enabled=True, registration_email_domain="example.com"
|
||
)
|
||
)
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(registration_email_domain_enabled=False)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/auth/registration-options")
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["email_domain"] is None
|
||
|
||
|
||
async def test_register_with_foreign_domain_when_verification_enabled_returns_400(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(
|
||
registration_email_domain_enabled=True, registration_email_domain="example.com"
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "alice@other.com",
|
||
"name_user": "Alice",
|
||
"password": "supersecret1",
|
||
},
|
||
)
|
||
assert response.status_code == 400
|
||
assert response.json()["detail"] == "invalid_email_domain"
|
||
|
||
|
||
async def test_register_with_matching_domain_case_insensitive_succeeds(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(
|
||
registration_email_domain_enabled=True, registration_email_domain="example.com"
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "bob@EXAMPLE.COM",
|
||
"name_user": "Bob",
|
||
"password": "supersecret1",
|
||
},
|
||
)
|
||
assert response.status_code == 201, response.text
|
||
|
||
|
||
async def test_register_any_domain_allowed_when_verification_disabled(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
"""Выключенная верификация (`_reset_registration_gating`) — любой домен email
|
||
проходит регистрацию."""
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={
|
||
"email": "carol@anything.example",
|
||
"name_user": "Carol",
|
||
"password": "supersecret1",
|
||
},
|
||
)
|
||
assert response.status_code == 201, response.text
|
||
|
||
|
||
async def test_register_sets_reply_to_when_contact_email_enabled(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
"""Контактный адрес инстанса включён — письмо подтверждения регистрации несёт `Reply-To`."""
|
||
await InstanceSettingsService(db_session).update(
|
||
SettingsUpdateIn(contact_email_enabled=True, contact_email="contact@vidconf.example")
|
||
)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": "dave@example.com", "name_user": "Dave", "password": "supersecret1"},
|
||
)
|
||
|
||
assert response.status_code == 201, response.text
|
||
assert email_backend.reply_to[-1] == "contact@vidconf.example"
|
||
|
||
|
||
async def test_register_no_reply_to_when_contact_email_disabled(
|
||
client: httpx.AsyncClient, email_backend: _CapturingEmailBackend
|
||
) -> None:
|
||
"""Контактный адрес выключен (дефолт) — `Reply-To` не проставляется."""
|
||
response = await client.post(
|
||
"/api/v1/auth/register",
|
||
json={"email": "erin@example.com", "name_user": "Erin", "password": "supersecret1"},
|
||
)
|
||
|
||
assert response.status_code == 201, response.text
|
||
assert email_backend.reply_to[-1] is None
|