7cf77adbc8
* 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>
143 lines
4.9 KiB
Python
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
|