166 lines
8.4 KiB
Python
166 lines
8.4 KiB
Python
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from aiogram import Bot
|
|
from aiogram.client.default import DefaultBotProperties
|
|
from aiogram.exceptions import TelegramBadRequest
|
|
from aiogram.methods import CopyMessage
|
|
from aiogram.types import Chat, Message, MessageEntity, Update, User as TelegramUser
|
|
from sqlalchemy import select
|
|
|
|
from src.core.broadcast_services import BroadcastService
|
|
from src.core.database import async_session_maker
|
|
from src.core.emoji_mapping_service import EmojiMappingService
|
|
from src.core.models import BlockedUser, EmojiMapping, P2PMessage
|
|
from src.core.services import UserService
|
|
from src.handlers.p2p_chat import handle_p2p_message
|
|
from src.utils.telegram_messages import copy_preserving_entities
|
|
from test_dispatcher import TelegramStub, dispatch, event
|
|
|
|
|
|
BODY = "😀 🎲 🎲 <b>"
|
|
FIRST_ID = "5368324170671202286"
|
|
SECOND_ID = "5368324170671202287"
|
|
|
|
|
|
def emoji_entities():
|
|
return [MessageEntity(type="bold", offset=3, length=5),
|
|
MessageEntity(type="custom_emoji", offset=3, length=2, custom_emoji_id=FIRST_ID),
|
|
MessageEntity(type="custom_emoji", offset=6, length=2, custom_emoji_id=SECOND_ID),
|
|
MessageEntity(type="text_link", offset=9, length=3, url="https://example.com")]
|
|
|
|
|
|
def source_message(bot, **values):
|
|
return Message(message_id=7, date=datetime.now(timezone.utc), chat=Chat(id=101, type="private"),
|
|
from_user=TelegramUser(id=101, is_bot=False, first_name="Sender"), **values).as_(bot)
|
|
|
|
|
|
def make_bot():
|
|
stub = TelegramStub()
|
|
return Bot("123456:TEST_TOKEN_FOR_ISOLATED_TESTS", session=stub,
|
|
default=DefaultBotProperties(parse_mode="Markdown")), stub
|
|
|
|
|
|
def assert_custom_emoji(text, entities):
|
|
custom = [entity for entity in entities if entity.type == "custom_emoji"]
|
|
assert [entity.custom_emoji_id for entity in custom] == [FIRST_ID, SECOND_ID]
|
|
assert [entity.extract_from(text) for entity in custom] == ["🎲", "🎲"]
|
|
link = next(entity for entity in entities if entity.type == "text_link")
|
|
assert link.extract_from(text) == "<b>"
|
|
|
|
|
|
async def test_custom_emoji_and_nested_formatting_survive_utf16_header():
|
|
bot, stub = make_bot()
|
|
message = source_message(bot, text=BODY, entities=emoji_entities())
|
|
await copy_preserving_entities(message, 102, "Кассир 🧑💻 <name>")
|
|
sent = stub.calls[0]
|
|
assert sent.parse_mode is None
|
|
assert sent.text.endswith(BODY)
|
|
assert_custom_emoji(sent.text, sent.entities)
|
|
assert [entity.offset for entity in message.entities] == [3, 3, 6, 9]
|
|
|
|
|
|
@pytest.mark.parametrize("media", ["photo", "video", "document", "animation", "audio", "voice"])
|
|
async def test_custom_emoji_survive_media_caption(media):
|
|
bot, stub = make_bot()
|
|
value = dict(file_id="file", file_unique_id="unique", width=1, height=1, duration=1)
|
|
message = source_message(bot, caption=BODY, caption_entities=emoji_entities(),
|
|
**{media: [value] if media == "photo" else value})
|
|
await copy_preserving_entities(message, 102, "Отправитель 😀")
|
|
sent = stub.calls[0]
|
|
assert sent.__api_method__ == "copyMessage"
|
|
assert sent.parse_mode is None
|
|
assert_custom_emoji(sent.caption, sent.caption_entities)
|
|
|
|
|
|
@pytest.mark.parametrize("caption", [False, True])
|
|
async def test_max_length_message_is_copied_without_trimming_emoji(caption):
|
|
bot, stub = make_bot()
|
|
limit = 1024 if caption else 4096
|
|
body = "x" * (limit - 2) + "🎲"
|
|
entities = [MessageEntity(type="custom_emoji", offset=limit - 2, length=2, custom_emoji_id=FIRST_ID)]
|
|
values = dict(caption=body, caption_entities=entities, document={"file_id": "file", "file_unique_id": "unique"}) if caption else dict(text=body, entities=entities)
|
|
message = source_message(bot, **values)
|
|
await copy_preserving_entities(message, 102, "Sender")
|
|
assert [method.__api_method__ for method in stub.calls] == ["sendMessage", "copyMessage"]
|
|
assert stub.calls[1].from_chat_id == 101 and stub.calls[1].message_id == 7
|
|
assert stub.calls[1].caption is None # Telegram copies the original caption and all its entities.
|
|
|
|
|
|
async def test_p2p_delivers_custom_emoji_and_stores_literal_text():
|
|
bot, stub = make_bot()
|
|
async with async_session_maker() as session:
|
|
sender = await UserService.get_or_create_user(session, 101, first_name="Sender")
|
|
recipient = await UserService.get_or_create_user(session, 102, first_name="Recipient")
|
|
recipient_id = recipient.id
|
|
state = SimpleNamespace(get_data=AsyncMock(return_value={"recipient_id": recipient_id, "recipient_telegram_id": 102}))
|
|
message = source_message(bot, text=BODY, entities=emoji_entities())
|
|
await handle_p2p_message(message, state)
|
|
delivery = next(call for call in stub.calls if getattr(call, "chat_id", None) == 102)
|
|
assert_custom_emoji(delivery.text, delivery.entities)
|
|
async with async_session_maker() as session:
|
|
saved = (await session.scalars(select(P2PMessage))).one()
|
|
assert saved.text == BODY
|
|
assert saved.recipient_message_id == 100
|
|
|
|
|
|
async def test_direct_and_channel_broadcast_copy_original_custom_emoji_message():
|
|
bot, stub = make_bot()
|
|
message = source_message(bot, text=BODY, entities=emoji_entities())
|
|
async with async_session_maker() as session:
|
|
admin = await UserService.get_or_create_user(session, 101)
|
|
recipient = await UserService.get_or_create_user(session, 102)
|
|
admin_id = admin.id
|
|
service = BroadcastService()
|
|
assert (await service.send_message_to_user(bot, recipient, message))[0]
|
|
assert await service.broadcast_to_channel(bot, message, -100123, admin_id)
|
|
assert [call.chat_id for call in stub.calls] == [102, -100123]
|
|
assert all(call.__api_method__ == "copyMessage" and call.from_chat_id == 101
|
|
and call.message_id == 7 and call.caption is None for call in stub.calls)
|
|
|
|
|
|
async def test_rejected_custom_emoji_does_not_block_future_broadcasts():
|
|
message = SimpleNamespace(copy_to=AsyncMock(side_effect=[
|
|
TelegramBadRequest(method=CopyMessage(chat_id=102, from_chat_id=101, message_id=7),
|
|
message="CUSTOM_EMOJI_INVALID"), SimpleNamespace(message_id=100)]))
|
|
service = BroadcastService()
|
|
user = SimpleNamespace(telegram_id=102)
|
|
assert await service.send_message_to_user(None, user, message) == (False, "bad_request")
|
|
assert await service.send_message_to_user(None, user, message) == (True, None)
|
|
async with async_session_maker() as session:
|
|
assert (await session.scalars(select(BlockedUser))).all() == []
|
|
|
|
|
|
async def test_real_admin_dialog_rejects_plain_emoji_and_registers_actual_id():
|
|
await dispatch(event(900001, text="/cancel"))
|
|
await dispatch(event(900001, text="/add_emoji"))
|
|
calls = await dispatch(event(900001, text="🎲"))
|
|
assert any("Обычный символ" in (getattr(call, "text", "") or "") for call in calls)
|
|
update = event(900001, text="🎲")
|
|
update = update.model_copy(update={"message": update.message.model_copy(update={
|
|
"entities": [MessageEntity(type="custom_emoji", offset=0, length=2, custom_emoji_id=FIRST_ID)]})})
|
|
await dispatch(update)
|
|
await dispatch(event(900001, text="Розыгрыш"))
|
|
async with async_session_maker() as session:
|
|
mapping = (await session.scalars(select(EmojiMapping))).one()
|
|
admin = await UserService.get_user_by_telegram_id(session, 900001)
|
|
assert mapping.admin_id == admin.id
|
|
assert mapping.emoji_id == FIRST_ID
|
|
assert (await EmojiMappingService(session).get_all_emoji_by_admin(900001))[0].id == mapping.id
|
|
|
|
|
|
async def test_catalog_renders_valid_safe_html_and_rejects_fake_id():
|
|
async with async_session_maker() as session:
|
|
service = EmojiMappingService(session)
|
|
with pytest.raises(ValueError, match="custom_emoji_id"):
|
|
await service.register_emoji("🎲", "🎲", 900001)
|
|
await service.register_emoji("🎲", FIRST_ID, 900001)
|
|
rendered = await service.replace_emojis_in_text("<b> & 🎲 🎲")
|
|
assert rendered == f'<b> & <tg-emoji emoji-id="{FIRST_ID}">🎲</tg-emoji> <tg-emoji emoji-id="{FIRST_ID}">🎲</tg-emoji>'
|
|
assert await service.restore_emojis_in_text(rendered) == "<b> & 🎲 🎲"
|
|
with pytest.raises(ValueError, match="уже зарегистрирован"):
|
|
await service.register_emoji("🎲", FIRST_ID, 900002)
|
|
assert len(await service.get_all_emojis()) == 1
|