Реализован функцоинал хранения сырых текстов поста в базе данных. Оформление поста происходит непосредственно перед его отправкой в канал.

- Реализованы методы `get_post_text_and_anonymity_by_message_id` и `get_post_text_and_anonymity_by_helper_id` в `PostRepository` для получения текста поста и флага анонимности.
- Обновлена модель `TelegramPost`, добавлено поле `is_anonymous`.
- Изменена схема базы данных для включения поля `is_anonymous` в таблицу `post_from_telegram_suggest`.
- Обновлены функции публикации постов в `PostPublishService` для учета анонимности.
- Добавлены тесты для проверки новых функций и корректности работы с анонимностью.
This commit is contained in:
2026-01-23 12:12:21 +03:00
parent c6ba90552d
commit 89022aedaf
12 changed files with 1064 additions and 48 deletions

View File

@@ -77,6 +77,7 @@ class TestPostRepository:
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 "is_anonymous INTEGER" in post_table_call
assert "FOREIGN KEY (author_id) REFERENCES our_users (user_id) ON DELETE CASCADE" in post_table_call
# Проверяем создание таблицы контента
@@ -104,14 +105,17 @@ class TestPostRepository:
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,
)
assert "is_anonymous" in query
assert "VALUES (?, ?, ?, ?, ?, ?)" in query
# Проверяем параметры: message_id, text, author_id, created_at, status, is_anonymous
assert params[0] == sample_post.message_id
assert params[1] == sample_post.text
assert params[2] == sample_post.author_id
assert params[3] == sample_post.created_at
assert params[4] == sample_post.status
# is_anonymous преобразуется в int (None -> None, True -> 1, False -> 0)
expected_is_anonymous = None if sample_post.is_anonymous is None else (1 if sample_post.is_anonymous else 0)
assert params[5] == expected_is_anonymous
@pytest.mark.asyncio
async def test_add_post_without_date(self, post_repository, sample_post_no_date):
@@ -132,6 +136,8 @@ class TestPostRepository:
assert params[3] == sample_post_no_date.created_at # created_at
assert params[4] == sample_post_no_date.status # status (default suggest)
# Проверяем is_anonymous (должен быть в параметрах)
assert len(params) == 6 # Всего 6 параметров включая is_anonymous
@pytest.mark.asyncio
async def test_add_post_logs_correctly(self, post_repository, sample_post):
@@ -476,6 +482,169 @@ class TestPostRepository:
# Проверяем, что logger.info не вызывался
post_repository.logger.info.assert_not_called()
@pytest.mark.asyncio
async def test_get_post_text_and_anonymity_by_message_id_found(self, post_repository):
"""Тест получения текста и is_anonymous по message_id (пост найден)."""
# Мокаем _execute_query_with_result
mock_result = [("Тестовый текст", 1)] # is_anonymous = 1 (True)
post_repository._execute_query_with_result = AsyncMock(return_value=mock_result)
post_repository.logger = MagicMock()
message_id = 12345
result = await post_repository.get_post_text_and_anonymity_by_message_id(message_id)
# Проверяем результат
text, is_anonymous = result
assert text == "Тестовый текст"
assert is_anonymous is True
# Проверяем вызов _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, is_anonymous FROM post_from_telegram_suggest WHERE message_id = ?" in query
assert params == (message_id,)
@pytest.mark.asyncio
async def test_get_post_text_and_anonymity_by_message_id_with_false(self, post_repository):
"""Тест получения текста и is_anonymous по message_id (is_anonymous = False)."""
# Мокаем _execute_query_with_result
mock_result = [("Тестовый текст", 0)] # is_anonymous = 0 (False)
post_repository._execute_query_with_result = AsyncMock(return_value=mock_result)
message_id = 12345
result = await post_repository.get_post_text_and_anonymity_by_message_id(message_id)
# Проверяем результат
text, is_anonymous = result
assert text == "Тестовый текст"
assert is_anonymous is False
@pytest.mark.asyncio
async def test_get_post_text_and_anonymity_by_message_id_with_null(self, post_repository):
"""Тест получения текста и is_anonymous по message_id (is_anonymous = NULL)."""
# Мокаем _execute_query_with_result
mock_result = [("Тестовый текст", None)] # is_anonymous = NULL
post_repository._execute_query_with_result = AsyncMock(return_value=mock_result)
message_id = 12345
result = await post_repository.get_post_text_and_anonymity_by_message_id(message_id)
# Проверяем результат
text, is_anonymous = result
assert text == "Тестовый текст"
assert is_anonymous is None
@pytest.mark.asyncio
async def test_get_post_text_and_anonymity_by_message_id_not_found(self, post_repository):
"""Тест получения текста и is_anonymous по message_id (пост не найден)."""
# Мокаем _execute_query_with_result
mock_result = []
post_repository._execute_query_with_result = AsyncMock(return_value=mock_result)
message_id = 12345
result = await post_repository.get_post_text_and_anonymity_by_message_id(message_id)
# Проверяем результат
text, is_anonymous = result
assert text is None
assert is_anonymous is None
@pytest.mark.asyncio
async def test_get_post_text_and_anonymity_by_helper_id_found(self, post_repository):
"""Тест получения текста и is_anonymous по helper_message_id (пост найден)."""
# Мокаем _execute_query_with_result
mock_result = [("Тестовый текст", 1)] # is_anonymous = 1 (True)
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_and_anonymity_by_helper_id(helper_message_id)
# Проверяем результат
text, is_anonymous = result
assert text == "Тестовый текст"
assert is_anonymous is True
# Проверяем вызов _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, is_anonymous FROM post_from_telegram_suggest WHERE helper_text_message_id = ?" in query
assert params == (helper_message_id,)
@pytest.mark.asyncio
async def test_add_post_with_is_anonymous_true(self, post_repository):
"""Тест добавления поста с is_anonymous=True."""
post = TelegramPost(
message_id=12345,
text="Тестовый пост анон",
author_id=67890,
created_at=int(datetime.now().timestamp()),
is_anonymous=True
)
post_repository._execute_query = AsyncMock()
await post_repository.add_post(post)
call_args = post_repository._execute_query.call_args
params = call_args[0][1]
# Проверяем, что is_anonymous преобразован в 1
assert params[5] == 1
@pytest.mark.asyncio
async def test_add_post_with_is_anonymous_false(self, post_repository):
"""Тест добавления поста с is_anonymous=False."""
post = TelegramPost(
message_id=12345,
text="Тестовый пост неанон",
author_id=67890,
created_at=int(datetime.now().timestamp()),
is_anonymous=False
)
post_repository._execute_query = AsyncMock()
await post_repository.add_post(post)
call_args = post_repository._execute_query.call_args
params = call_args[0][1]
# Проверяем, что is_anonymous преобразован в 0
assert params[5] == 0
@pytest.mark.asyncio
async def test_add_post_with_is_anonymous_none(self, post_repository):
"""Тест добавления поста с is_anonymous=None."""
post = TelegramPost(
message_id=12345,
text="Тестовый пост",
author_id=67890,
created_at=int(datetime.now().timestamp()),
is_anonymous=None
)
post_repository._execute_query = AsyncMock()
await post_repository.add_post(post)
call_args = post_repository._execute_query.call_args
params = call_args[0][1]
# Проверяем, что is_anonymous остался None
assert params[5] is None
@pytest.mark.asyncio
async def test_create_tables_logs_success(self, post_repository):
"""Тест логирования успешного создания таблиц."""

