267 lines
9.5 KiB
Python
267 lines
9.5 KiB
Python
"""Тесты раздела «Пользователи»: список, карточка и управление (без БД)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from common.repositories import users_admin_repo
|
|
|
|
USER: dict[str, Any] = {
|
|
"id_user": 42,
|
|
"username": "test_user",
|
|
"first_name": "Test",
|
|
"last_name": None,
|
|
"nick": "ТестовыйНик",
|
|
"lvl": 3,
|
|
"expr": 120,
|
|
"expr_streak": 2,
|
|
"dt_create": datetime(2026, 1, 1, 12, 0, tzinfo=UTC),
|
|
"id_title": 1,
|
|
"title": "Ветеран",
|
|
"msg_count": 42,
|
|
"last_message_at": datetime(2026, 10, 18, 20, 0, tzinfo=UTC),
|
|
"balance": 1500,
|
|
"eggs": 2,
|
|
"purification_count": 5,
|
|
"last_purification_at": datetime(2026, 10, 17, 16, 0, tzinfo=UTC),
|
|
"subscribed": True,
|
|
}
|
|
|
|
MONEY: list[dict[str, Any]] = [
|
|
{"id_money": 2, "amount": 50, "dt_action": datetime(2026, 10, 18, 16, 0, tzinfo=UTC), "action": "Очищение"},
|
|
{"id_money": 1, "amount": -20, "dt_action": datetime(2026, 10, 17, 12, 0, tzinfo=UTC), "action": "Покупка"},
|
|
]
|
|
|
|
TITLES: list[dict[str, Any]] = [
|
|
{"id_title": 1, "title": "Ветеран"},
|
|
{"id_title": 2, "title": "Новичок"},
|
|
]
|
|
|
|
HISTORY: list[dict[str, Any]] = [
|
|
{
|
|
"id": 1,
|
|
"field": "nick",
|
|
"old_value": "Старый",
|
|
"new_value": "ТестовыйНик",
|
|
"comment": None,
|
|
"editor_id": 1,
|
|
"changed_at": datetime(2026, 10, 18, 12, 0, tzinfo=UTC),
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_repo(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
|
state: dict[str, Any] = {"total": 1, "calls": []}
|
|
|
|
async def count_users(search: str = "") -> int:
|
|
state["search"] = search
|
|
return state["total"]
|
|
|
|
async def list_users(search: str = "", *, limit: int = 25, offset: int = 0) -> list[dict[str, Any]]:
|
|
state["calls"].append({"search": search, "limit": limit, "offset": offset})
|
|
return [USER]
|
|
|
|
async def get_user(user_id: int) -> dict[str, Any] | None:
|
|
return USER if user_id == USER["id_user"] else None
|
|
|
|
async def money_history(user_id: int, *, limit: int = 10) -> list[dict[str, Any]]:
|
|
return MONEY if user_id == USER["id_user"] else []
|
|
|
|
async def list_titles() -> list[dict[str, Any]]:
|
|
return TITLES
|
|
|
|
async def user_history(user_id: int, *, limit: int = 20) -> list[dict[str, Any]]:
|
|
return HISTORY if user_id == USER["id_user"] else []
|
|
|
|
async def set_nick(user_id: int, nick: str, *, editor_id: int | None = None) -> bool:
|
|
state["nick"] = (user_id, nick, editor_id)
|
|
return user_id == USER["id_user"]
|
|
|
|
async def set_title(user_id: int, title_id: int, *, editor_id: int | None = None) -> bool:
|
|
state["title"] = (user_id, title_id, editor_id)
|
|
return user_id == USER["id_user"]
|
|
|
|
async def set_subscription(user_id: int, subscribed: bool, *, editor_id: int | None = None) -> bool:
|
|
state["subscription"] = (user_id, subscribed, editor_id)
|
|
return user_id == USER["id_user"]
|
|
|
|
async def adjust_balance(
|
|
user_id: int,
|
|
amount: int,
|
|
reason: str,
|
|
*,
|
|
editor_id: int | None = None,
|
|
) -> int | None:
|
|
state["balance"] = (user_id, amount, reason, editor_id)
|
|
return USER["balance"] + amount if user_id == USER["id_user"] else None
|
|
|
|
monkeypatch.setattr(users_admin_repo, "count_users", count_users)
|
|
monkeypatch.setattr(users_admin_repo, "list_users", list_users)
|
|
monkeypatch.setattr(users_admin_repo, "get_user", get_user)
|
|
monkeypatch.setattr(users_admin_repo, "money_history", money_history)
|
|
monkeypatch.setattr(users_admin_repo, "list_titles", list_titles)
|
|
monkeypatch.setattr(users_admin_repo, "user_history", user_history)
|
|
monkeypatch.setattr(users_admin_repo, "set_nick", set_nick)
|
|
monkeypatch.setattr(users_admin_repo, "set_title", set_title)
|
|
monkeypatch.setattr(users_admin_repo, "set_subscription", set_subscription)
|
|
monkeypatch.setattr(users_admin_repo, "adjust_balance", adjust_balance)
|
|
return state
|
|
|
|
|
|
def _csrf(client: TestClient, csrf_token) -> str:
|
|
return csrf_token(client.get("/users/42").text)
|
|
|
|
|
|
def test_users_list(fake_repo: dict[str, Any], client_factory) -> None:
|
|
response = client_factory().get("/users")
|
|
|
|
assert response.status_code == 200
|
|
assert "ТестовыйНик" in response.text
|
|
assert "@test_user" in response.text
|
|
assert "1500" in response.text
|
|
|
|
|
|
def test_users_search_passed_to_repo(fake_repo: dict[str, Any], client_factory) -> None:
|
|
client_factory().get("/users", params={"q": "Тест"})
|
|
|
|
assert fake_repo["calls"][0]["search"] == "Тест"
|
|
|
|
|
|
def test_users_pagination(fake_repo: dict[str, Any], client_factory) -> None:
|
|
fake_repo["total"] = 30
|
|
|
|
response = client_factory().get("/users", params={"page": "2"})
|
|
|
|
assert "Страница 2 из 2" in response.text
|
|
assert fake_repo["calls"][0]["offset"] == 25
|
|
|
|
|
|
def test_user_card(fake_repo: dict[str, Any], client_factory) -> None:
|
|
response = client_factory().get("/users/42")
|
|
|
|
assert response.status_code == 200
|
|
assert "Ветеран" in response.text
|
|
assert "Управление" in response.text
|
|
assert "Очищение" in response.text
|
|
assert "+50" in response.text
|
|
assert "-20" in response.text
|
|
assert "Старый → ТестовыйНик" in response.text
|
|
|
|
|
|
def test_user_card_404(fake_repo: dict[str, Any], client_factory) -> None:
|
|
assert client_factory().get("/users/99").status_code == 404
|
|
|
|
|
|
def test_eggs_hidden_outside_event(fake_repo: dict[str, Any], client_factory) -> None:
|
|
response = client_factory().get("/users/42")
|
|
|
|
assert response.status_code == 200
|
|
assert "Яички" not in response.text
|
|
|
|
|
|
def test_eggs_visible_during_event(fake_repo: dict[str, Any], client_factory, settings_factory) -> None:
|
|
today = datetime.now(ZoneInfo("Europe/Moscow")).date()
|
|
client = client_factory(
|
|
settings=settings_factory(
|
|
egg_event_start=today - timedelta(days=1),
|
|
egg_event_end=today + timedelta(days=1),
|
|
)
|
|
)
|
|
|
|
assert "Яички" in client.get("/users/42").text
|
|
|
|
|
|
def test_update_nick(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/nick", data={"nick": "НовыйНик", "csrf_token": token})
|
|
|
|
assert response.status_code == 200
|
|
assert "Ник обновлён" in response.text
|
|
assert fake_repo["nick"] == (42, "НовыйНик", 1)
|
|
|
|
|
|
def test_update_nick_rejects_empty(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/nick", data={"nick": " ", "csrf_token": token})
|
|
|
|
assert response.status_code == 400
|
|
assert "Ник не может быть пустым" in response.text
|
|
assert "nick" not in fake_repo
|
|
|
|
|
|
def test_update_title(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/title", data={"title_id": "2", "csrf_token": token})
|
|
|
|
assert response.status_code == 200
|
|
assert "Звание обновлено" in response.text
|
|
assert fake_repo["title"] == (42, 2, 1)
|
|
|
|
|
|
def test_update_title_rejects_unknown(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/title", data={"title_id": "999", "csrf_token": token})
|
|
|
|
assert response.status_code == 400
|
|
assert "Звание не найдено" in response.text
|
|
|
|
|
|
def test_update_subscription(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/subscription", data={"subscribed": "0", "csrf_token": token})
|
|
|
|
assert response.status_code == 200
|
|
assert "Подписка обновлена" in response.text
|
|
assert fake_repo["subscription"] == (42, False, 1)
|
|
|
|
|
|
def test_update_balance(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post(
|
|
"/users/42/balance",
|
|
data={"amount": "-100", "reason": "Штраф", "csrf_token": token},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "Баланс скорректирован" in response.text
|
|
assert fake_repo["balance"] == (42, -100, "Штраф", 1)
|
|
|
|
|
|
def test_update_balance_requires_reason(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/balance", data={"amount": "100", "csrf_token": token})
|
|
|
|
assert response.status_code == 400
|
|
assert "Укажите причину" in response.text
|
|
assert "balance" not in fake_repo
|
|
|
|
|
|
def test_update_balance_rejects_zero(fake_repo: dict[str, Any], client_factory, csrf_token) -> None:
|
|
client: TestClient = client_factory()
|
|
token = _csrf(client, csrf_token)
|
|
|
|
response = client.post("/users/42/balance", data={"amount": "0", "reason": "Тест", "csrf_token": token})
|
|
|
|
assert response.status_code == 400
|
|
assert "Сумма не может быть нулевой" in response.text
|