tests/messaging/test_telegram_edge_cases.py

import asyncio
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from telegram.error import NetworkError, RetryAfter, TelegramError


def test_telegram_platform_init_raises_when_dependency_missing():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", False):
        from messaging.platforms.telegram import TelegramPlatform

        with pytest.raises(ImportError):
            TelegramPlatform(bot_token="x")


@pytest.mark.asyncio
async def test_telegram_platform_start_requires_token():
    with (
        patch.dict("os.environ", {}, clear=True),
        patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True),
    ):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token=None)
        with pytest.raises(ValueError):
            await platform.start()


@pytest.mark.asyncio
async def test_telegram_platform_stop_no_application_is_noop():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._application = None
        platform._connected = True
        await platform.stop()
        assert platform.is_connected is False


@pytest.mark.asyncio
async def test_with_retry_returns_none_when_message_not_modified_network_error():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        async def _f():
            raise NetworkError("Message is not modified")

        assert await platform._with_retry(_f) is None


@pytest.mark.asyncio
async def test_with_retry_retries_network_error_then_succeeds(monkeypatch):
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        monkeypatch.setattr(asyncio, "sleep", AsyncMock())

        calls = {"n": 0}

        async def _f():
            calls["n"] += 1
            if calls["n"] == 1:
                raise NetworkError("temporary")
            return "ok"

        assert await platform._with_retry(_f) == "ok"
        assert calls["n"] == 2


@pytest.mark.asyncio
async def test_with_retry_honors_retry_after_timedelta(monkeypatch):
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        monkeypatch.setattr(asyncio, "sleep", AsyncMock())

        calls = {"n": 0}

        async def _f():
            calls["n"] += 1
            if calls["n"] == 1:
                raise RetryAfter(retry_after=timedelta(seconds=0.01))
            return "ok"

        assert await platform._with_retry(_f) == "ok"
        assert calls["n"] == 2


@pytest.mark.asyncio
async def test_with_retry_drops_parse_mode_on_markdown_entity_error():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        calls = []

        async def _f(parse_mode=None):
            calls.append(parse_mode)
            if len(calls) == 1:
                raise TelegramError("Can't parse entities: bad markdown")
            return "ok"

        assert await platform._with_retry(_f, parse_mode="MarkdownV2") == "ok"
        assert calls == ["MarkdownV2", None]


@pytest.mark.asyncio
async def test_queue_send_message_without_limiter_calls_send_message():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._limiter = None
        with patch.object(
            platform, "send_message", new_callable=AsyncMock
        ) as mock_send:
            mock_send.return_value = "1"
            assert await platform.queue_send_message("c", "t") == "1"
            mock_send.assert_awaited_once()


@pytest.mark.asyncio
async def test_queue_edit_message_without_limiter_calls_edit_message():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._limiter = None
        with patch.object(
            platform, "edit_message", new_callable=AsyncMock
        ) as mock_edit:
            await platform.queue_edit_message("c", "1", "t")
            mock_edit.assert_awaited_once()


def test_fire_and_forget_non_coroutine_uses_ensure_future(monkeypatch):
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        ef = MagicMock()
        monkeypatch.setattr(asyncio, "ensure_future", ef)

        platform.fire_and_forget(MagicMock())
        ef.assert_called_once()


@pytest.mark.asyncio
async def test_on_start_command_replies_and_forwards():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        with patch.object(
            platform, "_on_telegram_message", new_callable=AsyncMock
        ) as mock_msg:
            update = MagicMock()
            update.message.reply_text = AsyncMock()

            await platform._on_start_command(update, MagicMock())
            update.message.reply_text.assert_awaited_once()
            mock_msg.assert_awaited_once()


@pytest.mark.asyncio
async def test_on_telegram_message_handler_error_sends_error_message():
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
        with patch.object(
            platform, "send_message", new_callable=AsyncMock
        ) as mock_send:

            async def _boom(_incoming):
                raise RuntimeError("bad")

            platform.on_message(_boom)

            update = MagicMock()
            update.message.text = "hello"
            update.message.message_id = 7
            update.message.reply_to_message = None
            update.effective_user.id = 123
            update.effective_chat.id = 456

            await platform._on_telegram_message(update, MagicMock())
            mock_send.assert_awaited_once()


@pytest.mark.asyncio
async def test_telegram_start_retries_on_network_error(monkeypatch):
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="token", allowed_user_id=None)

        monkeypatch.setattr(asyncio, "sleep", AsyncMock())

        with (
            patch("telegram.ext.Application.builder") as mock_builder,
            patch("messaging.limiter.MessagingRateLimiter.get_instance", AsyncMock()),
        ):
            mock_app = MagicMock()
            mock_app.initialize = AsyncMock(side_effect=[NetworkError("no"), None])
            mock_app.start = AsyncMock()
            mock_app.updater = None

            mock_builder.return_value.token.return_value.request.return_value.build.return_value = mock_app

            await platform.start()
            assert platform.is_connected is True


@pytest.mark.asyncio
async def test_edit_message_with_text_exceeding_4096_raises():
    """edit_message with text > 4096 raises TelegramError (BadRequest)."""
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._application = MagicMock()
        platform._application.bot = AsyncMock()
        platform._application.bot.edit_message_text = AsyncMock(
            side_effect=TelegramError("Bad Request: message is too long")
        )

        with pytest.raises(TelegramError):
            await platform.edit_message("c", "1", "x" * 5000)


@pytest.mark.asyncio
async def test_edit_message_empty_string():
    """edit_message with empty string - Telegram accepts (no-op edit)."""
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._application = MagicMock()
        platform._application.bot = AsyncMock()
        platform._application.bot.edit_message_text = AsyncMock()

        await platform.edit_message("c", "1", "")
        platform._application.bot.edit_message_text.assert_awaited_once_with(
            chat_id="c", message_id=1, text="", parse_mode="MarkdownV2"
        )


@pytest.mark.asyncio
async def test_send_message_empty_string():
    """send_message with empty string - Telegram may reject; we pass through."""
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")
        platform._application = MagicMock()
        mock_msg = MagicMock()
        mock_msg.message_id = 1
        platform._application.bot = AsyncMock()
        platform._application.bot.send_message = AsyncMock(return_value=mock_msg)

        msg_id = await platform.send_message("c", "")
        assert msg_id == "1"
        platform._application.bot.send_message.assert_awaited_once()


@pytest.mark.asyncio
async def test_on_telegram_message_non_text_update_ignored():
    """Update with message.photo but no text returns early without calling handler."""
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
        handler = AsyncMock()
        platform.on_message(handler)

        update = MagicMock()
        update.message.text = None
        update.message.photo = [MagicMock()]
        update.message.message_id = 7
        update.message.reply_to_message = None
        update.effective_user.id = 123
        update.effective_chat.id = 456

        await platform._on_telegram_message(update, MagicMock())
        handler.assert_not_called()


@pytest.mark.asyncio
async def test_with_retry_message_not_found_returns_none():
    """'message to edit not found' returns None without retry."""
    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
        from messaging.platforms.telegram import TelegramPlatform

        platform = TelegramPlatform(bot_token="t")

        async def _f():
            raise TelegramError("message to edit not found")

        result = await platform._with_retry(_f)
        assert result is None