wehub-resource-sync
313 lines
9.5 KiB
Python
313 lines
9.5 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from telegram.error import TelegramError
|
|
|
|
from free_claude_code.messaging.platforms.telegram import TelegramRuntime
|
|
|
|
|
|
def _limiter_mock() -> MagicMock:
|
|
limiter = MagicMock()
|
|
limiter.start = MagicMock()
|
|
limiter.shutdown = AsyncMock()
|
|
return limiter
|
|
|
|
|
|
def _telegram_runtime(
|
|
*args, limiter=None, transcriber=None, **kwargs
|
|
) -> TelegramRuntime:
|
|
return TelegramRuntime(
|
|
*args,
|
|
limiter=limiter or _limiter_mock(),
|
|
transcriber=transcriber,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def telegram_platform():
|
|
with patch(
|
|
"free_claude_code.messaging.platforms.telegram.TELEGRAM_AVAILABLE", True
|
|
):
|
|
platform = _telegram_runtime(bot_token="test_token", allowed_user_id="12345")
|
|
return platform
|
|
|
|
|
|
def test_telegram_platform_init_no_token():
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
platform = _telegram_runtime(bot_token=None)
|
|
assert platform.bot_token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_start_success(telegram_platform):
|
|
telegram_platform.outbound.send_message = AsyncMock()
|
|
with patch("telegram.ext.Application.builder") as mock_builder:
|
|
mock_app = MagicMock()
|
|
mock_app.initialize = AsyncMock()
|
|
mock_app.start = AsyncMock()
|
|
mock_app.updater.start_polling = AsyncMock()
|
|
|
|
mock_builder.return_value.token.return_value.request.return_value.build.return_value = mock_app
|
|
|
|
await telegram_platform.start()
|
|
|
|
assert telegram_platform._connected is True
|
|
mock_app.initialize.assert_called_once()
|
|
mock_app.start.assert_called_once()
|
|
telegram_platform._limiter.start.assert_called_once_with()
|
|
telegram_platform.outbound.send_message.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_start_with_proxy():
|
|
limiter = _limiter_mock()
|
|
with patch(
|
|
"free_claude_code.messaging.platforms.telegram.TELEGRAM_AVAILABLE", True
|
|
):
|
|
platform = _telegram_runtime(
|
|
bot_token="test_token",
|
|
allowed_user_id="12345",
|
|
telegram_proxy_url="socks5://127.0.0.1:1080",
|
|
limiter=limiter,
|
|
)
|
|
|
|
with (
|
|
patch("telegram.ext.Application.builder") as mock_builder,
|
|
patch(
|
|
"free_claude_code.messaging.platforms.telegram.HTTPXRequest"
|
|
) as request_cls,
|
|
):
|
|
mock_app = MagicMock()
|
|
mock_app.initialize = AsyncMock()
|
|
mock_app.start = AsyncMock()
|
|
mock_app.updater.start_polling = AsyncMock()
|
|
|
|
builder = mock_builder.return_value
|
|
builder.token.return_value = builder
|
|
builder.request.return_value = builder
|
|
builder.get_updates_request.return_value = builder
|
|
builder.build.return_value = mock_app
|
|
request = MagicMock()
|
|
update_request = MagicMock()
|
|
request_cls.side_effect = [request, update_request]
|
|
|
|
await platform.start()
|
|
|
|
assert request_cls.call_count == 2
|
|
request_cls.assert_any_call(
|
|
connection_pool_size=8,
|
|
connect_timeout=30.0,
|
|
read_timeout=30.0,
|
|
proxy="socks5://127.0.0.1:1080",
|
|
)
|
|
builder.request.assert_called_once_with(request)
|
|
builder.get_updates_request.assert_called_once_with(update_request)
|
|
assert platform._connected is True
|
|
limiter.start.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_send_message_success(telegram_platform):
|
|
mock_bot = AsyncMock()
|
|
mock_msg = MagicMock()
|
|
mock_msg.message_id = 999
|
|
mock_bot.send_message.return_value = mock_msg
|
|
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
msg_id = await telegram_platform.outbound.send_message("chat_1", "hello")
|
|
|
|
assert msg_id == "999"
|
|
mock_bot.send_message.assert_called_once_with(
|
|
chat_id="chat_1",
|
|
text="hello",
|
|
reply_to_message_id=None,
|
|
parse_mode="MarkdownV2",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_edit_message_success(telegram_platform):
|
|
mock_bot = AsyncMock()
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.edit_message("chat_1", "999", "new text")
|
|
|
|
mock_bot.edit_message_text.assert_called_once_with(
|
|
chat_id="chat_1", message_id=999, text="new text", parse_mode="MarkdownV2"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_delete_messages_uses_batch_api(telegram_platform):
|
|
mock_bot = AsyncMock()
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.delete_messages("chat_1", ["1", "2", "bad"])
|
|
|
|
mock_bot.delete_messages.assert_awaited_once_with(
|
|
chat_id="chat_1",
|
|
message_ids=[1, 2],
|
|
)
|
|
mock_bot.delete_message.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_delete_messages_chunks_batch_api(telegram_platform):
|
|
mock_bot = AsyncMock()
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.delete_messages(
|
|
"chat_1",
|
|
[str(i) for i in range(105)],
|
|
)
|
|
|
|
assert mock_bot.delete_messages.await_count == 2
|
|
assert mock_bot.delete_messages.await_args_list[0].kwargs["message_ids"] == list(
|
|
range(100)
|
|
)
|
|
assert mock_bot.delete_messages.await_args_list[1].kwargs["message_ids"] == list(
|
|
range(100, 105)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_delete_messages_falls_back_without_batch(
|
|
telegram_platform,
|
|
):
|
|
class BotWithoutBatch:
|
|
def __init__(self) -> None:
|
|
self.delete_message = AsyncMock()
|
|
|
|
bot = BotWithoutBatch()
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = bot
|
|
|
|
await telegram_platform.outbound.delete_messages("chat_1", ["1", "2"])
|
|
|
|
assert bot.delete_message.await_args_list[0].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 1,
|
|
}
|
|
assert bot.delete_message.await_args_list[1].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 2,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_delete_messages_falls_back_after_batch_failure(
|
|
telegram_platform,
|
|
):
|
|
mock_bot = AsyncMock()
|
|
mock_bot.delete_messages.side_effect = RuntimeError("bulk failed")
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.delete_messages("chat_1", ["1", "2"])
|
|
|
|
mock_bot.delete_messages.assert_awaited_once_with(
|
|
chat_id="chat_1",
|
|
message_ids=[1, 2],
|
|
)
|
|
assert mock_bot.delete_message.await_args_list[0].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 1,
|
|
}
|
|
assert mock_bot.delete_message.await_args_list[1].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 2,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_delete_messages_falls_back_after_swallowed_error(
|
|
telegram_platform,
|
|
):
|
|
mock_bot = AsyncMock()
|
|
mock_bot.delete_messages.side_effect = TelegramError("message can't be deleted")
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.delete_messages("chat_1", ["1", "2"])
|
|
|
|
mock_bot.delete_messages.assert_awaited_once_with(
|
|
chat_id="chat_1",
|
|
message_ids=[1, 2],
|
|
)
|
|
assert mock_bot.delete_message.await_args_list[0].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 1,
|
|
}
|
|
assert mock_bot.delete_message.await_args_list[1].kwargs == {
|
|
"chat_id": "chat_1",
|
|
"message_id": 2,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_single_delete_still_swallows_known_error(
|
|
telegram_platform,
|
|
):
|
|
mock_bot = AsyncMock()
|
|
mock_bot.delete_message.side_effect = TelegramError("message can't be deleted")
|
|
telegram_platform._application = MagicMock()
|
|
telegram_platform._application.bot = mock_bot
|
|
|
|
await telegram_platform.outbound.delete_message("chat_1", "1")
|
|
|
|
mock_bot.delete_message.assert_awaited_once_with(
|
|
chat_id="chat_1",
|
|
message_id=1,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_platform_queue_send_message(telegram_platform):
|
|
mock_limiter = telegram_platform._limiter
|
|
mock_limiter.enqueue = AsyncMock()
|
|
|
|
await telegram_platform.outbound.queue_send_message(
|
|
"chat_1", "hello", fire_and_forget=False
|
|
)
|
|
|
|
mock_limiter.enqueue.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_telegram_message_authorized(telegram_platform):
|
|
handler = AsyncMock()
|
|
telegram_platform.on_message(handler)
|
|
|
|
mock_update = MagicMock()
|
|
mock_update.message.text = "hello"
|
|
mock_update.message.message_id = 1
|
|
mock_update.effective_user.id = 12345
|
|
mock_update.effective_chat.id = 6789
|
|
mock_update.message.reply_to_message = None
|
|
|
|
await telegram_platform._on_telegram_message(mock_update, MagicMock())
|
|
|
|
handler.assert_called_once()
|
|
incoming = handler.call_args[0][0]
|
|
assert incoming.text == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_telegram_message_unauthorized(telegram_platform):
|
|
handler = AsyncMock()
|
|
telegram_platform.on_message(handler)
|
|
|
|
mock_update = MagicMock()
|
|
mock_update.message.text = "hello"
|
|
mock_update.effective_user.id = 99999 # Unauthorized
|
|
|
|
await telegram_platform._on_telegram_message(mock_update, MagicMock())
|
|
|
|
handler.assert_not_called()
|