396 lines
14 KiB
Python
396 lines
14 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
|
||
|
||
# Минимальные валидные по магическим байтам содержимые (без полноценного
|
||
# декодирования — `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=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=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=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
|