diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/database/__init__.py b/tests/database/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/database/conftest.py b/tests/database/conftest.py deleted file mode 100644 index 1b04bb0..0000000 --- a/tests/database/conftest.py +++ /dev/null @@ -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 diff --git a/tests/database/test_messages.py b/tests/database/test_messages.py deleted file mode 100644 index ba30d67..0000000 --- a/tests/database/test_messages.py +++ /dev/null @@ -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" diff --git a/tests/database/test_roles.py b/tests/database/test_roles.py deleted file mode 100644 index 7569b04..0000000 --- a/tests/database/test_roles.py +++ /dev/null @@ -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 diff --git a/tests/database/test_user_stats.py b/tests/database/test_user_stats.py deleted file mode 100644 index 448aefe..0000000 --- a/tests/database/test_user_stats.py +++ /dev/null @@ -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"