# -*- coding: utf-8 -*- """Telegram 消息构造与回调分发测试（不联网）: uv run python tests/test_telegram.py """ import json import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.notify.telegram import ( # noqa: E402 TelegramClient, TelegramError, format_duration, ) from reaction_board.services.favorites import ActionResult # noqa: E402 from reaction_board.services.telegram_poller import TelegramPoller # noqa: E402 class FakeTransport: def __init__(self, result=None): self.result = result self.calls = [] def post(self, url, data, timeout=30): self.calls.append({"url": url, "data": dict(data), "timeout": timeout}) return {"ok": True, "result": self.result if self.result is not None else {"message_id": 1}} def candidate(aid, title="咻咻满 reaction", up_name="UP", duration=3725): return { "aid": aid, "title": title, "up_name": up_name, "duration": duration, "play": 12345, "like": 678, } def test_format_duration(): assert format_duration(0) == "0:00" assert format_duration(65) == "1:05" assert format_duration(3725) == "1:02:05" def test_send_candidates_and_chunking(): transport = FakeTransport() bot = TelegramClient("token", 42, transport=transport) assert bot.send_candidates([candidate(1), candidate(2)]) is True assert len(transport.calls) == 1 call = transport.calls[0] assert call["url"].endswith("/sendMessage") assert call["data"]["chat_id"] == 42 assert "咻咻满 reaction" in call["data"]["text"] keyboard = json.loads(call["data"]["reply_markup"]) assert len(keyboard["inline_keyboard"]) == 2 assert [button["callback_data"] for button in keyboard["inline_keyboard"][0]] == [ "fav:1", "ign:1", "blk:1", ] transport = FakeTransport() bot = TelegramClient("token", 42, transport=transport) bot.send_candidates([candidate(i) for i in range(4)]) assert len(transport.calls) == 2 assert len(json.loads(transport.calls[1]["data"]["reply_markup"])["inline_keyboard"]) == 1 def test_send_candidates_dedupes_by_aid(): transport = FakeTransport() bot = TelegramClient("token", 42, transport=transport) bot.send_candidates([candidate(1), candidate(1), candidate(2)]) assert len(transport.calls) == 1 keyboard = json.loads(transport.calls[0]["data"]["reply_markup"]) assert len(keyboard["inline_keyboard"]) == 2 assert [row[0]["callback_data"] for row in keyboard["inline_keyboard"]] == [ "fav:1", "fav:2", ] def test_get_updates_and_answer(): transport = FakeTransport(result=[]) bot = TelegramClient("token", 42, transport=transport, poll_timeout=17) assert bot.get_updates(offset=99) == [] data = transport.calls[0]["data"] assert data["offset"] == 99 assert data["timeout"] == 17 assert json.loads(data["allowed_updates"]) == ["callback_query"] transport = FakeTransport(result=True) bot = TelegramClient("token", 42, transport=transport) bot.answer_callback("cb1", "已加入") assert transport.calls[0]["url"].endswith("/answerCallbackQuery") assert transport.calls[0]["data"]["callback_query_id"] == "cb1" transport = FakeTransport(result=True) bot = TelegramClient("token", 42, transport=transport) bot.edit_reply_markup(77) assert transport.calls[0]["url"].endswith("/editMessageReplyMarkup") assert transport.calls[0]["data"]["message_id"] == 77 assert json.loads(transport.calls[0]["data"]["reply_markup"]) == { "inline_keyboard": [] } def test_telegram_error_description_preserved(): class ErrorTransport: def post(self, url, data, timeout=30): return {"ok": False, "description": "query is too old"} bot = TelegramClient("token", 42, transport=ErrorTransport()) try: bot.answer_callback("cb1", "x") except TelegramError as exc: assert "too old" in str(exc) else: raise AssertionError("Telegram 业务错误应抛出描述") class FakeBot: chat_id = 42 def __init__(self, updates, *, fail_answer=False): self.updates = updates self.fail_answer = fail_answer self.answers = [] self.sent = [] self.edits = [] def get_updates(self, offset=None): return list(self.updates) def answer_callback(self, callback_id, text=""): if self.fail_answer: raise RuntimeError("HTTP Error 400: Bad Request") self.answers.append((callback_id, text)) def send_text(self, text): self.sent.append(text) def edit_reply_markup(self, message_id): self.edits.append(message_id) class FakeActions: def __init__(self): self.calls = [] def favorite(self, aid): self.calls.append(("fav", aid)) return ActionResult(True, aid, "favorite", "done", position=3, message="已加入收藏夹") def ignore(self, aid): self.calls.append(("ign", aid)) return ActionResult(True, aid, "ignore", "rejected", message="已忽略") def blacklist(self, aid): self.calls.append(("blk", aid)) return ActionResult(True, aid, "blacklist", "rejected", message="已拉黑") def update(update_id, data, *, chat_id=42, from_id=42): return { "update_id": update_id, "callback_query": { "id": f"cb{update_id}", "data": data, "from": {"id": from_id}, "message": {"message_id": update_id * 10, "chat": {"id": chat_id}}, }, } def test_poller_dispatch_and_authorization(): bot = FakeBot([ update(1, "fav:9"), update(2, "ign:10"), update(3, "blk:11"), update(4, "fav:12", chat_id=999), update(5, "bad"), ]) actions = FakeActions() poller = TelegramPoller(bot, actions) offset = poller.run_once(None) assert offset == 6 assert actions.calls == [("fav", 9), ("ign", 10), ("blk", 11)] assert bot.answers[0] == ("cb1", "已加入收藏夹（位置 #3）") assert bot.answers[3] == ("cb4", "无权限") assert bot.answers[4] == ("cb5", "未知操作") assert bot.edits == [10, 20, 30] def test_answer_failure_falls_back_to_text(): bot = FakeBot([update(1, "fav:9")], fail_answer=True) actions = FakeActions() poller = TelegramPoller(bot, actions) poller.run_once(None) assert actions.calls == [("fav", 9)] assert bot.answers == [] assert bot.sent == ["已加入收藏夹（位置 #3）"] assert bot.edits == [10] def main(): test_format_duration() test_send_candidates_and_chunking() test_send_candidates_dedupes_by_aid() test_get_updates_and_answer() test_telegram_error_description_preserved() test_poller_dispatch_and_authorization() test_answer_failure_falls_back_to_text() print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过") if __name__ == "__main__": main()