mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
a5e86c8b94
* fix(telegram): preserve attachment captions * Update astrbot/core/platform/sources/telegram/tg_adapter.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
import asyncio
|
|
import importlib
|
|
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import astrbot.api.message_components as Comp
|
|
from tests.fixtures.helpers import (
|
|
create_mock_file,
|
|
create_mock_update,
|
|
make_platform_config,
|
|
)
|
|
from tests.fixtures.mocks.telegram import create_mock_telegram_modules
|
|
|
|
_TELEGRAM_PLATFORM_ADAPTER = None
|
|
|
|
|
|
def _load_telegram_adapter():
|
|
global _TELEGRAM_PLATFORM_ADAPTER
|
|
if _TELEGRAM_PLATFORM_ADAPTER is not None:
|
|
return _TELEGRAM_PLATFORM_ADAPTER
|
|
|
|
mocks = create_mock_telegram_modules()
|
|
patched_modules = {
|
|
"telegram": mocks["telegram"],
|
|
"telegram.constants": mocks["telegram"].constants,
|
|
"telegram.error": mocks["telegram"].error,
|
|
"telegram.ext": mocks["telegram.ext"],
|
|
"telegramify_markdown": mocks["telegramify_markdown"],
|
|
"apscheduler": mocks["apscheduler"],
|
|
"apscheduler.schedulers": mocks["apscheduler"].schedulers,
|
|
"apscheduler.schedulers.asyncio": mocks["apscheduler"].schedulers.asyncio,
|
|
"apscheduler.schedulers.background": mocks["apscheduler"].schedulers.background,
|
|
}
|
|
with patch.dict(sys.modules, patched_modules):
|
|
sys.modules.pop("astrbot.core.platform.sources.telegram.tg_adapter", None)
|
|
module = importlib.import_module("astrbot.core.platform.sources.telegram.tg_adapter")
|
|
_TELEGRAM_PLATFORM_ADAPTER = module.TelegramPlatformAdapter
|
|
return _TELEGRAM_PLATFORM_ADAPTER
|
|
|
|
|
|
def _build_context() -> MagicMock:
|
|
context = MagicMock()
|
|
context.bot.username = "test_bot"
|
|
context.bot.id = 12345678
|
|
return context
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_document_caption_populates_message_text_and_plain():
|
|
TelegramPlatformAdapter = _load_telegram_adapter()
|
|
adapter = TelegramPlatformAdapter(
|
|
make_platform_config("telegram"),
|
|
{},
|
|
asyncio.Queue(),
|
|
)
|
|
document = create_mock_file("https://api.telegram.org/file/test/report.md")
|
|
document.file_name = "report.md"
|
|
mention = MagicMock(type="mention", offset=0, length=6)
|
|
update = create_mock_update(
|
|
message_text=None,
|
|
document=document,
|
|
caption="@alice 请总结这份文档",
|
|
caption_entities=[mention],
|
|
)
|
|
|
|
result = await adapter.convert_message(update, _build_context())
|
|
|
|
assert result is not None
|
|
assert result.message_str == "@alice 请总结这份文档"
|
|
assert any(isinstance(component, Comp.File) for component in result.message)
|
|
assert any(
|
|
isinstance(component, Comp.Plain)
|
|
and component.text == "@alice 请总结这份文档"
|
|
for component in result.message
|
|
)
|
|
assert any(
|
|
isinstance(component, Comp.At) and component.qq == "alice"
|
|
for component in result.message
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telegram_video_caption_populates_message_text_and_plain():
|
|
TelegramPlatformAdapter = _load_telegram_adapter()
|
|
adapter = TelegramPlatformAdapter(
|
|
make_platform_config("telegram"),
|
|
{},
|
|
asyncio.Queue(),
|
|
)
|
|
video = create_mock_file("https://api.telegram.org/file/test/lesson.mp4")
|
|
video.file_name = "lesson.mp4"
|
|
update = create_mock_update(
|
|
message_text=None,
|
|
video=video,
|
|
caption="这段视频讲了什么",
|
|
)
|
|
|
|
result = await adapter.convert_message(update, _build_context())
|
|
|
|
assert result is not None
|
|
assert result.message_str == "这段视频讲了什么"
|
|
assert any(isinstance(component, Comp.Video) for component in result.message)
|
|
assert any(
|
|
isinstance(component, Comp.Plain) and component.text == "这段视频讲了什么"
|
|
for component in result.message
|
|
)
|