331 lines
10 KiB
Python
331 lines
10 KiB
Python
# -*- 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
|
||
HELP_TEXT,
|
||
TelegramClient,
|
||
TelegramError,
|
||
format_duration,
|
||
format_manual_reply,
|
||
)
|
||
from reaction_board.services.library import ( # noqa: E402
|
||
ActionResult,
|
||
IngestResult,
|
||
ManualIngestResult,
|
||
)
|
||
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_manual_reply_truncation_notice():
|
||
result = ManualIngestResult(
|
||
refs=50, added=1, truncated=2,
|
||
results=[IngestResult(True, 1, "added", title="t", message="已入库")],
|
||
)
|
||
text = format_manual_reply(result)
|
||
assert "共识别 50 个引用" in text
|
||
assert "还有 2 个引用未处理" in text
|
||
assert format_manual_reply(ManualIngestResult()) == (
|
||
"未识别到 B 站视频引用，请发送视频链接或 BV 号，例如 BV1x8x166EzY。"
|
||
)
|
||
|
||
|
||
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]] == [
|
||
"add: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"]] == [
|
||
"add:1", "add: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 transport.calls[0]["timeout"] == 32
|
||
assert json.loads(data["allowed_updates"]) == ["callback_query", "message"]
|
||
|
||
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 = []
|
||
self.manual = []
|
||
|
||
def ingest_candidate(self, aid):
|
||
self.calls.append(("add", aid))
|
||
return IngestResult(
|
||
True, aid, "added", title="咻咻满 reaction", 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 ingest_text(self, text):
|
||
self.manual.append(text)
|
||
return ManualIngestResult(
|
||
refs=1, added=1,
|
||
results=[IngestResult(True, 9, "added", title="咻咻满 reaction", message="已入库")],
|
||
)
|
||
|
||
|
||
ADD_REPLY = "✅ 已入库《咻咻满 reaction》（aid 9）"
|
||
|
||
|
||
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 message_update(update_id, text, *, chat_id=42, from_id=42):
|
||
return {
|
||
"update_id": update_id,
|
||
"message": {
|
||
"message_id": update_id * 10,
|
||
"chat": {"id": chat_id},
|
||
"from": {"id": from_id},
|
||
"text": text,
|
||
},
|
||
}
|
||
|
||
|
||
def test_poller_dispatch_and_authorization():
|
||
bot = FakeBot([
|
||
update(1, "add:9"),
|
||
update(2, "ign:10"),
|
||
update(3, "blk:11"),
|
||
update(4, "add:12", chat_id=999),
|
||
update(5, "bad"),
|
||
update(6, "fav:13"),
|
||
])
|
||
actions = FakeActions()
|
||
poller = TelegramPoller(bot, actions)
|
||
offset = poller.run_once(None)
|
||
assert offset == 7
|
||
assert actions.calls == [("add", 9), ("ign", 10), ("blk", 11), ("add", 13)]
|
||
assert bot.answers[0] == ("cb1", ADD_REPLY)
|
||
assert bot.answers[1] == ("cb2", "• 已忽略（aid 10）")
|
||
assert bot.answers[2] == ("cb3", "• 已拉黑（aid 11）")
|
||
assert bot.answers[3] == ("cb4", "无权限")
|
||
assert bot.answers[4] == ("cb5", "未知操作")
|
||
assert bot.edits == [10, 20, 30, 60]
|
||
|
||
|
||
def test_poller_manual_message():
|
||
bot = FakeBot([
|
||
message_update(1, "https://www.bilibili.com/video/BV1x8x166EzY"),
|
||
message_update(2, "hello", chat_id=999),
|
||
message_update(3, "/help"),
|
||
])
|
||
actions = FakeActions()
|
||
poller = TelegramPoller(bot, actions)
|
||
poller.run_once(None)
|
||
assert actions.manual == ["https://www.bilibili.com/video/BV1x8x166EzY"]
|
||
assert "入库" in bot.sent[0]
|
||
assert HELP_TEXT in bot.sent
|
||
assert len(bot.sent) == 2
|
||
|
||
|
||
def test_answer_failure_falls_back_to_text():
|
||
bot = FakeBot([update(1, "add:9")], fail_answer=True)
|
||
actions = FakeActions()
|
||
poller = TelegramPoller(bot, actions)
|
||
poller.run_once(None)
|
||
assert actions.calls == [("add", 9)]
|
||
assert bot.answers == []
|
||
assert bot.sent == [ADD_REPLY]
|
||
assert bot.edits == [10]
|
||
|
||
|
||
class Recorder:
|
||
def __init__(self):
|
||
self.events = []
|
||
|
||
def _add(self, level, event, fields):
|
||
self.events.append((level, event, fields))
|
||
|
||
def info(self, event, **fields):
|
||
self._add("info", event, fields)
|
||
|
||
def warning(self, event, **fields):
|
||
self._add("warning", event, fields)
|
||
|
||
def error(self, event, **fields):
|
||
self._add("error", event, fields)
|
||
|
||
def debug(self, event, **fields):
|
||
self._add("debug", event, fields)
|
||
|
||
|
||
class FlakyBot(FakeBot):
|
||
def __init__(self):
|
||
super().__init__([])
|
||
self.calls = 0
|
||
|
||
def get_updates(self, offset=None):
|
||
self.calls += 1
|
||
if self.calls < 3:
|
||
raise OSError("SSL: UNEXPECTED_EOF_WHILE_READING")
|
||
return []
|
||
|
||
|
||
def test_poller_backoff_and_recovery():
|
||
sleeps = []
|
||
recorder = Recorder()
|
||
bot = FlakyBot()
|
||
poller = TelegramPoller(
|
||
bot, FakeActions(), logger=recorder, sleeper=sleeps.append
|
||
)
|
||
poller.run_forever(initial_delay=3, max_delay=60, max_iterations=3)
|
||
assert bot.calls == 3
|
||
assert sleeps == [3, 6]
|
||
assert [event[0] for event in recorder.events[:2]] == ["warning", "warning"]
|
||
assert any(event[1] == "telegram.poll_recovered" for event in recorder.events)
|
||
|
||
|
||
def main():
|
||
test_format_duration()
|
||
test_manual_reply_truncation_notice()
|
||
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_poller_manual_message()
|
||
test_answer_failure_falls_back_to_text()
|
||
test_poller_backoff_and_recovery()
|
||
print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|