File size: 4,502 Bytes
6172a47 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from telegram.error import NetworkError, RetryAfter, TelegramError
from messaging.platforms.telegram import TelegramPlatform
@pytest.fixture
def telegram_platform():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
platform = TelegramPlatform(bot_token="test_token", allowed_user_id="12345")
return platform
@pytest.mark.asyncio
async def test_telegram_retry_on_network_error(telegram_platform):
mock_bot = AsyncMock()
mock_msg = MagicMock()
mock_msg.message_id = 999
# Fail twice, then succeed
mock_bot.send_message.side_effect = [
NetworkError("Connection failed"),
NetworkError("Connection failed"),
mock_msg,
]
telegram_platform._application = MagicMock()
telegram_platform._application.bot = mock_bot
# We need to patch asyncio.sleep to speed up the test
with patch("asyncio.sleep", AsyncMock()) as mock_sleep:
msg_id = await telegram_platform.send_message("chat_1", "hello")
assert msg_id == "999"
assert mock_bot.send_message.call_count == 3
assert mock_sleep.call_count == 2
@pytest.mark.asyncio
async def test_telegram_retry_on_retry_after(telegram_platform):
mock_bot = AsyncMock()
mock_msg = MagicMock()
mock_msg.message_id = 1000
# Fail with RetryAfter, then succeed
mock_bot.send_message.side_effect = [RetryAfter(retry_after=5), mock_msg]
telegram_platform._application = MagicMock()
telegram_platform._application.bot = mock_bot
with patch("asyncio.sleep", AsyncMock()) as mock_sleep:
msg_id = await telegram_platform.send_message("chat_1", "hello")
assert msg_id == "1000"
assert mock_bot.send_message.call_count == 2
mock_sleep.assert_called_with(5)
@pytest.mark.asyncio
async def test_telegram_no_retry_on_bad_request(telegram_platform):
mock_bot = AsyncMock()
# Fail with generic TelegramError (should not retry unless specific conditions met)
mock_bot.send_message.side_effect = TelegramError("Bad Request: some error")
telegram_platform._application = MagicMock()
telegram_platform._application.bot = mock_bot
with pytest.raises(TelegramError):
await telegram_platform.send_message("chat_1", "hello")
assert mock_bot.send_message.call_count == 1
def test_handler_build_message_hardening():
# Formatting hardening now lives in TranscriptBuffer rendering.
from messaging.rendering.telegram_markdown import (
escape_md_v2,
escape_md_v2_code,
mdv2_bold,
mdv2_code_inline,
render_markdown_to_mdv2,
)
from messaging.transcript import RenderCtx, TranscriptBuffer
ctx = RenderCtx(
bold=mdv2_bold,
code_inline=mdv2_code_inline,
escape_code=escape_md_v2_code,
escape_text=escape_md_v2,
render_markdown=render_markdown_to_mdv2,
)
# Case 1: Empty transcript + no status => empty string.
t = TranscriptBuffer()
msg = t.render(ctx, limit_chars=3900, status=None)
assert msg == ""
# Case 2: Truncation with code block closing and status preserved.
t.apply({"type": "thinking_chunk", "text": ("thought " * 200)})
t.apply({"type": "text_chunk", "text": ("This is a very long message. " * 300)})
msg = t.render(ctx, limit_chars=3900, status="Finishing...")
assert len(msg) <= 4096
assert "Finishing..." in msg
if "```" in msg:
assert msg.count("```") % 2 == 0
def test_render_output_never_exceeds_4096():
"""Transcript render with various status lengths never exceeds Telegram 4096 limit."""
from messaging.rendering.telegram_markdown import (
escape_md_v2,
escape_md_v2_code,
mdv2_bold,
mdv2_code_inline,
render_markdown_to_mdv2,
)
from messaging.transcript import RenderCtx, TranscriptBuffer
ctx = RenderCtx(
bold=mdv2_bold,
code_inline=mdv2_code_inline,
escape_code=escape_md_v2_code,
escape_text=escape_md_v2,
render_markdown=render_markdown_to_mdv2,
)
t = TranscriptBuffer()
t.apply({"type": "thinking_chunk", "text": "x" * 500})
t.apply({"type": "text_chunk", "text": "y" * 3500})
for status in [None, "Done", "✅ *Complete*", "A" * 100]:
msg = t.render(ctx, limit_chars=3900, status=status)
assert len(msg) <= 4096, f"status={status!r} produced len={len(msg)}"
|