import pytest import asyncio from datetime import datetime from unittest.mock import AsyncMock, MagicMock from database.repositories.post_repository import PostRepository from database.models import TelegramPost, PostContent, MessageContentLink class TestPostRepository: """Тесты для PostRepository.""" @pytest.fixture def mock_db_path(self): """Фикстура для пути к тестовой БД.""" return ":memory:" @pytest.fixture def post_repository(self, mock_db_path): """Фикстура для PostRepository.""" return PostRepository(mock_db_path) @pytest.fixture def sample_post(self): """Фикстура для тестового поста.""" return TelegramPost( message_id=12345, text="Тестовый пост", author_id=67890, helper_text_message_id=None, created_at=int(datetime.now().timestamp()) ) @pytest.fixture def sample_post_no_date(self): """Фикстура для тестового поста без даты.""" return TelegramPost( message_id=12346, text="Тестовый пост без даты", author_id=67890, helper_text_message_id=None, created_at=None, status="suggest", ) @pytest.fixture def sample_post_content(self): """Фикстура для тестового контента поста.""" return PostContent( message_id=12345, content_name="/path/to/file.jpg", content_type="photo" ) @pytest.fixture def sample_message_link(self): """Фикстура для тестовой связи сообщения с контентом.""" return MessageContentLink( post_id=12345, message_id=67890 ) @pytest.mark.asyncio async def test_create_tables(self, post_repository): """Тест создания таблиц.""" # Мокаем _execute_query post_repository._execute_query = AsyncMock() await post_repository.create_tables() # Проверяем, что create_tables вызвался 3 раза (для каждой таблицы) assert post_repository._execute_query.call_count == 3 # Проверяем создание таблицы постов calls = post_repository._execute_query.call_args_list post_table_call = calls[0][0][0] assert "CREATE TABLE IF NOT EXISTS post_from_telegram_suggest" in post_table_call assert "message_id INTEGER NOT NULL PRIMARY KEY" in post_table_call assert "created_at INTEGER NOT NULL" in post_table_call assert "status TEXT NOT NULL DEFAULT 'suggest'" in post_table_call assert "FOREIGN KEY (author_id) REFERENCES our_users (user_id) ON DELETE CASCADE" in post_table_call # Проверяем создание таблицы контента content_table_call = calls[1][0][0] assert "CREATE TABLE IF NOT EXISTS content_post_from_telegram" in content_table_call assert "PRIMARY KEY (message_id, content_name)" in content_table_call # Проверяем создание таблицы связей link_table_call = calls[2][0][0] assert "CREATE TABLE IF NOT EXISTS message_link_to_content" in link_table_call assert "PRIMARY KEY (post_id, message_id)" in link_table_call @pytest.mark.asyncio async def test_add_post_with_date(self, post_repository, sample_post): """Тест добавления поста с датой.""" # Мокаем _execute_query post_repository._execute_query = AsyncMock() await post_repository.add_post(sample_post) post_repository._execute_query.assert_called_once() call_args = post_repository._execute_query.call_args query = call_args[0][0] params = call_args[0][1] assert "INSERT INTO post_from_telegram_suggest" in query assert "status" in query assert "VALUES (?, ?, ?, ?, ?)" in query assert params == ( sample_post.message_id, sample_post.text, sample_post.author_id, sample_post.created_at, sample_post.status, ) @pytest.mark.asyncio async def test_add_post_without_date(self, post_repository, sample_post_no_date): """Тест добавления поста без даты (должна генерироваться автоматически).""" # Мокаем _execute_query post_repository._execute_query = AsyncMock() await post_repository.add_post(sample_post_no_date) # Проверяем, что дата была установлена assert sample_post_no_date.created_at is not None assert isinstance(sample_post_no_date.created_at, int) assert sample_post_no_date.created_at > 0 post_repository._execute_query.assert_called_once() call_args = post_repository._execute_query.call_args params = call_args[0][1] assert params[3] == sample_post_no_date.created_at # created_at assert params[4] == sample_post_no_date.status # status (default suggest) @pytest.mark.asyncio async def test_add_post_logs_correctly(self, post_repository, sample_post): """Тест логирования при добавлении поста.""" # Мокаем _execute_query и logger post_repository._execute_query = AsyncMock() post_repository.logger = MagicMock() await post_repository.add_post(sample_post) post_repository.logger.info.assert_called_once_with( f"Пост добавлен: message_id={sample_post.message_id}" ) @pytest.mark.asyncio async def test_update_helper_message(self, post_repository): """Тест обновления helper сообщения.""" # Мокаем _execute_query post_repository._execute_query = AsyncMock() message_id = 12345 helper_message_id = 67890 await post_repository.update_helper_message(message_id, helper_message_id) post_repository._execute_query.assert_called_once() call_args = post_repository._execute_query.call_args query = call_args[0][0] params = call_args[0][1] assert "UPDATE post_from_telegram_suggest SET helper_text_message_id = ? WHERE message_id = ?" in query assert params == (helper_message_id, message_id) @pytest.mark.asyncio async def test_update_status_by_message_id(self, post_repository): """Тест обновления статуса поста по message_id.""" post_repository._execute_query = AsyncMock() post_repository.logger = MagicMock() message_id = 12345 status = "approved" await post_repository.update_status_by_message_id(message_id, status) post_repository._execute_query.assert_called_once() call_args = post_repository._execute_query.call_args query = call_args[0][0] params = call_args[0][1] assert "UPDATE post_from_telegram_suggest" in query assert "SET status = ? WHERE message_id = ?" in query assert params == (status, message_id) post_repository.logger.info.assert_called_once() @pytest.mark.asyncio async def test_update_status_for_media_group_by_helper_id(self, post_repository): """Тест обновления статуса медиагруппы по helper_message_id.""" post_repository._execute_query = AsyncMock() post_repository.logger = MagicMock() helper_message_id = 99999 status = "declined" await post_repository.update_status_for_media_group_by_helper_id( helper_message_id, status ) post_repository._execute_query.assert_called_once() call_args = post_repository._execute_query.call_args query = call_args[0][0] params = call_args[0][1] assert "UPDATE post_from_telegram_suggest" in query assert "SET status = ?" in query assert "message_id = ? OR helper_text_message_id = ?" in query assert params == (status, helper_message_id, helper_message_id) post_repository.logger.info.assert_called_once() @pytest.mark.asyncio async def test_add_post_content_success(self, post_repository): """Тест успешного добавления контента поста.""" # Мокаем _execute_query post_repository._execute_query = AsyncMock() post_repository.logger = MagicMock() post_id = 12345 message_id = 67890 content_name = "/path/to/file.jpg" content_type = "photo" result = await post_repository.add_post_content(post_id, message_id, content_name, content_type) # Проверяем, что результат True assert result is True # Проверяем, что _execute_query вызвался 2 раза (для связи и контента) assert post_repository._execute_query.call_count == 2 # Проверяем вызов для связи link_call = post_repository._execute_query.call_args_list[0] link_query = link_call[0][0] link_params = link_call[0][1] assert "INSERT OR IGNORE INTO message_link_to_content" in link_query assert link_params == (post_id, message_id) # Проверяем вызов для контента content_call = post_repository._execute_query.call_args_list[1] content_query = content_call[0][0] content_params = content_call[0][1] assert "INSERT OR IGNORE INTO content_post_from_telegram" in content_query assert content_params == (message_id, content_name, content_type) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Контент поста добавлен: post_id={post_id}, message_id={message_id}" ) @pytest.mark.asyncio async def test_add_post_content_exception(self, post_repository): """Тест обработки исключения при добавлении контента поста.""" # Мокаем _execute_query чтобы вызвать исключение post_repository._execute_query = AsyncMock(side_effect=Exception("Database error")) post_repository.logger = MagicMock() post_id = 12345 message_id = 67890 content_name = "/path/to/file.jpg" content_type = "photo" result = await post_repository.add_post_content(post_id, message_id, content_name, content_type) # Проверяем, что результат False assert result is False # Проверяем логирование ошибки post_repository.logger.error.assert_called_once() error_call = post_repository.logger.error.call_args[0][0] assert "Ошибка при добавлении контента поста:" in error_call @pytest.mark.asyncio async def test_get_post_content_by_helper_id(self, post_repository): """Тест получения контента поста по helper ID.""" # Мокаем _execute_query_with_result mock_result = [ ("/path/to/photo1.jpg", "photo"), ("/path/to/video1.mp4", "video"), ("/path/to/photo2.jpg", "photo") ] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 67890 result = await post_repository.get_post_content_by_helper_id(helper_message_id) # Проверяем результат assert result == mock_result # Проверяем вызов _execute_query_with_result post_repository._execute_query_with_result.assert_called_once() call_args = post_repository._execute_query_with_result.call_args query = call_args[0][0] params = call_args[0][1] assert "SELECT cpft.content_name, cpft.content_type" in query assert "WHERE pft.helper_text_message_id = ?" in query assert params == (helper_message_id,) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Получен контент поста: {len(mock_result)} элементов" ) @pytest.mark.asyncio async def test_get_post_text_by_helper_id_found(self, post_repository): """Тест получения текста поста по helper ID (пост найден).""" # Мокаем _execute_query_with_result mock_result = [("Тестовый текст поста",)] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 67890 result = await post_repository.get_post_text_by_helper_id(helper_message_id) # Проверяем результат assert result == "Тестовый текст поста" # Проверяем вызов _execute_query_with_result post_repository._execute_query_with_result.assert_called_once() call_args = post_repository._execute_query_with_result.call_args query = call_args[0][0] params = call_args[0][1] assert "SELECT text FROM post_from_telegram_suggest WHERE helper_text_message_id = ?" in query assert params == (helper_message_id,) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Получен текст поста для helper_message_id={helper_message_id}" ) @pytest.mark.asyncio async def test_get_post_text_by_helper_id_not_found(self, post_repository): """Тест получения текста поста по helper ID (пост не найден).""" # Мокаем _execute_query_with_result mock_result = [] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 67890 result = await post_repository.get_post_text_by_helper_id(helper_message_id) # Проверяем результат assert result is None # Проверяем, что logger.info не вызывался post_repository.logger.info.assert_not_called() @pytest.mark.asyncio async def test_get_post_ids_by_helper_id(self, post_repository): """Тест получения ID сообщений по helper ID.""" # Мокаем _execute_query_with_result mock_result = [(12345,), (67890,), (11111,)] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 67890 result = await post_repository.get_post_ids_by_helper_id(helper_message_id) # Проверяем результат assert result == [12345, 67890, 11111] # Проверяем вызов _execute_query_with_result post_repository._execute_query_with_result.assert_called_once() call_args = post_repository._execute_query_with_result.call_args query = call_args[0][0] params = call_args[0][1] assert "SELECT mltc.message_id" in query assert "WHERE pft.helper_text_message_id = ?" in query assert params == (helper_message_id,) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Получены ID сообщений: {len(mock_result)} элементов" ) @pytest.mark.asyncio async def test_get_author_id_by_message_id_found(self, post_repository): """Тест получения ID автора по message ID (автор найден).""" # Мокаем _execute_query_with_result mock_result = [(67890,)] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() message_id = 12345 result = await post_repository.get_author_id_by_message_id(message_id) # Проверяем результат assert result == 67890 # Проверяем вызов _execute_query_with_result post_repository._execute_query_with_result.assert_called_once() call_args = post_repository._execute_query_with_result.call_args query = call_args[0][0] params = call_args[0][1] assert "SELECT author_id FROM post_from_telegram_suggest WHERE message_id = ?" in query assert params == (message_id,) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Получен author_id: {67890} для message_id={message_id}" ) @pytest.mark.asyncio async def test_get_author_id_by_message_id_not_found(self, post_repository): """Тест получения ID автора по message ID (автор не найден).""" # Мокаем _execute_query_with_result mock_result = [] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() message_id = 12345 result = await post_repository.get_author_id_by_message_id(message_id) # Проверяем результат assert result is None # Проверяем, что logger.info не вызывался post_repository.logger.info.assert_not_called() @pytest.mark.asyncio async def test_get_author_id_by_helper_message_id_found(self, post_repository): """Тест получения ID автора по helper message ID (автор найден).""" # Мокаем _execute_query_with_result mock_result = [(67890,)] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 12345 result = await post_repository.get_author_id_by_helper_message_id(helper_message_id) # Проверяем результат assert result == 67890 # Проверяем вызов _execute_query_with_result post_repository._execute_query_with_result.assert_called_once() call_args = post_repository._execute_query_with_result.call_args query = call_args[0][0] params = call_args[0][1] assert "SELECT author_id FROM post_from_telegram_suggest WHERE helper_text_message_id = ?" in query assert params == (helper_message_id,) # Проверяем логирование post_repository.logger.info.assert_called_once_with( f"Получен author_id: {67890} для helper_message_id={helper_message_id}" ) @pytest.mark.asyncio async def test_get_author_id_by_helper_message_id_not_found(self, post_repository): """Тест получения ID автора по helper message ID (автор не найден).""" # Мокаем _execute_query_with_result mock_result = [] post_repository._execute_query_with_result = AsyncMock(return_value=mock_result) post_repository.logger = MagicMock() helper_message_id = 12345 result = await post_repository.get_author_id_by_helper_message_id(helper_message_id) # Проверяем результат assert result is None # Проверяем, что logger.info не вызывался post_repository.logger.info.assert_not_called() @pytest.mark.asyncio async def test_create_tables_logs_success(self, post_repository): """Тест логирования успешного создания таблиц.""" # Мокаем _execute_query и logger post_repository._execute_query = AsyncMock() post_repository.logger = MagicMock() await post_repository.create_tables() post_repository.logger.info.assert_called_once_with("Таблицы для постов созданы")