Удалил папку tests из репозитория
This commit is contained in:
admin
@@ -1,106 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
import asyncio
|
||||
from asyncio import AbstractEventLoop
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import AsyncGenerator, Any, Generator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
from database import BotDatabase, RoleRegion
|
||||
|
||||
# Добавляем путь к корню проекта
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop() -> Generator[AbstractEventLoop, Any, None]:
|
||||
"""
|
||||
Создаёт event loop для асинхронных тестов.
|
||||
Scope: session, чтобы использовать один loop на всю сессию тестов.
|
||||
"""
|
||||
policy = asyncio.get_event_loop_policy()
|
||||
loop: asyncio.AbstractEventLoop = policy.new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session")
|
||||
async def test_db() -> AsyncGenerator[BotDatabase, None]:
|
||||
"""
|
||||
Создаёт тестовую базу данных в памяти.
|
||||
Инициализирует тестовые роли.
|
||||
"""
|
||||
db: BotDatabase = BotDatabase("sqlite+aiosqlite:///:memory:", echo=False)
|
||||
await db.init_db()
|
||||
|
||||
# Инициализируем тестовые роли
|
||||
test_roles = [
|
||||
("Альбедо", RoleRegion.MONDSTADT),
|
||||
("Нахида", RoleRegion.SUMERU),
|
||||
("Кафка", RoleRegion.HSR_STAR),
|
||||
("Броння", RoleRegion.HSR_STAR),
|
||||
("Чжун Ли", RoleRegion.LIYUE)
|
||||
]
|
||||
await db.init_roles(test_roles)
|
||||
|
||||
yield db
|
||||
await db.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_session(test_db: BotDatabase) -> AsyncGenerator:
|
||||
"""
|
||||
Создаёт тестовую сессию для работы с БД.
|
||||
Scope: function (по умолчанию).
|
||||
"""
|
||||
async with test_db.session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user(test_db: BotDatabase) -> int:
|
||||
"""
|
||||
Создаёт тестового пользователя.
|
||||
Возвращает user_id.
|
||||
"""
|
||||
user_id: int = 123456789
|
||||
await test_db.add_user(
|
||||
user_id=user_id,
|
||||
username="test_user",
|
||||
full_name="Test User"
|
||||
)
|
||||
return user_id
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user_with_messages(test_db: BotDatabase, test_user: int) -> int:
|
||||
"""
|
||||
Создаёт пользователя с тестовыми сообщениями за разные периоды.
|
||||
Сообщения распределены по месяцам, неделям и дням.
|
||||
"""
|
||||
now: datetime = datetime.now(timezone.utc)
|
||||
|
||||
# Даты сообщений: > месяца назад, в текущем месяце, в текущей неделе, сегодня
|
||||
test_dates: list[datetime] = [
|
||||
now - timedelta(days=40),
|
||||
now - timedelta(days=35),
|
||||
now - timedelta(days=20),
|
||||
now - timedelta(days=15),
|
||||
now - timedelta(days=8),
|
||||
now - timedelta(days=5),
|
||||
now - timedelta(days=2),
|
||||
now - timedelta(hours=12),
|
||||
now - timedelta(hours=1),
|
||||
now
|
||||
]
|
||||
|
||||
for i, date in enumerate(test_dates):
|
||||
await test_db.add_message(
|
||||
user_id=test_user,
|
||||
message_text=f"Тестовое сообщение {i + 1}",
|
||||
created_at=date
|
||||
)
|
||||
|
||||
return test_user
|
||||
@@ -1,125 +0,0 @@
|
||||
from datetime import datetime, timezone
|
||||
from typing import List
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select, Sequence
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from database import UserMessage, BotDatabase
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestMessageManagement:
|
||||
"""Тесты для управления сообщениями с полной строгой типизацией"""
|
||||
|
||||
async def test_message_creation(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест создания сообщения.
|
||||
Проверяет, что сообщение успешно сохраняется в базе и содержит правильные данные.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
test_text: str = "Тестовое сообщение для проверки"
|
||||
|
||||
await test_db.add_message(user_id, test_text)
|
||||
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await test_session.execute(stmt)
|
||||
messages: Sequence[UserMessage] = result.scalars().all()
|
||||
|
||||
assert len(messages) == 1
|
||||
assert messages[0].message_text == test_text
|
||||
assert messages[0].user_id == user_id
|
||||
assert messages[0].created_at is not None
|
||||
|
||||
async def test_message_with_custom_date(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession
|
||||
) -> None:
|
||||
"""
|
||||
Тест добавления сообщения с кастомной датой.
|
||||
Проверяет, что дата создания сохраняется корректно.
|
||||
"""
|
||||
user_id: int = 999888777
|
||||
custom_date: datetime = datetime(2024, 1, 15, 12, 30, 0, tzinfo=timezone.utc)
|
||||
|
||||
await test_db.add_user(user_id, "test_user", "Test User")
|
||||
await test_db.add_message(
|
||||
user_id=user_id,
|
||||
message_text="Сообщение с кастомной датой",
|
||||
created_at=custom_date
|
||||
)
|
||||
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await test_session.execute(stmt)
|
||||
messages: Sequence[UserMessage] = result.scalars().all()
|
||||
|
||||
assert len(messages) == 1
|
||||
db_date: datetime = messages[0].created_at
|
||||
if db_date.tzinfo is not None:
|
||||
db_date = db_date.replace(tzinfo=None)
|
||||
expected_date: datetime = custom_date.replace(tzinfo=None)
|
||||
assert db_date == expected_date
|
||||
|
||||
async def test_multiple_messages(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест добавления нескольких сообщений.
|
||||
Проверяет, что все сообщения корректно сохраняются в базе.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
# Удаляем старые сообщения
|
||||
async with test_db.session_factory() as session:
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await session.execute(stmt)
|
||||
old_messages: Sequence[UserMessage] = result.scalars().all()
|
||||
for msg in old_messages:
|
||||
await session.delete(msg)
|
||||
await session.commit()
|
||||
|
||||
# Добавляем несколько сообщений
|
||||
for i in range(5):
|
||||
await test_db.add_message(
|
||||
user_id=user_id,
|
||||
message_text=f"Сообщение {i + 1}"
|
||||
)
|
||||
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await test_session.execute(stmt)
|
||||
messages: Sequence[UserMessage] = result.scalars().all()
|
||||
|
||||
assert len(messages) == 5
|
||||
|
||||
async def test_message_ordering(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест проверки порядка сообщений по дате создания.
|
||||
Сообщения должны возвращаться в порядке возрастания даты.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
# Очищаем старые сообщения
|
||||
async with test_db.session_factory() as session:
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await session.execute(stmt)
|
||||
old_messages: Sequence[UserMessage] = result.scalars().all()
|
||||
for msg in old_messages:
|
||||
await session.delete(msg)
|
||||
await session.commit()
|
||||
|
||||
texts: List[str] = ["Сообщение 1", "Сообщение 2", "Сообщение 3"]
|
||||
|
||||
for text in texts:
|
||||
await test_db.add_message(user_id, text)
|
||||
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id).order_by(UserMessage.created_at.asc())
|
||||
result = await test_session.execute(stmt)
|
||||
messages: Sequence[UserMessage] = result.scalars().all()
|
||||
|
||||
assert len(messages) == 3
|
||||
assert messages[0].message_text == "Сообщение 1"
|
||||
assert messages[1].message_text == "Сообщение 2"
|
||||
assert messages[2].message_text == "Сообщение 3"
|
||||
@@ -1,207 +0,0 @@
|
||||
import pytest
|
||||
from typing import List, Dict
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from database import Role, RoleRegion, BotDatabase
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestRoleSystem:
|
||||
"""Тесты для системы ролей с полной строгой типизацией"""
|
||||
|
||||
async def test_role_creation(self, test_db: BotDatabase, test_session: AsyncSession) -> None:
|
||||
"""
|
||||
Тест создания ролей.
|
||||
Проверяет, что тестовые роли существуют в базе.
|
||||
"""
|
||||
stmt = select(Role)
|
||||
result = await test_session.execute(stmt)
|
||||
roles: List[Role] = result.scalars().all()
|
||||
|
||||
assert len(roles) >= 5
|
||||
role_names: List[str] = [role.name for role in roles]
|
||||
assert "Альбедо" in role_names
|
||||
assert "Нахида" in role_names
|
||||
assert "Кафка" in role_names
|
||||
|
||||
async def test_assign_role(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест назначения роли пользователю.
|
||||
Проверяет успешное назначение свободной роли и правильное сохранение в БД.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
# Освобождаем роль на всякий случай
|
||||
await test_db.release_role("Альбедо")
|
||||
|
||||
# Назначаем роль
|
||||
success: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
assert success, "Не удалось назначить роль"
|
||||
|
||||
# Проверяем, что роль действительно назначена
|
||||
stmt = select(Role).where(Role.name == "Альбедо")
|
||||
result = await test_session.execute(stmt)
|
||||
role: Role = result.scalar_one()
|
||||
|
||||
assert role.occupied_by == user_id
|
||||
|
||||
async def test_assign_occupied_role(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест назначения уже занятой роли.
|
||||
Проверяет, что нельзя назначить роль, если она уже занята другим пользователем.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
other_user_id: int = 999000111
|
||||
|
||||
await test_db.release_role("Альбедо")
|
||||
await test_db.add_user(other_user_id, "other_user", "Other User")
|
||||
|
||||
# Назначаем роль первому пользователю
|
||||
success_first: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
assert success_first, "Не удалось назначить роль первому пользователю"
|
||||
|
||||
# Пытаемся назначить ту же роль другому пользователю
|
||||
success_second: bool = await test_db.assign_role("Альбедо", other_user_id)
|
||||
assert not success_second, "Нельзя назначить занятую роль"
|
||||
|
||||
async def test_release_role(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест освобождения роли.
|
||||
Проверяет, что роль успешно освобождается.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
await test_db.release_role("Нахида")
|
||||
success_assign: bool = await test_db.assign_role("Нахида", user_id)
|
||||
assert success_assign
|
||||
|
||||
success_release: bool = await test_db.release_role("Нахида")
|
||||
assert success_release
|
||||
|
||||
stmt = select(Role).where(Role.name == "Нахида")
|
||||
result = await test_session.execute(stmt)
|
||||
role: Role = result.scalar_one()
|
||||
assert role.occupied_by is None
|
||||
|
||||
async def test_release_unoccupied_role(self, test_db: BotDatabase) -> None:
|
||||
"""
|
||||
Тест освобождения свободной роли.
|
||||
Проверяет, что нельзя освободить уже свободную роль.
|
||||
"""
|
||||
await test_db.release_role("Кафка")
|
||||
success: bool = await test_db.release_role("Кафка")
|
||||
assert not success, "Нельзя освободить свободную роль"
|
||||
|
||||
async def test_get_user_roles(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест получения ролей пользователя.
|
||||
Проверяет, что возвращается корректный список назначенных ролей.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
await test_db.release_role("Альбедо")
|
||||
await test_db.release_role("Нахида")
|
||||
|
||||
success1: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
success2: bool = await test_db.assign_role("Нахида", user_id)
|
||||
assert success1 and success2
|
||||
|
||||
roles: List[str] = await test_db.get_roles_by_user(user_id)
|
||||
assert len(roles) == 2
|
||||
assert "Альбедо" in roles
|
||||
assert "Нахида" in roles
|
||||
|
||||
async def test_get_available_roles(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест получения доступных ролей.
|
||||
Проверяет, что назначенные роли не включены в список свободных.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
# Освобождаем все роли
|
||||
for role_name in ["Альбедо", "Нахида", "Кафка", "Броння", "Чжун Ли"]:
|
||||
await test_db.release_role(role_name)
|
||||
|
||||
# Назначаем одну роль
|
||||
success: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
assert success
|
||||
|
||||
available_roles: List[Role] = await test_db.get_available_roles()
|
||||
role_names: List[str] = [role.name for role in available_roles]
|
||||
|
||||
assert "Альбедо" not in role_names
|
||||
assert len(available_roles) > 0
|
||||
|
||||
for role in available_roles:
|
||||
assert role.occupied_by is None
|
||||
|
||||
async def test_get_occupied_roles(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест получения занятых ролей.
|
||||
Проверяет, что все назначенные роли возвращаются корректно.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
await test_db.release_role("Альбедо")
|
||||
await test_db.release_role("Нахида")
|
||||
|
||||
success1: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
success2: bool = await test_db.assign_role("Нахида", user_id)
|
||||
assert success1 and success2
|
||||
|
||||
occupied_roles: List[Role] = await test_db.get_occupied_roles()
|
||||
role_names: List[str] = [role.name for role in occupied_roles]
|
||||
|
||||
assert "Альбедо" in role_names
|
||||
assert "Нахида" in role_names
|
||||
assert len(occupied_roles) >= 2
|
||||
|
||||
async def test_region_filter(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession
|
||||
) -> None:
|
||||
"""
|
||||
Тест фильтрации ролей по регионам.
|
||||
Проверяет, что метод возвращает роли только указанного региона.
|
||||
"""
|
||||
await test_db.release_role("Альбедо")
|
||||
mondstadt_roles: List[Role] = await test_db.get_available_roles(RoleRegion.MONDSTADT)
|
||||
|
||||
assert len(mondstadt_roles) == 1
|
||||
assert mondstadt_roles[0].name == "Альбедо"
|
||||
assert mondstadt_roles[0].region == RoleRegion.MONDSTADT
|
||||
|
||||
async def test_region_stats(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест статистики по регионам.
|
||||
Проверяет, что метод возвращает корректное количество занятых ролей по регионам.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
|
||||
await test_db.release_role("Альбедо")
|
||||
await test_db.release_role("Нахида")
|
||||
|
||||
success1: bool = await test_db.assign_role("Альбедо", user_id)
|
||||
success2: bool = await test_db.assign_role("Нахида", user_id)
|
||||
assert success1 and success2
|
||||
|
||||
stats: Dict[RoleRegion, Dict[str, int]] = await test_db.get_region_stats()
|
||||
|
||||
assert RoleRegion.MONDSTADT in stats
|
||||
assert RoleRegion.SUMERU in stats
|
||||
assert stats[RoleRegion.MONDSTADT]["occupied"] == 1
|
||||
assert stats[RoleRegion.MONDSTADT]["total"] == 1
|
||||
@@ -1,193 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select, Sequence
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from database import User, UserMessage, BotDatabase
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestUserStatistics:
|
||||
"""Тесты для статистики пользователей с полной строгой типизацией"""
|
||||
|
||||
async def test_add_user(self, test_db: BotDatabase, test_session: AsyncSession) -> None:
|
||||
"""
|
||||
Тест добавления пользователя.
|
||||
Проверяет, что пользователь создаётся с правильными данными и статусом 'active'.
|
||||
"""
|
||||
user_id: int = 111222333
|
||||
|
||||
await test_db.add_user(
|
||||
user_id=user_id,
|
||||
username="new_user",
|
||||
full_name="New User"
|
||||
)
|
||||
|
||||
user: User | None = await test_session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.username == "new_user"
|
||||
assert user.status.value == "active"
|
||||
|
||||
async def test_add_message_creates_user(
|
||||
self, test_db: BotDatabase, test_session: AsyncSession
|
||||
) -> None:
|
||||
"""
|
||||
Тест, что добавление сообщения создаёт пользователя, если его нет.
|
||||
Проверяет, что пользователь и сообщение корректно создаются.
|
||||
"""
|
||||
user_id: int = 111222333
|
||||
|
||||
await test_db.add_message(
|
||||
user_id=user_id,
|
||||
message_text="Тестовое сообщение"
|
||||
)
|
||||
|
||||
user: User | None = await test_session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "active"
|
||||
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await test_session.execute(stmt)
|
||||
messages: Sequence[UserMessage] = result.scalars().all()
|
||||
|
||||
assert len(messages) == 1
|
||||
assert messages[0].message_text == "Тестовое сообщение"
|
||||
|
||||
async def test_message_stats_calculation(
|
||||
self, test_db: BotDatabase, test_user_with_messages: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест расчёта статистики сообщений пользователя.
|
||||
Проверяет корректность статистики по дням, неделям, месяцам и общему количеству сообщений.
|
||||
"""
|
||||
user_id: int = test_user_with_messages
|
||||
|
||||
# Получаем статистику
|
||||
day: int
|
||||
week: int
|
||||
month: int
|
||||
total: int
|
||||
day, week, month, total = await test_db.get_message_stats(user_id)
|
||||
|
||||
assert total >= 10, f"Ожидается минимум 10 сообщений, получено {total}"
|
||||
assert day >= 0
|
||||
assert week >= 0
|
||||
assert month >= 0
|
||||
assert total >= 0
|
||||
assert day <= week <= month <= total
|
||||
|
||||
async def test_message_stats_with_dates(
|
||||
self, test_db: BotDatabase, test_user: int
|
||||
) -> None:
|
||||
"""
|
||||
Тест статистики с конкретными известными датами сообщений.
|
||||
Проверяет подсчёт сообщений за день, неделю, месяц и общее количество.
|
||||
"""
|
||||
user_id: int = test_user
|
||||
now: datetime = datetime.now(timezone.utc)
|
||||
|
||||
# Очищаем старые сообщения
|
||||
async with test_db.session_factory() as session:
|
||||
stmt = select(UserMessage).where(UserMessage.user_id == user_id)
|
||||
result = await session.execute(stmt)
|
||||
old_messages: Sequence[UserMessage] = result.scalars().all()
|
||||
for msg in old_messages:
|
||||
await session.delete(msg)
|
||||
await session.commit()
|
||||
|
||||
# Создаём сообщения с фиксированными датами
|
||||
test_messages: list[tuple[datetime, str]] = [
|
||||
(now - timedelta(days=45), "45 дней назад"),
|
||||
(now - timedelta(days=30), "30 дней назад"),
|
||||
(now - timedelta(days=15), "15 дней назад"),
|
||||
(now - timedelta(days=7), "7 дней назад"),
|
||||
(now - timedelta(days=3), "3 дня назад"),
|
||||
(now - timedelta(hours=6), "6 часов назад"),
|
||||
(now, "сейчас")
|
||||
]
|
||||
|
||||
for date, text in test_messages:
|
||||
await test_db.add_message(user_id, text, date)
|
||||
|
||||
day: int
|
||||
week: int
|
||||
month: int
|
||||
total: int
|
||||
day, week, month, total = await test_db.get_message_stats(user_id)
|
||||
|
||||
assert total == 7, f"Ожидалось 7 сообщений, получено {total}"
|
||||
|
||||
day_start: datetime = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
expected_day: int = sum(1 for date, _ in test_messages if date >= day_start)
|
||||
assert day == expected_day, f"За день: ожидалось {expected_day}, получено {day}"
|
||||
|
||||
monday: datetime = (now - timedelta(days=now.weekday())).replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
expected_week: int = sum(1 for date, _ in test_messages if date >= monday)
|
||||
assert week == expected_week, f"За неделю: ожидалось {expected_week}, получено {week}"
|
||||
|
||||
month_start: datetime = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
||||
expected_month: int = sum(1 for date, _ in test_messages if date >= month_start)
|
||||
assert month == expected_month, f"За месяц: ожидалось {expected_month}, получено {month}"
|
||||
|
||||
async def test_empty_user_stats(self, test_db: BotDatabase) -> None:
|
||||
"""
|
||||
Тест статистики для пользователя без сообщений.
|
||||
Все значения должны быть равны нулю.
|
||||
"""
|
||||
user_id: int = 0o00111222
|
||||
await test_db.add_user(user_id, "empty_user", "Empty User")
|
||||
|
||||
day: int
|
||||
week: int
|
||||
month: int
|
||||
total: int
|
||||
day, week, month, total = await test_db.get_message_stats(user_id)
|
||||
|
||||
assert day == 0
|
||||
assert week == 0
|
||||
assert month == 0
|
||||
assert total == 0
|
||||
|
||||
async def test_user_management(self, test_db: BotDatabase) -> None:
|
||||
"""
|
||||
Тест управления пользователями.
|
||||
Проверяет добавление, назначение админа, бан/разбан и возврат статуса пользователя.
|
||||
"""
|
||||
user_id: int = 555666777
|
||||
|
||||
# Добавление пользователя
|
||||
await test_db.add_user(user_id, "managed_user", "Managed User")
|
||||
|
||||
async with test_db.session_factory() as session:
|
||||
user: User | None = await session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "active"
|
||||
|
||||
# Назначение админом
|
||||
await test_db.set_admin(user_id, True)
|
||||
async with test_db.session_factory() as session:
|
||||
user = await session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "admin"
|
||||
|
||||
# Бан пользователя
|
||||
await test_db.ban_user(user_id)
|
||||
async with test_db.session_factory() as session:
|
||||
user = await session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "banned"
|
||||
|
||||
# Разбан
|
||||
await test_db.unban_user(user_id)
|
||||
async with test_db.session_factory() as session:
|
||||
user = await session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "active"
|
||||
|
||||
# Снятие админки
|
||||
await test_db.set_admin(user_id, False)
|
||||
async with test_db.session_factory() as session:
|
||||
user = await session.get(User, user_id)
|
||||
assert user is not None
|
||||
assert user.status.value == "active"
|
||||
Reference in New Issue
Block a user