style: isort + black
This commit is contained in:
@@ -4,17 +4,18 @@ 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('''
|
||||
await message_repository._execute_query("""
|
||||
CREATE TABLE IF NOT EXISTS our_users (
|
||||
user_id INTEGER NOT NULL PRIMARY KEY,
|
||||
first_name TEXT,
|
||||
@@ -28,36 +29,42 @@ class TestMessageRepositoryIntegration:
|
||||
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()))
|
||||
(
|
||||
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:
|
||||
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):
|
||||
"""Фикстура для тестового сообщения."""
|
||||
@@ -65,9 +72,9 @@ class TestMessageRepositoryIntegration:
|
||||
message_text="Интеграционное тестовое сообщение",
|
||||
user_id=12345,
|
||||
telegram_message_id=67890,
|
||||
date=int(datetime.now().timestamp())
|
||||
date=int(datetime.now().timestamp()),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_message_no_date(self):
|
||||
"""Фикстура для тестового сообщения без даты."""
|
||||
@@ -75,137 +82,155 @@ class TestMessageRepositoryIntegration:
|
||||
message_text="Интеграционное тестовое сообщение без даты",
|
||||
user_id=12345,
|
||||
telegram_message_id=67891,
|
||||
date=None
|
||||
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())
|
||||
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):
|
||||
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)
|
||||
|
||||
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):
|
||||
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)
|
||||
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):
|
||||
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
|
||||
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)
|
||||
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):
|
||||
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())
|
||||
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)
|
||||
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())
|
||||
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)
|
||||
user_id = await message_repository.get_user_by_message_id(
|
||||
invalid_message.telegram_message_id
|
||||
)
|
||||
assert user_id is None
|
||||
except Exception:
|
||||
# Ожидаемое поведение при нарушении внешнего ключа
|
||||
|
||||
Reference in New Issue
Block a user