kairee
4a9c37f2f3
私聊新增 /candidate [页码]（待决策列表）、/candi_del（标题按钮逐条删除，删后原地重建键盘、可翻页连续删）、/candi_clear（全部/待决策/取消）。重构 handle_callback 为前缀分派（cdel/cpage/cclr，保留 add/ign/blk/fav）；repository 加 list_pending_candidates/delete_all_candidates；LibraryService 加 pending_candidates/delete_candidate/clear_candidates 并刷看板缓存。
481 lines
15 KiB
Python
481 lines
15 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
|
||
CANDIDATE_DELETE_HEADER,
|
||
HELP_TEXT,
|
||
TelegramClient,
|
||
TelegramError,
|
||
candidate_delete_keyboard,
|
||
clear_candidates_keyboard,
|
||
format_candidate_list,
|
||
format_duration,
|
||
format_manual_reply,
|
||
truncate_title,
|
||
)
|
||
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.markups = []
|
||
self.edits = []
|
||
self.edit_markups = []
|
||
|
||
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, reply_markup=None):
|
||
self.sent.append(text)
|
||
self.markups.append(reply_markup)
|
||
|
||
def edit_reply_markup(self, message_id, reply_markup=None):
|
||
self.edits.append(message_id)
|
||
self.edit_markups.append(reply_markup)
|
||
|
||
|
||
class FakeActions:
|
||
def __init__(self):
|
||
self.calls = []
|
||
self.manual = []
|
||
self.candidates = []
|
||
self.clears = []
|
||
|
||
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="已入库")],
|
||
)
|
||
|
||
def pending_candidates(self):
|
||
return list(self.candidates)
|
||
|
||
def delete_candidate(self, aid):
|
||
self.calls.append(("cdel", aid))
|
||
for index, candidate in enumerate(self.candidates):
|
||
if candidate["aid"] == aid:
|
||
return self.candidates.pop(index)
|
||
return None
|
||
|
||
def clear_candidates(self, *, pending_only=False):
|
||
self.clears.append(pending_only)
|
||
total = len(self.candidates)
|
||
if pending_only:
|
||
self.candidates = [
|
||
c for c in self.candidates if c.get("decision") is not None
|
||
]
|
||
return total - len(self.candidates)
|
||
self.candidates = []
|
||
return total
|
||
|
||
|
||
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]
|
||
|
||
|
||
def candidates(n, *, start=1):
|
||
return [
|
||
{"aid": 100 + i, "title": f"咻咻满 reaction {i}"}
|
||
for i in range(start, start + n)
|
||
]
|
||
|
||
|
||
def test_truncate_and_keyboard_helpers():
|
||
assert truncate_title("短标题") == "短标题"
|
||
long_title = "咻咻满" + "反" * 40
|
||
out = truncate_title(long_title, 20)
|
||
assert len(out) == 20 and out.endswith("…")
|
||
assert truncate_title("") == "(无标题)"
|
||
|
||
items = candidates(12)
|
||
text = format_candidate_list(items, 1)
|
||
assert "共 12 条 · 第 1/2 页" in text
|
||
assert "咻咻满 reaction 1" in text
|
||
assert "咻咻满 reaction 11" not in text
|
||
assert "第 2/2 页" in format_candidate_list(items, 2)
|
||
assert "第 2/2 页" in format_candidate_list(items, 99) # 页码越界夹取到末页
|
||
assert format_candidate_list([], 1) == "暂无待决策候选。"
|
||
|
||
keyboard = candidate_delete_keyboard(items, 1)
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "cdel:1:101"
|
||
assert keyboard["inline_keyboard"][-1][0]["callback_data"] == "cpage:del:0"
|
||
clear_kb = clear_candidates_keyboard()
|
||
assert [b["callback_data"] for b in clear_kb["inline_keyboard"][0]] == [
|
||
"cclr:all", "cclr:pending", "cclr:cancel",
|
||
]
|
||
|
||
|
||
def test_candidate_list_command_pages():
|
||
bot = FakeBot([
|
||
message_update(1, "/candidate"),
|
||
message_update(2, "/candidate 2"),
|
||
message_update(3, "/candidate@re_xxm_bot"),
|
||
])
|
||
actions = FakeActions()
|
||
actions.candidates = candidates(12)
|
||
TelegramPoller(bot, actions).run_once(None)
|
||
assert "共 12 条 · 第 1/2 页" in bot.sent[0]
|
||
assert "第 2/2 页" in bot.sent[1]
|
||
assert "第 1/2 页" in bot.sent[2] # @BotName 后缀也能解析
|
||
|
||
|
||
def test_candi_del_keyboard_and_rebuild():
|
||
bot = FakeBot([
|
||
message_update(1, "/candi_del"),
|
||
update(2, "cdel:1:101"),
|
||
update(3, "cpage:del:2"),
|
||
])
|
||
actions = FakeActions()
|
||
actions.candidates = candidates(12)
|
||
TelegramPoller(bot, actions).run_once(None)
|
||
|
||
assert bot.sent[0] == CANDIDATE_DELETE_HEADER
|
||
keyboard = bot.markups[0]
|
||
assert keyboard["inline_keyboard"][0][0]["text"] == "咻咻满 reaction 1"
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "cdel:1:101"
|
||
|
||
assert ("cdel", 101) in actions.calls
|
||
assert bot.answers[0] == ("cb2", "✅ 已删除：咻咻满 reaction 1")
|
||
# 删除 101 后重建第一页键盘（新首条 102）
|
||
assert bot.edit_markups[0]["inline_keyboard"][0][0]["callback_data"] == "cdel:1:102"
|
||
|
||
# 翻到第 2 页（剩余 11 条：102..112 → 第 2 页仅 112）
|
||
assert bot.answers[1] == ("cb3", "第 2 页")
|
||
assert bot.edit_markups[1]["inline_keyboard"][0][0]["callback_data"] == "cdel:2:112"
|
||
|
||
|
||
def test_candi_del_empty():
|
||
bot = FakeBot([message_update(1, "/candi_del")])
|
||
TelegramPoller(bot, FakeActions()).run_once(None)
|
||
assert bot.sent == ["暂无待决策候选。"]
|
||
assert bot.markups == [None]
|
||
|
||
|
||
def test_candi_clear_buttons():
|
||
bot = FakeBot([
|
||
message_update(1, "/candi_clear"),
|
||
update(2, "cclr:pending"),
|
||
update(3, "cclr:cancel"),
|
||
])
|
||
actions = FakeActions()
|
||
actions.candidates = candidates(3) + [
|
||
{"aid": 999, "title": "已决策", "decision": "rejected"}
|
||
]
|
||
TelegramPoller(bot, actions).run_once(None)
|
||
|
||
keyboard = bot.markups[0]
|
||
assert [b["callback_data"] for b in keyboard["inline_keyboard"][0]] == [
|
||
"cclr:all", "cclr:pending", "cclr:cancel",
|
||
]
|
||
assert actions.clears == [True] # 只清待决策
|
||
assert bot.answers[0] == ("cb2", "已清空 3 条")
|
||
assert bot.edit_markups[0] == {"inline_keyboard": []}
|
||
assert bot.answers[1] == ("cb3", "已取消")
|
||
|
||
|
||
def test_candidate_commands_unauthorized_and_unknown():
|
||
bot = FakeBot([message_update(1, "/candidate", chat_id=999)])
|
||
actions = FakeActions()
|
||
actions.candidates = candidates(3)
|
||
TelegramPoller(bot, actions).run_once(None)
|
||
assert bot.sent == []
|
||
|
||
bot2 = FakeBot([message_update(1, "/nope")])
|
||
TelegramPoller(bot2, FakeActions()).run_once(None)
|
||
assert bot2.sent == [HELP_TEXT]
|
||
|
||
|
||
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_truncate_and_keyboard_helpers()
|
||
test_candidate_list_command_pages()
|
||
test_candi_del_keyboard_and_rebuild()
|
||
test_candi_del_empty()
|
||
test_candi_clear_buttons()
|
||
test_candidate_commands_unauthorized_and_unknown()
|
||
test_poller_backoff_and_recovery()
|
||
print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|