Files
camera-2018 7cf77adbc8 feat(telegram): supports sendMessageDraft API (#5726)
* feat(telegram): 使用 sendMessageDraft API 实现私聊流式输出

- 新增 _send_message_draft 方法封装 Telegram Bot API sendMessageDraft
- 私聊流式输出使用 sendMessageDraft 推送草稿动画,群聊保留 edit_message_text 回退
- 使用独立异步发送循环 (_draft_sender_loop) 按固定间隔推送最新缓冲区内容,
  完全解耦 token 到达速度与 API 网络延迟
- 流式结束后发送真实消息保留最终内容(draft 是临时的)
- 使用模块级递增 draft_id 替代随机生成,确保 Telegram 端动画连续性

* fix(telegram): convert draft text to Markdown before sending message draft

* chore(telegram): telegram 适配器重构

- 提取公共方法
- 有新 token 到达时触发流式
- 生成结束后清除draft内容
- 默认draft发送md格式

* style(telegram): ruff format

* style(telegram): ruff check

---------

Co-authored-by: Soulter <905617992@qq.com>
2026-03-05 11:20:28 +08:00

143 lines
4.9 KiB
Python

"""Telegram 模块 Mock 工具。
提供统一的 Telegram 相关模块 mock 设置,避免在测试文件中重复定义。
"""
import sys
from unittest.mock import AsyncMock, MagicMock
import pytest
def create_mock_telegram_modules():
"""创建 Telegram 相关的 mock 模块。
Returns:
dict: 包含 telegram 和相关模块的 mock 对象
"""
mock_telegram = MagicMock()
mock_telegram.BotCommand = MagicMock
mock_telegram.Update = MagicMock
mock_telegram.constants = MagicMock()
mock_telegram.constants.ChatType = MagicMock()
mock_telegram.constants.ChatType.PRIVATE = "private"
mock_telegram.constants.ChatAction = MagicMock()
mock_telegram.constants.ChatAction.TYPING = "typing"
mock_telegram.constants.ChatAction.UPLOAD_VOICE = "upload_voice"
mock_telegram.constants.ChatAction.UPLOAD_DOCUMENT = "upload_document"
mock_telegram.constants.ChatAction.UPLOAD_PHOTO = "upload_photo"
mock_telegram.error = MagicMock()
mock_telegram.error.BadRequest = Exception
mock_telegram.ReactionTypeCustomEmoji = MagicMock
mock_telegram.ReactionTypeEmoji = MagicMock
mock_telegram_ext = MagicMock()
mock_telegram_ext.ApplicationBuilder = MagicMock
mock_telegram_ext.ContextTypes = MagicMock
mock_telegram_ext.ExtBot = MagicMock
mock_telegram_ext.filters = MagicMock()
mock_telegram_ext.filters.ALL = MagicMock()
mock_telegram_ext.MessageHandler = MagicMock
# Mock telegramify_markdown
mock_telegramify = MagicMock()
mock_telegramify.markdownify = lambda text, **kwargs: text
# Mock apscheduler
mock_apscheduler = MagicMock()
mock_apscheduler.schedulers = MagicMock()
mock_apscheduler.schedulers.asyncio = MagicMock()
mock_apscheduler.schedulers.asyncio.AsyncIOScheduler = MagicMock
mock_apscheduler.schedulers.background = MagicMock()
mock_apscheduler.schedulers.background.BackgroundScheduler = MagicMock
return {
"telegram": mock_telegram,
"telegram.ext": mock_telegram_ext,
"telegramify_markdown": mock_telegramify,
"apscheduler": mock_apscheduler,
}
@pytest.fixture(scope="module", autouse=True)
def mock_telegram_modules():
"""Mock Telegram 相关模块的 fixture。
自动应用于使用此 fixture 的测试模块。
"""
mocks = create_mock_telegram_modules()
monkeypatch = pytest.MonkeyPatch()
monkeypatch.setitem(sys.modules, "telegram", mocks["telegram"])
monkeypatch.setitem(sys.modules, "telegram.constants", mocks["telegram"].constants)
monkeypatch.setitem(sys.modules, "telegram.error", mocks["telegram"].error)
monkeypatch.setitem(sys.modules, "telegram.ext", mocks["telegram.ext"])
monkeypatch.setitem(sys.modules, "telegramify_markdown", mocks["telegramify_markdown"])
monkeypatch.setitem(sys.modules, "apscheduler", mocks["apscheduler"])
monkeypatch.setitem(
sys.modules, "apscheduler.schedulers", mocks["apscheduler"].schedulers
)
monkeypatch.setitem(
sys.modules,
"apscheduler.schedulers.asyncio",
mocks["apscheduler"].schedulers.asyncio,
)
monkeypatch.setitem(
sys.modules,
"apscheduler.schedulers.background",
mocks["apscheduler"].schedulers.background,
)
yield
monkeypatch.undo()
class MockTelegramBuilder:
"""构建 Telegram 测试 mock 对象的工具类。"""
@staticmethod
def create_bot():
"""创建 mock Telegram bot 实例。"""
bot = MagicMock()
bot.username = "test_bot"
bot.id = 12345678
bot.base_url = "https://api.telegram.org/bottest_token_123/"
bot.send_message = AsyncMock()
bot.send_photo = AsyncMock()
bot.send_document = AsyncMock()
bot.send_voice = AsyncMock()
bot.send_chat_action = AsyncMock()
bot.delete_my_commands = AsyncMock()
bot.set_my_commands = AsyncMock()
bot.set_message_reaction = AsyncMock()
bot.edit_message_text = AsyncMock()
bot.send_message_draft = AsyncMock()
return bot
@staticmethod
def create_application():
"""创建 mock Telegram Application 实例。"""
from tests.fixtures.helpers import NoopAwaitable
app = MagicMock()
app.bot = MagicMock()
app.bot.username = "test_bot"
app.bot.base_url = "https://api.telegram.org/bottest_token_123/"
app.initialize = AsyncMock()
app.start = AsyncMock()
app.stop = AsyncMock()
app.add_handler = MagicMock()
app.updater = MagicMock()
app.updater.start_polling = MagicMock(return_value=NoopAwaitable())
app.updater.stop = AsyncMock()
return app
@staticmethod
def create_scheduler():
"""创建 mock APScheduler 实例。"""
scheduler = MagicMock()
scheduler.add_job = MagicMock()
scheduler.start = MagicMock()
scheduler.running = True
scheduler.shutdown = MagicMock()
return scheduler