kandrusyak_bot/tests/test_telegram_utils.py

import asyncio
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock

from telegram.constants import ChatAction, ParseMode
from telegram.error import BadRequest

from assistant_bot.telegram_utils import (
    FORMATTING_FALLBACK_NOTICE,
    build_inline_results,
    escape_markdown_text,
    render_markdown_chunks,
    reply_markdown,
    send_markdown,
    typing_action,
)


class MarkdownRenderingTests(unittest.TestCase):
    def test_converts_common_markdown_to_markdown_v2(self) -> None:
        chunks = render_markdown_chunks(
            "**📝 Заметки**\n\n- `#3` Проект\\_v2\\.0"
        )

        self.assertEqual(len(chunks), 1)
        markdown_v2, plain_text = chunks[0]
        self.assertIn("*📝 Заметки*", markdown_v2)
        self.assertIn("Проект\\_v2\\.0", markdown_v2)
        self.assertNotIn("**", markdown_v2)
        self.assertIn("Проект_v2.0", plain_text)

    def test_splits_long_formatted_text_without_losing_format(self) -> None:
        chunks = render_markdown_chunks(f"**{'слово ' * 999}слово**")

        self.assertGreater(len(chunks), 1)
        for markdown_v2, plain_text in chunks:
            self.assertLessEqual(len(plain_text.encode("utf-16-le")) // 2, 3900)
            self.assertLessEqual(len(markdown_v2.encode("utf-16-le")) // 2, 3900)
            self.assertTrue(markdown_v2.startswith("*"))
            self.assertTrue(markdown_v2.rstrip().endswith("*"))

    def test_splits_by_rendered_markdown_v2_length(self) -> None:
        chunks = render_markdown_chunks(escape_markdown_text("_" * 5000))

        self.assertGreater(len(chunks), 2)
        for markdown_v2, _plain_text in chunks:
            self.assertLessEqual(len(markdown_v2.encode("utf-16-le")) // 2, 3900)

    def test_inline_results_use_markdown_v2(self) -> None:
        results = build_inline_results("проект_v2")

        bold_content = results[1].input_message_content
        self.assertEqual(bold_content.parse_mode, ParseMode.MARKDOWN_V2)
        self.assertEqual(bold_content.message_text, "*проект\\_v2*")


class MarkdownDeliveryTests(unittest.IsolatedAsyncioTestCase):
    async def test_send_markdown_uses_the_shared_formatted_delivery(self) -> None:
        sent_message = SimpleNamespace()
        bot = SimpleNamespace(
            send_message=AsyncMock(return_value=sent_message)
        )

        result = await send_markdown(bot, 42, "**Готово.**")

        self.assertIs(result, sent_message)
        bot.send_message.assert_awaited_once_with(
            chat_id=42,
            text="*Готово\\.*",
            parse_mode=ParseMode.MARKDOWN_V2,
        )

    async def test_retries_plain_text_with_notice_when_telegram_rejects_markup(
        self,
    ) -> None:
        fallback_message = SimpleNamespace()
        message = SimpleNamespace(
            reply_text=AsyncMock(
                side_effect=[
                    BadRequest("Can't parse entities"),
                    fallback_message,
                ]
            )
        )

        sent = await reply_markdown(message, "**Готово.**")

        self.assertIs(sent, fallback_message)
        self.assertEqual(message.reply_text.await_count, 2)
        formatted_call, fallback_call = message.reply_text.await_args_list
        self.assertEqual(
            formatted_call.kwargs["parse_mode"],
            ParseMode.MARKDOWN_V2,
        )
        self.assertNotIn("parse_mode", fallback_call.kwargs)
        self.assertEqual(
            fallback_call.args[0],
            f"Готово.\n\n{FORMATTING_FALLBACK_NOTICE}",
        )

    async def test_places_fallback_notice_only_in_last_chunk(self) -> None:
        first_message = SimpleNamespace()
        markdown = f"**{'слово ' * 999}слово**"
        expected_chunks = render_markdown_chunks(markdown)
        call_number = 0

        async def send(*_args, **_kwargs):
            nonlocal call_number
            call_number += 1
            if call_number == 1:
                raise BadRequest("Can't parse entities")
            return first_message if call_number == 2 else SimpleNamespace()

        message = SimpleNamespace(reply_text=AsyncMock(side_effect=send))

        await reply_markdown(message, markdown)

        self.assertEqual(message.reply_text.await_count, len(expected_chunks) + 1)
        first_plain_call = message.reply_text.await_args_list[1]
        last_formatted_call = message.reply_text.await_args_list[-1]
        self.assertNotIn(FORMATTING_FALLBACK_NOTICE, first_plain_call.args[0])
        self.assertIn(
            "Не удалось применить форматирование",
            last_formatted_call.args[0],
        )
        self.assertEqual(
            last_formatted_call.kwargs["parse_mode"],
            ParseMode.MARKDOWN_V2,
        )


class TypingActionTests(unittest.IsolatedAsyncioTestCase):
    async def test_refreshes_typing_until_context_exits(self) -> None:
        refreshed = asyncio.Event()
        call_count = 0

        async def send_chat_action(**_kwargs) -> None:
            nonlocal call_count
            call_count += 1
            if call_count >= 2:
                refreshed.set()

        bot = SimpleNamespace(
            send_chat_action=AsyncMock(side_effect=send_chat_action)
        )

        async with typing_action(bot, chat_id=42, interval=0.001):
            await asyncio.wait_for(refreshed.wait(), timeout=0.5)

        calls_after_exit = call_count
        await asyncio.sleep(0.01)

        self.assertGreaterEqual(calls_after_exit, 2)
        self.assertEqual(call_count, calls_after_exit)
        for call in bot.send_chat_action.await_args_list:
            self.assertEqual(call.kwargs["chat_id"], 42)
            self.assertEqual(call.kwargs["action"], ChatAction.TYPING)


if __name__ == "__main__":
    unittest.main()