Max Ronzhin
fec9255baa
Отключаемый в админке модуль `virtual_background` (дефолт — выключен, чтобы
обновление не меняло продукт у тех, кто ничего не просил). Флаг едет клиенту
двумя путями: на публичные страницы входа — через `GET /public/settings`,
участнику комнаты — в join-ответе (`JoinOut`), потому что значение нужно на
руках ДО первого рендера комнаты, а `/admin/settings` доступен только админу.
Свои картинки пользователя (`/users/me/backgrounds`, GET/POST/DELETE):
файлы на диске (`backgrounds/{user_id}/{id}.{ext}`), в БД только путь — как у
аватаров, «чтобы не грузили БД». Лимит в 10 штук проверяется на сервере под
блокировкой строки пользователя: две одновременные загрузки иначе обе увидели
бы «уже девять» и обе прошли бы. Удаление сносит и запись, и файл; чужую
картинку по её id удалить нельзя — владелец в условии запроса.
Валидация загрузки (допустимые форматы, магические байты, реальный размер)
выделена из `services/avatars.py` в общий `services/images.py`: правила у
аватара и фона одни и те же, а разъехавшись, они дали бы дыру ровно там, ради
чего проверка и написана. Публичный API аватаров не изменился.
Сжимает картинку клиент (Pillow на бэкенде нет), но серверная валидация
остаётся полноценной — запрос может прийти и мимо интерфейса.
Новый ключ настройки вписан в `_MANAGED_KEYS` тестов: без этого включённый
в общей dev-БД модуль ронял чужие тесты, которые считают себя изолированными.
544 lines
20 KiB
Python
544 lines
20 KiB
Python
"""Интеграционные тесты `/api/v1/users`: список пользователей, профиль, аватар, картинки фона."""
|
||
|
||
import uuid
|
||
from collections.abc import Generator
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from core.config import get_settings
|
||
from core.security import create_access_token, hash_password
|
||
from models.team import Team
|
||
from models.user import User
|
||
from services.backgrounds import MAX_BACKGROUNDS_PER_USER
|
||
|
||
# Минимальные валидные по магическим байтам содержимые (без полноценного
|
||
# декодирования — `services/avatars.py` проверяет только сигнатуру/размер).
|
||
_JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 32
|
||
_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
|
||
_WEBP_BYTES = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 32
|
||
|
||
|
||
async def _make_user(session: AsyncSession) -> User:
|
||
user = User(
|
||
email=f"{uuid.uuid4()}@example.com",
|
||
name_user="List Tester",
|
||
password_hash=await hash_password("password123"),
|
||
email_verified=True,
|
||
)
|
||
session.add(user)
|
||
await session.flush()
|
||
return user
|
||
|
||
|
||
async def _make_team(session: AsyncSession) -> Team:
|
||
team = Team(name=f"Team {uuid.uuid4()}")
|
||
session.add(team)
|
||
await session.flush()
|
||
return team
|
||
|
||
|
||
def _auth_headers(user: User) -> dict[str, str]:
|
||
token = create_access_token(user.id, user.role)
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
@pytest.fixture
|
||
def media_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[Path, None, None]:
|
||
"""Подменить `MEDIA_ROOT` на изолированный `tmp_path` на время теста.
|
||
|
||
`get_settings()` кэширован (`lru_cache`) — без явного сброса кэша тест
|
||
продолжил бы видеть значение, закэшированное первым же вызовом в сессии
|
||
pytest. Перечисляется в сигнатуре теста ПЕРВЫМ параметром, чтобы pytest
|
||
инициализировал его до `client`/`app` (см. `main.py::create_app`,
|
||
монтирующий `StaticFiles` по значению `media_root` на момент вызова).
|
||
"""
|
||
monkeypatch.setenv("MEDIA_ROOT", str(tmp_path))
|
||
get_settings.cache_clear()
|
||
yield tmp_path
|
||
get_settings.cache_clear()
|
||
|
||
|
||
async def test_list_users_returns_id_and_display_name(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
requester = await _make_user(db_session)
|
||
other = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/users", headers=_auth_headers(requester))
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
ids = {item["id"] for item in body}
|
||
assert str(requester.id) in ids
|
||
assert str(other.id) in ids
|
||
item = next(i for i in body if i["id"] == str(other.id))
|
||
assert item["display_name"] == other.name_user
|
||
|
||
|
||
async def test_list_users_requires_auth(client: httpx.AsyncClient) -> None:
|
||
response = await client.get("/api/v1/users")
|
||
assert response.status_code == 401
|
||
|
||
|
||
async def test_get_me_with_reserved_tld_email_does_not_500(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Регрессия: `UserOut.email` — `str`, а не `EmailStr` (иначе 500 `ResponseValidationError`).
|
||
|
||
`.local`/`.test` и т.п. — синтаксически валидные, но зарезервированные
|
||
special-use домены (RFC 6761); email-validator (`EmailStr`) их отклоняет,
|
||
хотя в БД такой email мог оказаться и до этого правила (например,
|
||
сид-админ `admin@vidconf.local` в старых окружениях).
|
||
"""
|
||
# Уникальный email (не сталкивается с реально засеянным admin@vidconf.local
|
||
# в дев-БД — тесты идут в общей БД под savepoint, а email уникален).
|
||
legacy_email = f"legacy-{uuid.uuid4()}@vidconf.local"
|
||
user = User(
|
||
email=legacy_email,
|
||
name_user="Legacy Admin",
|
||
password_hash=await hash_password("password123"),
|
||
email_verified=True,
|
||
role="admin",
|
||
)
|
||
db_session.add(user)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/users/me", headers=_auth_headers(user))
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["email"] == legacy_email
|
||
assert body["role"] == "admin"
|
||
|
||
|
||
# --- Профиль ------------------------------------------------------
|
||
|
||
|
||
async def test_get_me_returns_profile_fields(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/users/me", headers=_auth_headers(user))
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["avatar_url"] is None
|
||
assert body["team_id"] is None
|
||
assert body["team_name"] is None
|
||
|
||
|
||
async def test_patch_me_updates_name_and_team_but_ignores_email(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
team = await _make_team(db_session)
|
||
await db_session.commit()
|
||
original_email = user.email
|
||
|
||
response = await client.patch(
|
||
"/api/v1/users/me",
|
||
json={"name_user": "Renamed Self", "team_id": str(team.id), "email": "new@example.com"},
|
||
headers=_auth_headers(user),
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["name_user"] == "Renamed Self"
|
||
assert body["team_id"] == str(team.id)
|
||
assert body["team_name"] == team.name
|
||
assert body["email"] == original_email # email — read-only, тело проигнорировано
|
||
|
||
|
||
async def test_patch_me_with_unknown_team_returns_404(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.patch(
|
||
"/api/v1/users/me",
|
||
json={"team_id": str(uuid.uuid4())},
|
||
headers=_auth_headers(user),
|
||
)
|
||
assert response.status_code == 404
|
||
assert response.json()["detail"] == "team_not_found"
|
||
|
||
|
||
async def test_patch_me_explicit_null_team_clears_it(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
team = await _make_team(db_session)
|
||
await db_session.commit()
|
||
|
||
await client.patch(
|
||
"/api/v1/users/me", json={"team_id": str(team.id)}, headers=_auth_headers(user)
|
||
)
|
||
response = await client.patch(
|
||
"/api/v1/users/me", json={"team_id": None}, headers=_auth_headers(user)
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["team_id"] is None
|
||
|
||
|
||
# --- Аватар -------------------------------------------------------
|
||
|
||
|
||
async def test_upload_avatar_jpeg_happy_path(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.jpg", _JPEG_BYTES, "image/jpeg")},
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["avatar_url"] is not None
|
||
assert body["avatar_url"].startswith(f"/media/avatars/{user.id}.jpg?v=")
|
||
assert (media_root / "avatars" / f"{user.id}.jpg").read_bytes() == _JPEG_BYTES
|
||
|
||
|
||
async def test_upload_avatar_png_happy_path(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.png", _PNG_BYTES, "image/png")},
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
assert (media_root / "avatars" / f"{user.id}.png").exists()
|
||
|
||
|
||
async def test_upload_avatar_replaces_previous_file_with_different_extension(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.jpg", _JPEG_BYTES, "image/jpeg")},
|
||
)
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.webp", _WEBP_BYTES, "image/webp")},
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
assert not (media_root / "avatars" / f"{user.id}.jpg").exists()
|
||
assert (media_root / "avatars" / f"{user.id}.webp").exists()
|
||
|
||
|
||
async def test_upload_avatar_spoofed_content_type_returns_415(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""`Content-Type: image/png`, но байты — JPEG (подделка заголовка клиентом)."""
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.png", _JPEG_BYTES, "image/png")},
|
||
)
|
||
assert response.status_code == 415
|
||
assert response.json()["detail"] == "avatar_invalid_type"
|
||
|
||
|
||
async def test_upload_avatar_unsupported_content_type_returns_415(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.txt", b"not-an-image", "text/plain")},
|
||
)
|
||
assert response.status_code == 415
|
||
|
||
|
||
async def test_upload_avatar_too_large_returns_413(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
oversized = _JPEG_BYTES + b"\x00" * (2 * 1024 * 1024)
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.jpg", oversized, "image/jpeg")},
|
||
)
|
||
assert response.status_code == 413
|
||
assert response.json()["detail"] == "avatar_too_large"
|
||
|
||
|
||
async def test_delete_avatar_removes_file_and_clears_url(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
await client.post(
|
||
"/api/v1/users/me/avatar",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("avatar.jpg", _JPEG_BYTES, "image/jpeg")},
|
||
)
|
||
|
||
response = await client.delete("/api/v1/users/me/avatar", headers=_auth_headers(user))
|
||
assert response.status_code == 204
|
||
assert not (media_root / "avatars" / f"{user.id}.jpg").exists()
|
||
|
||
me = await client.get("/api/v1/users/me", headers=_auth_headers(user))
|
||
assert me.json()["avatar_url"] is None
|
||
|
||
|
||
# --- Смена пароля ---------------------------------------------------------------------
|
||
|
||
|
||
async def test_change_password_success_new_password_works_old_does_not(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
email = user.email
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/password",
|
||
json={"current_password": "password123", "new_password": "new-password456"},
|
||
headers=_auth_headers(user),
|
||
)
|
||
assert response.status_code == 204, response.text
|
||
|
||
new_login = await client.post(
|
||
"/api/v1/auth/token", data={"username": email, "password": "new-password456"}
|
||
)
|
||
assert new_login.status_code == 200, new_login.text
|
||
|
||
old_login = await client.post(
|
||
"/api/v1/auth/token", data={"username": email, "password": "password123"}
|
||
)
|
||
assert old_login.status_code == 401
|
||
|
||
|
||
async def test_change_password_wrong_current_returns_400(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/password",
|
||
json={"current_password": "wrong-password", "new_password": "new-password456"},
|
||
headers=_auth_headers(user),
|
||
)
|
||
assert response.status_code == 400
|
||
assert response.json()["detail"] == "invalid_current_password"
|
||
|
||
|
||
async def test_change_password_too_short_returns_422(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/password",
|
||
json={"current_password": "password123", "new_password": "short"},
|
||
headers=_auth_headers(user),
|
||
)
|
||
assert response.status_code == 422
|
||
|
||
|
||
async def test_change_password_requires_auth(client: httpx.AsyncClient) -> None:
|
||
response = await client.post(
|
||
"/api/v1/users/me/password",
|
||
json={"current_password": "a", "new_password": "new-password456"},
|
||
)
|
||
assert response.status_code == 401
|
||
|
||
|
||
# --- Пикер участников: поиск `q` + `avatar_url` -------------------
|
||
|
||
|
||
async def test_list_users_search_by_q_filters_by_name_or_email(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
requester = await _make_user(db_session)
|
||
unique_marker = str(uuid.uuid4())
|
||
match = User(
|
||
email=f"{unique_marker}@example.com",
|
||
name_user=f"Findable {unique_marker}",
|
||
password_hash=await hash_password("password123"),
|
||
email_verified=True,
|
||
)
|
||
db_session.add(match)
|
||
await db_session.commit()
|
||
|
||
response = await client.get(
|
||
"/api/v1/users", params={"q": unique_marker}, headers=_auth_headers(requester)
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
ids = {item["id"] for item in response.json()}
|
||
assert str(match.id) in ids
|
||
assert str(requester.id) not in ids
|
||
|
||
|
||
# --- Свои картинки фона (`/users/me/backgrounds`) -----------------
|
||
|
||
|
||
async def _upload_background(
|
||
client: httpx.AsyncClient, user: User, *, name: str = "bg.webp"
|
||
) -> httpx.Response:
|
||
return await client.post(
|
||
"/api/v1/users/me/backgrounds",
|
||
headers=_auth_headers(user),
|
||
files={"file": (name, _WEBP_BYTES, "image/webp")},
|
||
)
|
||
|
||
|
||
async def test_backgrounds_list_is_empty_by_default_and_reports_limit(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.get("/api/v1/users/me/backgrounds", headers=_auth_headers(user))
|
||
assert response.status_code == 200, response.text
|
||
body = response.json()
|
||
assert body["items"] == []
|
||
assert body["limit"] == MAX_BACKGROUNDS_PER_USER
|
||
|
||
|
||
async def test_upload_background_saves_file_and_returns_url(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await _upload_background(client, user)
|
||
assert response.status_code == 201, response.text
|
||
items = response.json()["items"]
|
||
assert len(items) == 1
|
||
url = items[0]["url"]
|
||
assert url == f"/media/backgrounds/{user.id}/{items[0]['id']}.webp"
|
||
# Файл лежит на диске, в БД только путь — см. `services/backgrounds.py`.
|
||
assert (media_root / url.removeprefix("/media/")).read_bytes() == _WEBP_BYTES
|
||
|
||
|
||
async def test_upload_background_spoofed_content_type_returns_415(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""`Content-Type: image/webp`, но байты — JPEG: клиенту не верим и здесь."""
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/backgrounds",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("bg.webp", _JPEG_BYTES, "image/webp")},
|
||
)
|
||
assert response.status_code == 415
|
||
assert response.json()["detail"] == "background_invalid_type"
|
||
|
||
|
||
async def test_upload_background_too_large_returns_413(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
oversized = _WEBP_BYTES + b"\x00" * (2 * 1024 * 1024)
|
||
|
||
response = await client.post(
|
||
"/api/v1/users/me/backgrounds",
|
||
headers=_auth_headers(user),
|
||
files={"file": ("bg.webp", oversized, "image/webp")},
|
||
)
|
||
assert response.status_code == 413
|
||
assert response.json()["detail"] == "background_too_large"
|
||
|
||
|
||
async def test_upload_background_over_limit_returns_409_and_keeps_ten(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Лимит держится на СЕРВЕРЕ: интерфейс можно обойти прямым запросом."""
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
for _ in range(MAX_BACKGROUNDS_PER_USER):
|
||
assert (await _upload_background(client, user)).status_code == 201
|
||
|
||
response = await _upload_background(client, user)
|
||
assert response.status_code == 409
|
||
assert response.json()["detail"] == "background_limit_reached"
|
||
|
||
listing = await client.get("/api/v1/users/me/backgrounds", headers=_auth_headers(user))
|
||
assert len(listing.json()["items"]) == MAX_BACKGROUNDS_PER_USER
|
||
|
||
|
||
async def test_delete_background_removes_file_and_row(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
user = await _make_user(db_session)
|
||
await db_session.commit()
|
||
created = (await _upload_background(client, user)).json()["items"][0]
|
||
file_path = media_root / created["url"].removeprefix("/media/")
|
||
assert file_path.exists()
|
||
|
||
response = await client.delete(
|
||
f"/api/v1/users/me/backgrounds/{created['id']}", headers=_auth_headers(user)
|
||
)
|
||
assert response.status_code == 204
|
||
assert not file_path.exists()
|
||
|
||
listing = await client.get("/api/v1/users/me/backgrounds", headers=_auth_headers(user))
|
||
assert listing.json()["items"] == []
|
||
|
||
|
||
async def test_delete_other_users_background_returns_404(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
"""Знание id чужой картинки не даёт её удалить — владелец в условии запроса."""
|
||
owner = await _make_user(db_session)
|
||
stranger = await _make_user(db_session)
|
||
await db_session.commit()
|
||
created = (await _upload_background(client, owner)).json()["items"][0]
|
||
|
||
response = await client.delete(
|
||
f"/api/v1/users/me/backgrounds/{created['id']}", headers=_auth_headers(stranger)
|
||
)
|
||
assert response.status_code == 404
|
||
assert (media_root / created["url"].removeprefix("/media/")).exists()
|
||
|
||
|
||
async def test_backgrounds_list_does_not_leak_other_users_images(
|
||
media_root: Path, client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
owner = await _make_user(db_session)
|
||
stranger = await _make_user(db_session)
|
||
await db_session.commit()
|
||
await _upload_background(client, owner)
|
||
|
||
response = await client.get("/api/v1/users/me/backgrounds", headers=_auth_headers(stranger))
|
||
assert response.json()["items"] == []
|
||
|
||
|
||
async def test_backgrounds_require_authentication(
|
||
media_root: Path, client: httpx.AsyncClient
|
||
) -> None:
|
||
assert (await client.get("/api/v1/users/me/backgrounds")).status_code == 401
|
||
assert (
|
||
await client.delete(f"/api/v1/users/me/backgrounds/{uuid.uuid4()}")
|
||
).status_code == 401
|