snowy-yang
755013b9db
基于 qqbotsdk-py 的入群验证：进群发按钮验证码、完成前禁言、 超时踢出（11253 时兜底长期禁言）、乱点他人按钮处罚、状态持久化 与重启恢复。38 个单元/集成测试。
224 lines
7.7 KiB
Python
224 lines
7.7 KiB
Python
"""@机器人 指令的端到端装配测试（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())
|