215 lines
7.8 KiB
Python
215 lines
7.8 KiB
Python
"""INTERACTION_CREATE 按钮回调流程测试（FakeApi 注入，无网络）。
|
||
|
||
覆盖：本人答对通过、非本人点击计数与处罚、旧题点击不处罚。
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import time
|
||
|
||
from qqbotsdk import Bot
|
||
from qqbotsdk.api import BotApi
|
||
|
||
from authbot import captcha
|
||
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.interactions_responded: list[str] = []
|
||
self.mute_effective = True
|
||
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 {}
|
||
|
||
async def respond_interaction(self, interaction_id, code=0):
|
||
self.interactions_responded.append(interaction_id)
|
||
return {}
|
||
|
||
|
||
def make_bot() -> Bot:
|
||
os.environ.setdefault("APPID", "test-appid")
|
||
os.environ.setdefault("APPSECRET", "test-secret")
|
||
return Bot(intents=0)
|
||
|
||
|
||
def interaction_payload(token, value, member, group="GROUP1", seq=1):
|
||
return {
|
||
"op": 0,
|
||
"id": f"INTERACTION_CREATE:evt-{seq}",
|
||
"t": "INTERACTION_CREATE",
|
||
"d": {
|
||
"id": f"inner-uuid-{seq}",
|
||
"type": 11,
|
||
"chat_type": 1,
|
||
"group_openid": group,
|
||
"group_member_openid": member,
|
||
"data": {"resolved": {"button_data": captcha.encode_button_data(token, value)}},
|
||
},
|
||
}
|
||
|
||
|
||
def test_subject_correct_answer_passes(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
|
||
|
||
key = pending_key("GROUP1", "MEMBER_A")
|
||
state.put(PendingRecord(
|
||
group_openid="GROUP1", member_openid="MEMBER_A",
|
||
deadline=time.time() + 600, token="subj", answer=7, created_at=time.time(),
|
||
))
|
||
|
||
await bot._emitter.emit(interaction_payload("subj", 7, "MEMBER_A", seq=1))
|
||
await asyncio.sleep(0.05)
|
||
|
||
# 解禁（op=del）+ 通过播报
|
||
mutes = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")]
|
||
assert len(mutes) == 1
|
||
entry = mutes[0]["members"][0]
|
||
assert entry["op"] == "del"
|
||
assert entry["member_openid"] == "MEMBER_A"
|
||
assert "验证通过" in (fake.group_messages[-1]["content"] or "")
|
||
assert state.get(key) is None # pending 已清除
|
||
assert state.is_verified_recently(key, 99)
|
||
|
||
asyncio.run(scenario())
|
||
|
||
|
||
def test_foreign_clicker_muted_on_third_click(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
|
||
|
||
# MEMBER_A 的待验证题
|
||
state.put(PendingRecord(
|
||
group_openid="GROUP1", member_openid="MEMBER_A",
|
||
deadline=time.time() + 600, token="subj", answer=7, created_at=time.time(),
|
||
))
|
||
|
||
# 旁观者 MEMBER_B 连点 3 次别人的题
|
||
for seq in range(1, 4):
|
||
await bot._emitter.emit(interaction_payload("subj", 9, "MEMBER_B", seq=seq))
|
||
await asyncio.sleep(0.03)
|
||
|
||
# 每次点击都回应了互动
|
||
assert len(fake.interactions_responded) == 3
|
||
# 前两次只有警告（markdown 内联提及点击者），第三次才禁言 5 分钟
|
||
warns = [(m["markdown"] or "") for m in fake.group_messages[:2]]
|
||
assert all("不是给你的" in md for md in warns)
|
||
assert all(f"<@MEMBER_B>" in md for md in warns)
|
||
mutes = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")]
|
||
assert len(mutes) == 1
|
||
entry = mutes[0]["members"][0]
|
||
assert entry["op"] == "add"
|
||
assert entry["member_openid"] == "MEMBER_B"
|
||
assert "已禁言 5 分钟" in (fake.group_messages[-1]["markdown"] or "")
|
||
assert "<@MEMBER_B>" in (fake.group_messages[-1]["markdown"] or "")
|
||
# 被试者的记录不受影响
|
||
assert state.get(pending_key("GROUP1", "MEMBER_A")) is not None
|
||
state.pop(pending_key("GROUP1", "MEMBER_A"))
|
||
|
||
asyncio.run(scenario())
|
||
|
||
|
||
def test_stale_button_clicks_counted_as_foreign(tmp_path, monkeypatch):
|
||
"""点已失效的残留按钮同样计入乱点处罚：3 次后禁言。"""
|
||
|
||
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
|
||
|
||
# token 不属于任何在途题目（验证通过后残留的按钮）
|
||
for seq in range(1, 4):
|
||
await bot._emitter.emit(interaction_payload("ghost", 1, "MEMBER_C", seq=seq))
|
||
await asyncio.sleep(0.03)
|
||
|
||
# 前两次带计数的警告（@点击者），第三次禁言
|
||
warns = [(m["markdown"] or "") for m in fake.group_messages[:2]]
|
||
assert all("不是给你的" in md and "<@MEMBER_C>" in md for md in warns)
|
||
assert "再乱点" in warns[0] and "再乱点 1 次" in warns[1]
|
||
mutes = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")]
|
||
assert len(mutes) == 1
|
||
entry = mutes[0]["members"][0]
|
||
assert entry["op"] == "add"
|
||
assert entry["member_openid"] == "MEMBER_C"
|
||
|
||
asyncio.run(scenario())
|
||
|
||
|
||
def test_verified_member_clicking_stale_button_not_punished(tmp_path, monkeypatch):
|
||
"""刚通过验证的成员点自己残留的旧按钮：礼貌回复，不计处罚。"""
|
||
|
||
async def scenario():
|
||
import time as _time
|
||
|
||
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||
monkeypatch.setenv("VERIFIED_TRUST_DAYS", "3") # 开启信任期，已通过成员才免计
|
||
bot = make_bot()
|
||
cfg = AuthConfig.load()
|
||
state = setup(bot, cfg)
|
||
fake = FakeApi()
|
||
bot.services[BotApi] = fake
|
||
|
||
key = pending_key("GROUP1", "MEMBER_C")
|
||
state.mark_verified(key) # trust 窗口内
|
||
await bot._emitter.emit(interaction_payload("ghost", 1, "MEMBER_C", seq=1))
|
||
await asyncio.sleep(0.05)
|
||
|
||
assert fake.raw_posts == [] # 不触发禁言
|
||
assert "已经通过验证" in (fake.group_messages[-1]["content"] or "")
|
||
|
||
asyncio.run(scenario())
|