Jack1808/Claude_Code

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