Andrey
3d6b4353f9
All checks were successful
CI pipeline / Test & Code Quality (push) Successful in 34s
242 lines
9.9 KiB
Python
242 lines
9.9 KiB
Python
import asyncio
|
||
import os
|
||
import tempfile
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from database.models import UserMessage
|
||
from database.repositories.message_repository import MessageRepository
|
||
|
||
|
||
class TestMessageRepositoryIntegration:
|
||
"""Интеграционные тесты для MessageRepository с реальной БД."""
|
||
|
||
async def _setup_test_database(self, message_repository):
|
||
"""Вспомогательная функция для настройки тестовой БД."""
|
||
# Сначала создаем таблицу our_users для тестов
|
||
await message_repository._execute_query("""
|
||
CREATE TABLE IF NOT EXISTS our_users (
|
||
user_id INTEGER NOT NULL PRIMARY KEY,
|
||
first_name TEXT,
|
||
full_name TEXT,
|
||
username TEXT,
|
||
is_bot BOOLEAN DEFAULT 0,
|
||
language_code TEXT,
|
||
has_stickers BOOLEAN DEFAULT 0 NOT NULL,
|
||
emoji TEXT,
|
||
date_added INTEGER NOT NULL,
|
||
date_changed INTEGER NOT NULL,
|
||
voice_bot_welcome_received BOOLEAN DEFAULT 0
|
||
)
|
||
""")
|
||
|
||
# Добавляем тестового пользователя
|
||
await message_repository._execute_query(
|
||
"INSERT OR REPLACE INTO our_users (user_id, first_name, full_name, date_added, date_changed) VALUES (?, ?, ?, ?, ?)",
|
||
(
|
||
12345,
|
||
"Test",
|
||
"Test User",
|
||
int(datetime.now().timestamp()),
|
||
int(datetime.now().timestamp()),
|
||
),
|
||
)
|
||
|
||
# Теперь создаем таблицу user_messages
|
||
await message_repository.create_tables()
|
||
|
||
@pytest.fixture
|
||
def temp_db_path(self):
|
||
"""Фикстура для временного пути к БД."""
|
||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
||
temp_path = f.name
|
||
|
||
yield temp_path
|
||
|
||
# Очистка после тестов
|
||
try:
|
||
os.unlink(temp_path)
|
||
except OSError:
|
||
pass
|
||
|
||
@pytest.fixture
|
||
def message_repository(self, temp_db_path):
|
||
"""Фикстура для MessageRepository с реальной БД."""
|
||
return MessageRepository(temp_db_path)
|
||
|
||
@pytest.fixture
|
||
def sample_message(self):
|
||
"""Фикстура для тестового сообщения."""
|
||
return UserMessage(
|
||
message_text="Интеграционное тестовое сообщение",
|
||
user_id=12345,
|
||
telegram_message_id=67890,
|
||
date=int(datetime.now().timestamp()),
|
||
)
|
||
|
||
@pytest.fixture
|
||
def sample_message_no_date(self):
|
||
"""Фикстура для тестового сообщения без даты."""
|
||
return UserMessage(
|
||
message_text="Интеграционное тестовое сообщение без даты",
|
||
user_id=12345,
|
||
telegram_message_id=67891,
|
||
date=None,
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables_integration(self, message_repository):
|
||
"""Интеграционный тест создания таблиц."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Проверяем, что таблица создана, пытаясь добавить сообщение
|
||
message = UserMessage(
|
||
message_text="Тест создания таблиц",
|
||
user_id=12345,
|
||
telegram_message_id=67890,
|
||
date=int(datetime.now().timestamp()),
|
||
)
|
||
|
||
# Не должно вызывать ошибку
|
||
await message_repository.add_message(message)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_and_retrieve_message_integration(
|
||
self, message_repository, sample_message
|
||
):
|
||
"""Интеграционный тест добавления и получения сообщения."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Добавляем сообщение
|
||
await message_repository.add_message(sample_message)
|
||
|
||
# Получаем пользователя по message_id
|
||
user_id = await message_repository.get_user_by_message_id(
|
||
sample_message.telegram_message_id
|
||
)
|
||
|
||
# Проверяем результат
|
||
assert user_id == sample_message.user_id
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_message_without_date_integration(
|
||
self, message_repository, sample_message_no_date
|
||
):
|
||
"""Интеграционный тест добавления сообщения без даты."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Добавляем сообщение без даты
|
||
await message_repository.add_message(sample_message_no_date)
|
||
|
||
# Проверяем, что дата была установлена
|
||
assert sample_message_no_date.date is not None
|
||
assert isinstance(sample_message_no_date.date, int)
|
||
assert sample_message_no_date.date > 0
|
||
|
||
# Проверяем, что сообщение можно найти
|
||
user_id = await message_repository.get_user_by_message_id(
|
||
sample_message_no_date.telegram_message_id
|
||
)
|
||
assert user_id == sample_message_no_date.user_id
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_by_message_id_not_found_integration(
|
||
self, message_repository
|
||
):
|
||
"""Интеграционный тест поиска несуществующего сообщения."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Ищем несуществующее сообщение
|
||
user_id = await message_repository.get_user_by_message_id(99999)
|
||
|
||
# Должно вернуть None
|
||
assert user_id is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_multiple_messages_integration(self, message_repository):
|
||
"""Интеграционный тест работы с несколькими сообщениями."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Добавляем несколько сообщений (используем существующий user_id 12345)
|
||
messages = [
|
||
UserMessage(
|
||
message_text=f"Сообщение {i}",
|
||
user_id=12345, # Используем существующий user_id
|
||
telegram_message_id=2000 + i,
|
||
date=int(datetime.now().timestamp()) + i,
|
||
)
|
||
for i in range(1, 4)
|
||
]
|
||
|
||
for message in messages:
|
||
await message_repository.add_message(message)
|
||
|
||
# Проверяем, что все сообщения можно найти
|
||
for message in messages:
|
||
user_id = await message_repository.get_user_by_message_id(
|
||
message.telegram_message_id
|
||
)
|
||
assert user_id == message.user_id
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_message_with_special_characters_integration(
|
||
self, message_repository
|
||
):
|
||
"""Интеграционный тест сообщения со специальными символами."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Сообщение со специальными символами
|
||
special_message = UserMessage(
|
||
message_text="Сообщение с 'кавычками' и \"двойными кавычками\" и эмодзи 😊",
|
||
user_id=12345,
|
||
telegram_message_id=67892,
|
||
date=int(datetime.now().timestamp()),
|
||
)
|
||
|
||
# Добавляем сообщение
|
||
await message_repository.add_message(special_message)
|
||
|
||
# Проверяем, что можно найти
|
||
user_id = await message_repository.get_user_by_message_id(
|
||
special_message.telegram_message_id
|
||
)
|
||
assert user_id == special_message.user_id
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_foreign_key_constraint_integration(self, message_repository):
|
||
"""Интеграционный тест ограничения внешнего ключа."""
|
||
# Настраиваем тестовую БД
|
||
await self._setup_test_database(message_repository)
|
||
|
||
# Пытаемся добавить сообщение с несуществующим user_id
|
||
invalid_message = UserMessage(
|
||
message_text="Сообщение с несуществующим пользователем",
|
||
user_id=99999, # Несуществующий пользователь
|
||
telegram_message_id=67893,
|
||
date=int(datetime.now().timestamp()),
|
||
)
|
||
|
||
# В SQLite с включенными внешними ключами это должно вызвать ошибку
|
||
# Теперь у нас есть таблица our_users, поэтому внешний ключ должен работать
|
||
try:
|
||
await message_repository.add_message(invalid_message)
|
||
# Если не вызвало ошибку, проверяем что сообщение не добавилось
|
||
user_id = await message_repository.get_user_by_message_id(
|
||
invalid_message.telegram_message_id
|
||
)
|
||
assert user_id is None
|
||
except Exception:
|
||
# Ожидаемое поведение при нарушении внешнего ключа
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__])
|