287
tests/test_post_service.py Normal file
View File

@@ -0,0 +1,287 @@
"""Tests for PostService"""
import pytest
from unittest.mock import Mock, AsyncMock, MagicMock, patch
from datetime import datetime
from aiogram import types
from helper_bot.handlers.private.services import PostService, BotSettings
from database.models import TelegramPost, User
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()
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"""
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=200):
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 = "Тестовый пост анон"
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=200):
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 = "Медиагруппа подпись"
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=302):
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=True):
with patch('asyncio.sleep', return_value=None):
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 == 300
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
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=303):
with patch('helper_bot.handlers.private.services.determine_anonymity', return_value=False):
with patch('asyncio.sleep', return_value=None):
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

View File

@@ -6,6 +6,7 @@ import os
from helper_bot.utils.helper_func import (
get_first_name,
get_text_message,
determine_anonymity,
check_username_and_full_name,
safe_html_escape,
download_file,
@@ -64,12 +65,13 @@ class TestHelperFunctions:
def test_get_text_message(self, mock_message):
"""Тест функции обработки текста сообщения"""
# Тест с обычным текстом
# Тест с обычным текстом (legacy - определяется по тексту)
text = "Привет, это тестовое сообщение"
result = get_text_message(text, "Test", "testuser")
assert "Test" in result
assert "testuser" in result
assert "тестовое сообщение" in result
assert "Автор поста" in result
# Тест с пустым текстом
result = get_text_message("", "Test", "testuser")
@@ -83,6 +85,98 @@ class TestHelperFunctions:
assert "testuser" in result
assert "Обычный текст без специальных слов" in result
def test_get_text_message_with_is_anonymous_true(self, mock_message):
"""Тест функции get_text_message с is_anonymous=True"""
text = "Тестовый пост"
result = get_text_message(text, "Test", "testuser", is_anonymous=True)
assert "Пост из ТГ:" in result
assert "Тестовый пост" in result
assert "Пост опубликован анонимно" in result
assert "Автор поста" not in result
def test_get_text_message_with_is_anonymous_false(self, mock_message):
"""Тест функции get_text_message с is_anonymous=False"""
text = "Тестовый пост"
result = get_text_message(text, "Test", "testuser", is_anonymous=False)
assert "Пост из ТГ:" in result
assert "Тестовый пост" in result
assert "Автор поста" in result
assert "Test" in result
assert "testuser" in result
assert "Пост опубликован анонимно" not in result
def test_get_text_message_with_is_anonymous_none_legacy(self, mock_message):
"""Тест функции get_text_message с is_anonymous=None (legacy - определяется по тексту)"""
# Тест с "анон" в тексте
text = "Тестовый пост анон"
result = get_text_message(text, "Test", "testuser", is_anonymous=None)
assert "Пост из ТГ:" in result
assert "Тестовый пост анон" in result
assert "Пост опубликован анонимно" in result
# Тест с "неанон" в тексте
text = "Тестовый пост неанон"
result = get_text_message(text, "Test", "testuser", is_anonymous=None)
assert "Пост из ТГ:" in result
assert "Тестовый пост неанон" in result
assert "Автор поста" in result
# Тест с "не анон" в тексте
text = "Тестовый пост не анон"
result = get_text_message(text, "Test", "testuser", is_anonymous=None)
assert "Автор поста" in result
def test_get_text_message_with_username_none(self, mock_message):
"""Тест функции get_text_message без username"""
text = "Тестовый пост"
result = get_text_message(text, "Test", None, is_anonymous=False)
assert "Test" in result
assert "(Ник не указан)" in result
assert "@" not in result
def test_determine_anonymity_with_anon(self):
"""Тест функции determine_anonymity с 'анон' в тексте"""
assert determine_anonymity("Этот пост анон") is True
assert determine_anonymity("анон") is True
assert determine_anonymity("АНОН") is True # Проверка регистра
assert determine_anonymity("пост анонимный анон") is True
def test_determine_anonymity_with_neanon(self):
"""Тест функции determine_anonymity с 'неанон' в тексте"""
assert determine_anonymity("Этот пост неанон") is False
assert determine_anonymity("неанон") is False
assert determine_anonymity("НЕАНОН") is False # Проверка регистра
assert determine_anonymity("пост неанон") is False
def test_determine_anonymity_with_ne_anon(self):
"""Тест функции determine_anonymity с 'не анон' в тексте"""
assert determine_anonymity("Этот пост не анон") is False
assert determine_anonymity("не анон") is False
assert determine_anonymity("НЕ АНОН") is False # Проверка регистра
assert determine_anonymity("пост не анон") is False
def test_determine_anonymity_priority_neanon_over_anon(self):
"""Тест приоритета 'неанон' над 'анон'"""
# Если есть и "анон" и "неанон", должен вернуть False
assert determine_anonymity("анон неанон") is False
assert determine_anonymity("неанон анон") is False
assert determine_anonymity("не анон анон") is False
def test_determine_anonymity_without_keywords(self):
"""Тест функции determine_anonymity без ключевых слов"""
assert determine_anonymity("Обычный текст") is False
assert determine_anonymity("") is False
assert determine_anonymity("Пост без специальных слов") is False
def test_determine_anonymity_with_none(self):
"""Тест функции determine_anonymity с None"""
assert determine_anonymity(None) is False
def test_determine_anonymity_with_empty_string(self):
"""Тест функции determine_anonymity с пустой строкой"""
assert determine_anonymity("") is False
assert determine_anonymity(" ") is False # Только пробелы
@pytest.mark.asyncio
async def test_check_username_and_full_name(self):
"""Тест функции проверки изменений username и full_name"""