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__])