"""@机器人 指令的端到端装配测试（FakeApi 注入，不发起网络请求）。""" import asyncio import os from qqbotsdk import Bot from qqbotsdk.api import BotApi from authbot.config import AuthConfig from authbot.handlers import setup from authbot.state import PendingRecord, pending_key class FakeApi(BotApi): """替身：只记录调用，不出网。""" def __init__(self): # noqa: D107 - 跳过 BaseApi.__init__（不需要其组件） self.group_messages: list[dict] = [] self.raw_posts: list[tuple[str, dict | None]] = [] self.mute_effective = True # 置 False 模拟禁言被服务端静默忽略 self._muted: dict[str, str] = {} async def post_group_message( self, group_openid, content="", msg_id=None, msg_seq=1, media=None, event_id=None, markdown=None, keyboard=None, **extra, ): self.group_messages.append({ "group_openid": group_openid, "content": content, "msg_id": msg_id, "event_id": event_id, "markdown": markdown, "keyboard": keyboard, }) return {"id": "fake-msg"} async def post(self, path, **kwargs): self.raw_posts.append((path, kwargs.get("json"))) return {} async def mute_group_members(self, group_openid, members): self.raw_posts.append((f"/v2/groups/{group_openid}/restrict_chat_setting", {"members": members})) for entry in members: if entry.get("op") == "del": self._muted.pop(entry.get("member_openid"), None) else: self._muted[entry.get("member_openid")] = entry.get("mute_expire_at") return {} async def get_group_mute_state(self, group_openid): members = [ {"member_openid": m, "mute_expire_at": exp} for m, exp in self._muted.items() ] if self.mute_effective else [] return {"global_rule": {"mode": "none"}, "members": members} async def get(self, path, **kwargs): return {} def make_bot() -> Bot: os.environ.setdefault("APPID", "test-appid") os.environ.setdefault("APPSECRET", "test-secret") return Bot(intents=0) def at_message_payload(content, member="MEMBER123456", msg_id="m1", group="GROUP1"): return { "op": 0, "id": msg_id, "t": "GROUP_AT_MESSAGE_CREATE", "d": { "id": msg_id, "content": content, "group_openid": group, "author": {"member_openid": member, "union_openid": ""}, }, } def group_message_payload(content, member="MEMBER123456", msg_id="g1", group="GROUP1"): """全量消息模式（GROUP_MESSAGE_CREATE），字段与 @ 消息一致。""" payload = at_message_payload(content, member, msg_id, group) payload["t"] = "GROUP_MESSAGE_CREATE" return payload def test_command_works_in_all_message_mode(tmp_path, monkeypatch): """平台"接收全部消息"模式下，指令经 GROUP_MESSAGE_CREATE 到达也能触发。""" async def scenario(): monkeypatch.setenv("DATA_DIR", str(tmp_path)) bot = make_bot() cfg = AuthConfig.load() state = setup(bot, cfg) fake = FakeApi() bot.services[BotApi] = fake await bot._emitter.emit(group_message_payload(" @AuthBot 验证状态")) await asyncio.sleep(0.05) assert len(fake.group_messages) == 1 assert "没有待验证" in (fake.group_messages[0]["content"] or "") assert state.all_pending() == [] asyncio.run(scenario()) def test_verify_me_command_runs_full_flow(tmp_path, monkeypatch): async def scenario(): monkeypatch.setenv("DATA_DIR", str(tmp_path)) bot = make_bot() cfg = AuthConfig.load() state = setup(bot, cfg) fake = FakeApi() bot.services[BotApi] = fake # handler 按类型标注从 services 拿到替身 await bot._emitter.emit(at_message_payload(" 验证我")) await asyncio.sleep(0.05) # background handler 异步执行 # 发出了带按钮的验证码，且是对指令消息的被动回复（msg_id 凭据） assert len(fake.group_messages) == 1 sent = fake.group_messages[0] assert sent["msg_id"] == "m1" assert sent["markdown"] and "验证测试" in sent["markdown"] assert sent["keyboard"] # 对发起人本人按成员禁言（官方嵌套结构：members[].op=add） mute_calls = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")] assert len(mute_calls) == 1 entry = mute_calls[0]["members"][0] assert entry["op"] == "add" assert entry["member_openid"] == "MEMBER123456" assert "mute_expire_at" in entry # 进入 pending，按钮 data 里的 token 与记录一致 key = pending_key("GROUP1", "MEMBER123456") record = state.get(key) assert record is not None btn = sent["keyboard"]["content"]["rows"][0]["buttons"][0] assert btn["action"]["data"].split("|")[1] == record.token state.pop(key) # 清理，避免定时器任务悬挂到用例外 asyncio.run(scenario()) def test_verify_me_keeps_original_deadline(tmp_path, monkeypatch): """已在验证中再次发起测试：换题但倒计时不重置。""" async def scenario(): import time monkeypatch.setenv("DATA_DIR", str(tmp_path)) bot = make_bot() cfg = AuthConfig.load() state = setup(bot, cfg) bot.services[BotApi] = FakeApi() key = pending_key("GROUP1", "MEMBER123456") old_deadline = time.time() + 123.0 state.put(PendingRecord( group_openid="GROUP1", member_openid="MEMBER123456", deadline=old_deadline, token="old", answer=1, created_at=time.time(), )) await bot._emitter.emit(at_message_payload(" 测试验证", msg_id="m2")) await asyncio.sleep(0.05) record = state.get(key) assert record is not None assert record.deadline == old_deadline assert record.token != "old" # 题目已更换 state.pop(key) asyncio.run(scenario()) def test_verify_me_reports_ineffective_mute(tmp_path, monkeypatch): """禁言被服务端静默忽略时（群主/管理员），验证我 应明确提示原因。""" async def scenario(): monkeypatch.setenv("DATA_DIR", str(tmp_path)) bot = make_bot() cfg = AuthConfig.load() state = setup(bot, cfg) fake = FakeApi() fake.mute_effective = False bot.services[BotApi] = fake await bot._emitter.emit(at_message_payload(" 验证我")) await asyncio.sleep(0.05) # 验证码 + 一条"禁言未生效"提示 assert len(fake.group_messages) == 2 assert fake.group_messages[0]["markdown"] assert "禁言未能生效" in (fake.group_messages[1]["content"] or "") key = pending_key("GROUP1", "MEMBER123456") assert state.get(key) is not None # 验证流程仍然继续 state.pop(key) asyncio.run(scenario()) def test_status_command_no_side_effects(tmp_path, monkeypatch): async def scenario(): monkeypatch.setenv("DATA_DIR", str(tmp_path)) bot = make_bot() cfg = AuthConfig.load() state = setup(bot, cfg) fake = FakeApi() bot.services[BotApi] = fake await bot._emitter.emit(at_message_payload("验证状态", msg_id="m3")) await asyncio.sleep(0.05) assert len(fake.group_messages) == 1 assert "没有待验证" in (fake.group_messages[0]["content"] or "") assert fake.raw_posts == [] # 不触发禁言等写操作 assert state.all_pending() == [] asyncio.run(scenario())