mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-04-28 03:20:01 +00:00
311 lines
11 KiB
Python
311 lines
11 KiB
Python
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
|