code-server
97ad118615
Processes local files and remote URLs Optimizes images automatically Groups 2+ images/videos into albums Handles caption overflow (>1024 chars) Routes to correct Telegram API methods Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
"""Integration tests for Telegram media sending."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from nanobot.bus.events import OutboundMessage
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.channels.telegram import TelegramChannel
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_telegram_app():
|
|
"""Mock python-telegram-bot Application."""
|
|
app = MagicMock()
|
|
app.bot = MagicMock()
|
|
app.bot.send_photo = AsyncMock()
|
|
app.bot.send_video = AsyncMock()
|
|
app.bot.send_audio = AsyncMock()
|
|
app.bot.send_document = AsyncMock()
|
|
app.bot.send_media_group = AsyncMock()
|
|
app.bot.send_message = AsyncMock()
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_single_image(mock_telegram_app, tmp_path):
|
|
"""Test sending single image."""
|
|
# Create test image
|
|
from PIL import Image
|
|
img_path = tmp_path / "test.jpg"
|
|
img = Image.new("RGB", (100, 100), color="red")
|
|
img.save(img_path, format="JPEG")
|
|
|
|
# Setup channel
|
|
bus = MessageBus()
|
|
config = MagicMock()
|
|
config.token = "fake_token"
|
|
config.proxy = None
|
|
|
|
channel = TelegramChannel(config, bus)
|
|
channel._app = mock_telegram_app
|
|
channel._running = True
|
|
|
|
# Send message with media
|
|
msg = OutboundMessage(
|
|
channel="telegram",
|
|
chat_id="12345",
|
|
content="Test image",
|
|
media=[str(img_path)]
|
|
)
|
|
|
|
with patch("nanobot.channels.telegram._markdown_to_telegram_html", return_value="Test image"):
|
|
await channel.send(msg)
|
|
|
|
# Verify send_photo was called
|
|
mock_telegram_app.bot.send_photo.assert_called_once()
|
|
call_args = mock_telegram_app.bot.send_photo.call_args
|
|
assert call_args.kwargs["chat_id"] == 12345
|
|
assert call_args.kwargs["caption"] == "Test image"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_album(mock_telegram_app, tmp_path):
|
|
"""Test sending multiple images as album."""
|
|
from PIL import Image
|
|
|
|
# Create test images
|
|
img_paths = []
|
|
for i in range(3):
|
|
img_path = tmp_path / f"test{i}.jpg"
|
|
img = Image.new("RGB", (100, 100), color="red")
|
|
img.save(img_path, format="JPEG")
|
|
img_paths.append(str(img_path))
|
|
|
|
# Setup channel
|
|
bus = MessageBus()
|
|
config = MagicMock()
|
|
config.token = "fake_token"
|
|
config.proxy = None
|
|
|
|
channel = TelegramChannel(config, bus)
|
|
channel._app = mock_telegram_app
|
|
channel._running = True
|
|
|
|
# Send message with multiple images
|
|
msg = OutboundMessage(
|
|
channel="telegram",
|
|
chat_id="12345",
|
|
content="Album test",
|
|
media=img_paths
|
|
)
|
|
|
|
with patch("nanobot.channels.telegram._markdown_to_telegram_html", return_value="Album test"):
|
|
await channel.send(msg)
|
|
|
|
# Verify send_media_group was called
|
|
mock_telegram_app.bot.send_media_group.assert_called_once()
|
|
call_args = mock_telegram_app.bot.send_media_group.call_args
|
|
assert call_args.kwargs["chat_id"] == 12345
|
|
assert len(call_args.kwargs["media"]) == 3
|