Jack1808/Claude_Code

1from unittest.mock import AsyncMock, MagicMock, patch2 3import pytest4 5from messaging.platforms.telegram import TelegramPlatform6 7 8@pytest.fixture9def telegram_platform():10    with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):11        platform = TelegramPlatform(bot_token="test_token", allowed_user_id="12345")12        return platform13 14 15def test_telegram_platform_init_no_token():16    with patch.dict("os.environ", {}, clear=True):17        platform = TelegramPlatform(bot_token=None)18        assert platform.bot_token is None19 20 21@pytest.mark.asyncio22async def test_telegram_platform_start_success(telegram_platform):23    with patch("telegram.ext.Application.builder") as mock_builder:24        mock_app = MagicMock()25        mock_app.initialize = AsyncMock()26        mock_app.start = AsyncMock()27        mock_app.updater.start_polling = AsyncMock()28 29        mock_builder.return_value.token.return_value.request.return_value.build.return_value = mock_app30 31        # Mock MessagingRateLimiter32        with patch("messaging.limiter.MessagingRateLimiter.get_instance", AsyncMock()):33            await telegram_platform.start()34 35            assert telegram_platform._connected is True36            mock_app.initialize.assert_called_once()37            mock_app.start.assert_called_once()38 39 40@pytest.mark.asyncio41async def test_telegram_platform_send_message_success(telegram_platform):42    mock_bot = AsyncMock()43    mock_msg = MagicMock()44    mock_msg.message_id = 99945    mock_bot.send_message.return_value = mock_msg46 47    telegram_platform._application = MagicMock()48    telegram_platform._application.bot = mock_bot49 50    msg_id = await telegram_platform.send_message("chat_1", "hello")51 52    assert msg_id == "999"53    mock_bot.send_message.assert_called_once_with(54        chat_id="chat_1",55        text="hello",56        reply_to_message_id=None,57        parse_mode="MarkdownV2",58    )59 60 61@pytest.mark.asyncio62async def test_telegram_platform_edit_message_success(telegram_platform):63    mock_bot = AsyncMock()64    telegram_platform._application = MagicMock()65    telegram_platform._application.bot = mock_bot66 67    await telegram_platform.edit_message("chat_1", "999", "new text")68 69    mock_bot.edit_message_text.assert_called_once_with(70        chat_id="chat_1", message_id=999, text="new text", parse_mode="MarkdownV2"71    )72 73 74@pytest.mark.asyncio75async def test_telegram_platform_queue_send_message(telegram_platform):76    mock_limiter = AsyncMock()77    telegram_platform._limiter = mock_limiter78 79    await telegram_platform.queue_send_message("chat_1", "hello", fire_and_forget=False)80 81    mock_limiter.enqueue.assert_called_once()82 83 84@pytest.mark.asyncio85async def test_on_telegram_message_authorized(telegram_platform):86    handler = AsyncMock()87    telegram_platform.on_message(handler)88 89    mock_update = MagicMock()90    mock_update.message.text = "hello"91    mock_update.message.message_id = 192    mock_update.effective_user.id = 1234593    mock_update.effective_chat.id = 678994    mock_update.message.reply_to_message = None95 96    await telegram_platform._on_telegram_message(mock_update, MagicMock())97 98    handler.assert_called_once()99    incoming = handler.call_args[0][0]100    assert incoming.text == "hello"101    assert incoming.user_id == "12345"102 103 104@pytest.mark.asyncio105async def test_on_telegram_message_unauthorized(telegram_platform):106    handler = AsyncMock()107    telegram_platform.on_message(handler)108 109    mock_update = MagicMock()110    mock_update.message.text = "hello"111    mock_update.effective_user.id = 99999  # Unauthorized112 113    await telegram_platform._on_telegram_message(mock_update, MagicMock())114 115    handler.assert_not_called()116