Andrey
e2b1353408
- Создана система отслеживания миграций (MigrationRepository, таблица migrations) - Добавлен скрипт apply_migrations.py для автоматического применения миграций - Созданы CI/CD пайплайны (.github/workflows/ci.yml, deploy.yml) - Обновлена документация по миграциям в database-patterns.md - Миграции применяются автоматически при деплое в продакшн
300 lines
17 KiB
Python
300 lines
17 KiB
Python
"""Tests for PostService"""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
|
|
|
import pytest
|
|
from aiogram import types
|
|
from database.models import TelegramPost, User
|
|
from helper_bot.handlers.private.services import BotSettings, PostService
|
|
|
|
|
|
class TestPostService:
|
|
"""Test class for PostService"""
|
|
|
|
@pytest.fixture
|
|
def mock_db(self):
|
|
"""Mock database"""
|
|
db = Mock()
|
|
db.add_post = AsyncMock()
|
|
db.update_helper_message = AsyncMock()
|
|
db.get_user_by_id = AsyncMock()
|
|
db.add_message_link = AsyncMock()
|
|
return db
|
|
|
|
@pytest.fixture
|
|
def mock_settings(self):
|
|
"""Mock bot settings"""
|
|
return BotSettings(
|
|
group_for_posts="test_posts",
|
|
group_for_message="test_message",
|
|
main_public="test_public",
|
|
group_for_logs="test_logs",
|
|
important_logs="test_important",
|
|
preview_link="test_link",
|
|
logs="test_logs_setting",
|
|
test="test_test_setting"
|
|
)
|
|
|
|
@pytest.fixture
|
|
def post_service(self, mock_db, mock_settings):
|
|
"""Create PostService instance"""
|
|
return PostService(mock_db, mock_settings)
|
|
|
|
@pytest.fixture
|
|
def mock_message(self):
|
|
"""Mock Telegram message"""
|
|
message = Mock(spec=types.Message)
|
|
from_user = Mock()
|
|
from_user.id = 12345
|
|
from_user.first_name = "Test"
|
|
from_user.username = "testuser"
|
|
from_user.full_name = "Test User"
|
|
message.from_user = from_user
|
|
message.text = "Тестовый пост"
|
|
message.message_id = 100
|
|
message.bot = AsyncMock()
|
|
message.chat = Mock()
|
|
message.chat.id = 12345
|
|
return message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_text_post_saves_raw_text(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_text_post saves raw text to database"""
|
|
mock_sent_message = Mock()
|
|
mock_sent_message.message_id = 200
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted text"):
|
|
with patch('helper_bot.handlers.private.services.send_text_message', return_value=mock_sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
|
|
await post_service.handle_text_post(mock_message, "Test")
|
|
|
|
# Check that add_post was called
|
|
mock_db.add_post.assert_called_once()
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
|
|
# Check that raw text is saved
|
|
assert isinstance(call_args, TelegramPost)
|
|
assert call_args.text == "Тестовый пост" # Raw text
|
|
assert call_args.message_id == 200
|
|
assert call_args.author_id == 12345
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_text_post_determines_anonymity(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_text_post determines anonymity correctly"""
|
|
mock_message.text = "Тестовый пост анон"
|
|
mock_sent_message = Mock()
|
|
mock_sent_message.message_id = 200
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted text"):
|
|
with patch('helper_bot.handlers.private.services.send_text_message', return_value=mock_sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=True):
|
|
|
|
await post_service.handle_text_post(mock_message, "Test")
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.is_anonymous is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_photo_post_saves_raw_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_photo_post saves raw caption to database"""
|
|
mock_message.caption = "Тестовая подпись"
|
|
mock_message.photo = [Mock()]
|
|
mock_message.photo[-1].file_id = "photo_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 201
|
|
sent_message.caption = "Formatted caption"
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted caption"):
|
|
with patch('helper_bot.handlers.private.services.send_photo_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_photo_post(mock_message, "Test")
|
|
|
|
mock_db.add_post.assert_called_once()
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
|
|
# Check that raw caption is saved
|
|
assert call_args.text == "Тестовая подпись" # Raw caption
|
|
assert call_args.message_id == 201
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_photo_post_without_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_photo_post handles missing caption"""
|
|
mock_message.caption = None
|
|
mock_message.photo = [Mock()]
|
|
mock_message.photo[-1].file_id = "photo_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 202
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value=""):
|
|
with patch('helper_bot.handlers.private.services.send_photo_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_photo_post(mock_message, "Test")
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.text == "" # Empty string for missing caption
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_video_post_saves_raw_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_video_post saves raw caption to database"""
|
|
mock_message.caption = "Видео подпись"
|
|
mock_message.video = Mock()
|
|
mock_message.video.file_id = "video_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 203
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted"):
|
|
with patch('helper_bot.handlers.private.services.send_video_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=True):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_video_post(mock_message, "Test")
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.text == "Видео подпись" # Raw caption
|
|
assert call_args.is_anonymous is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_audio_post_saves_raw_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_audio_post saves raw caption to database"""
|
|
mock_message.caption = "Аудио подпись"
|
|
mock_message.audio = Mock()
|
|
mock_message.audio.file_id = "audio_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 204
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted"):
|
|
with patch('helper_bot.handlers.private.services.send_audio_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_audio_post(mock_message, "Test")
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.text == "Аудио подпись" # Raw caption
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_video_note_post_saves_empty_string(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_video_note_post saves empty string"""
|
|
mock_message.video_note = Mock()
|
|
mock_message.video_note.file_id = "video_note_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 205
|
|
|
|
with patch('helper_bot.handlers.private.services.send_video_note_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_video_note_post(mock_message)
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.text == "" # Empty string
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_voice_post_saves_empty_string(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_voice_post saves empty string"""
|
|
mock_message.voice = Mock()
|
|
mock_message.voice.file_id = "voice_123"
|
|
|
|
sent_message = Mock()
|
|
sent_message.message_id = 206
|
|
|
|
with patch('helper_bot.handlers.private.services.send_voice_message', return_value=sent_message):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('helper_bot.handlers.private.services.add_in_db_media', return_value=True):
|
|
|
|
await post_service.handle_voice_post(mock_message)
|
|
|
|
call_args = mock_db.add_post.call_args[0][0]
|
|
assert call_args.text == "" # Empty string
|
|
assert call_args.is_anonymous is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_media_group_post_saves_raw_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_media_group_post saves raw caption to database"""
|
|
mock_message.message_id = 300
|
|
mock_message.media_group_id = 1
|
|
|
|
album = [Mock()]
|
|
album[0].caption = "Медиагруппа подпись"
|
|
|
|
mock_helper_message = Mock()
|
|
mock_helper_message.message_id = 302
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value="Formatted"):
|
|
with patch('helper_bot.handlers.private.services.prepare_media_group_from_middlewares', return_value=[]):
|
|
with patch('helper_bot.handlers.private.services.send_media_group_message_to_private_chat', return_value=[301]):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.send_text_message', return_value=mock_helper_message):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=True):
|
|
with patch('asyncio.sleep', new_callable=AsyncMock):
|
|
|
|
await post_service.handle_media_group_post(mock_message, album, "Test")
|
|
|
|
# Check main post
|
|
calls = mock_db.add_post.call_args_list
|
|
main_post = calls[0][0][0]
|
|
|
|
assert main_post.text == "Медиагруппа подпись" # Raw caption
|
|
assert main_post.message_id == 301 # Последний message_id из списка
|
|
assert main_post.is_anonymous is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_media_group_post_without_caption(self, post_service, mock_message, mock_db):
|
|
"""Test that handle_media_group_post handles missing caption"""
|
|
mock_message.message_id = 301
|
|
mock_message.media_group_id = 1
|
|
|
|
album = [Mock()]
|
|
album[0].caption = None
|
|
|
|
mock_helper_message = Mock()
|
|
mock_helper_message.message_id = 303
|
|
|
|
with patch('helper_bot.handlers.private.services.get_text_message', return_value=" "):
|
|
with patch('helper_bot.handlers.private.services.prepare_media_group_from_middlewares', return_value=[]):
|
|
with patch('helper_bot.handlers.private.services.send_media_group_message_to_private_chat', return_value=[302]):
|
|
with patch('helper_bot.handlers.private.services.get_first_name', return_value="Test"):
|
|
with patch('helper_bot.handlers.private.services.get_reply_keyboard_for_post', return_value=None):
|
|
with patch('helper_bot.handlers.private.services.send_text_message', return_value=mock_helper_message):
|
|
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
|
|
with patch('asyncio.sleep', new_callable=AsyncMock):
|
|
|
|
await post_service.handle_media_group_post(mock_message, album, "Test")
|
|
|
|
calls = mock_db.add_post.call_args_list
|
|
main_post = calls[0][0][0]
|
|
|
|
assert main_post.text == "" # Empty string for missing caption
|
|
assert main_post.is_anonymous is False
|