初始提交：QQ 群入群验证机器人

基于 qqbotsdk-py 的入群验证：进群发按钮验证码、完成前禁言、
超时踢出（11253 时兜底长期禁言）、乱点他人按钮处罚、状态持久化
与重启恢复。38 个单元/集成测试。

tests/test_captcha.py

@@ -0,0 +1,70 @@
+"""验证码生成与编解码测试。"""
+
+import pytest
+
+from authbot import captcha
+
+
+def test_generate_basic_properties():
+    for _ in range(50):
+        cap = captcha.generate(4)
+        assert len(cap.options) == 4
+        assert len(set(cap.options)) == 4  # 选项互不重复
+        assert cap.answer in cap.options
+        assert all(v >= 0 for v in cap.options)
+        # 运算式与答案自洽
+        expected = cap.a + cap.b if cap.op == "+" else cap.a - cap.b if cap.op == "-" else cap.a * cap.b
+        assert expected == cap.answer
+        assert cap.op in ("+", "-", "×")
+
+
+def test_generate_option_count_variants():
+    for n in (2, 3, 5):
+        cap = captcha.generate(n)
+        assert len(cap.options) == n
+        assert len(set(cap.options)) == n
+
+
+def test_token_unique_per_captcha():
+    tokens = {captcha.generate().token for _ in range(100)}
+    assert len(tokens) == 100
+
+
+def test_parse_button_data_roundtrip():
+    token, value = "abc123", 42
+    encoded = captcha.encode_button_data(token, value)
+    assert captcha.parse_button_data(encoded) == (token, value)
+
+
+@pytest.mark.parametrize(
+    "data",
+    ["", "v|", "v|abc", "v|abc|x", "hello|abc|1", "cmd|abc|1", "v|abc|1|2"],
+)
+def test_parse_button_data_rejects_invalid(data):
+    assert captcha.parse_button_data(data) is None
+
+
+def test_keyboard_structure():
+    cap = captcha.generate(4)
+    kb = cap.keyboard()
+    rows = kb["content"]["rows"]
+    assert len(rows) == 2  # 每行 2 个按钮
+    flat = [btn for row in rows for btn in row["buttons"]]
+    assert len(flat) == 4
+    labels = {btn["render_data"]["label"] for btn in flat}
+    assert labels == {str(v) for v in cap.options}
+    for btn in flat:
+        token, value = captcha.parse_button_data(btn["action"]["data"])
+        assert token == cap.token
+        assert value in cap.options
+        assert btn["action"]["type"] == 1  # 回调按钮
+
+
+def test_markdown_templates():
+    cap = captcha.generate(4)
+    welcome = captcha.welcome_markdown(cap, 10)
+    assert "10 分钟" in welcome
+    assert cap.question in welcome
+    retry = captcha.retry_markdown(cap)
+    assert cap.question in retry
+    assert "换一题" in retry

tests/test_command.py

@@ -0,0 +1,223 @@
+"""@机器人 指令的端到端装配测试（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())

tests/test_interaction.py

@@ -0,0 +1,178 @@
+"""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
+        # 前两次只有警告，第三次才禁言 5 分钟
+        warns = [m["content"] for m in fake.group_messages[:2]]
+        assert all("不是给你的" in (c or "") for c 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]["content"] 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_click_not_punished(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
+
+        # token 不属于任何在途题目（旧题残留按钮）
+        for seq in range(1, 4):
+            await bot._emitter.emit(interaction_payload("ghost", 1, "MEMBER_C", seq=seq))
+            await asyncio.sleep(0.03)
+
+        assert fake.raw_posts == []  # 三次都不触发禁言
+        assert all("不是给你的" in (m["content"] or "") for m in fake.group_messages)
+
+    asyncio.run(scenario())

tests/test_state.py

@@ -0,0 +1,229 @@
+"""验证状态持久化与超时定时器测试。"""
+
+import asyncio
+import json
+import tempfile
+import time
+from pathlib import Path
+
+from authbot.state import (
+    ForeignClickTracker,
+    PendingRecord,
+    RecentSet,
+    VerifyState,
+    pending_key,
+)
+
+
+def make_record(group="G1", member="M1", deadline=None, token="tok1", answer=7):
+    return PendingRecord(
+        group_openid=group,
+        member_openid=member,
+        deadline=deadline if deadline is not None else time.time() + 600,
+        token=token,
+        answer=answer,
+        created_at=time.time(),
+    )
+
+
+def test_put_get_pop_roundtrip(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        state = VerifyState(tmp_path, on_timeout=noop)
+        rec = make_record()
+        state.put(rec)
+        assert state.get(rec.key) is not None
+
+        # 新实例从磁盘恢复（模拟重启）
+        state2 = VerifyState(tmp_path, on_timeout=noop)
+        restored = state2.get(rec.key)
+        assert restored is not None
+        assert restored.token == rec.token
+        assert restored.answer == rec.answer
+        assert restored.deadline == rec.deadline
+
+        assert state2.pop(rec.key).token == rec.token
+        assert state2.get(rec.key) is None
+        assert state2.pop(rec.key) is None  # 重复 pop 安全
+
+    asyncio.run(scenario())
+
+
+def test_pop_persists_deletion(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        state = VerifyState(tmp_path, on_timeout=noop)
+        rec = make_record()
+        state.put(rec)
+        state.pop(rec.key)
+        assert VerifyState(tmp_path, on_timeout=noop).get(rec.key) is None
+
+    asyncio.run(scenario())
+
+
+def test_verified_trust_window(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        state = VerifyState(tmp_path, on_timeout=noop)
+        key = pending_key("G1", "M1")
+        state.mark_verified(key)
+
+        # trust_days=0：即使刚通过也不免验
+        assert state.is_verified_recently(key, 0) is False
+        assert state.is_verified_recently(key, 1) is True
+
+        # 手动把通过时间改到 3 天前
+        verified = json.loads((tmp_path / "verified.json").read_text("utf-8"))
+        verified[key] = time.time() - 3 * 86400
+        (tmp_path / "verified.json").write_text(json.dumps(verified), "utf-8")
+        state2 = VerifyState(tmp_path, on_timeout=noop)
+        assert state2.is_verified_recently(key, 1) is False
+        assert state2.is_verified_recently(key, 7) is True
+
+    asyncio.run(scenario())
+
+
+def test_summary_remaining_seconds(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        state = VerifyState(tmp_path, on_timeout=noop)
+        rec = make_record(deadline=time.time() + 120)
+        state.put(rec)
+        [(got, remain)] = state.summary()
+        assert got.key == rec.key
+        assert 110 < remain <= 120
+
+    asyncio.run(scenario())
+
+
+def test_timeout_timer_fires_and_pop_cancels(tmp_path):
+    async def scenario():
+        fired = []
+
+        async def on_timeout(record):
+            fired.append(record.key)
+
+        state = VerifyState(tmp_path, on_timeout=on_timeout)
+        # 已到点的：立即触发
+        state.put(make_record(member="EXPIRED", deadline=time.time() - 1))
+        # 未到点的随后被 pop：不应触发
+        state.put(make_record(member="CANCELLED", deadline=time.time() + 5))
+        state.pop(pending_key("G1", "CANCELLED"))
+        await asyncio.sleep(0.3)
+        assert fired == [pending_key("G1", "EXPIRED")]
+
+    asyncio.run(scenario())
+
+
+def test_restore_rearms_from_disk(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        # 先写入一条已过期与一条未过期的记录
+        VerifyState(tmp_path, on_timeout=noop).put(
+            make_record(member="OLD", deadline=time.time() - 1)
+        )
+        VerifyState(tmp_path, on_timeout=noop).put(
+            make_record(member="NEW", deadline=time.time() + 600)
+        )
+
+        fired = []
+
+        async def on_timeout(record):
+            fired.append(record.key)
+
+        state = VerifyState(tmp_path, on_timeout=on_timeout)
+        await state.restore()
+        await asyncio.sleep(0.3)
+        # 只有已过期的触发，未到期的不触发
+        assert fired == [pending_key("G1", "OLD")]
+        assert state.get(pending_key("G1", "NEW")) is not None
+
+    asyncio.run(scenario())
+
+
+def test_timeout_callback_exception_isolated(tmp_path):
+    """回调抛异常不炸定时器任务本身，也不影响其他记录。"""
+
+    async def scenario():
+        async def boom(record):
+            raise RuntimeError("boom")
+
+        state = VerifyState(tmp_path, on_timeout=boom)
+        state.put(make_record(member="A", deadline=time.time() - 1))
+        state.put(make_record(member="B", deadline=time.time() - 1))
+        await asyncio.sleep(0.3)  # 两条回调都跑完且异常被记录
+
+    asyncio.run(scenario())
+
+
+def test_corrupt_state_file_backed_up(tmp_path):
+    async def scenario():
+        async def noop(record):
+            pass
+
+        (tmp_path / "pending.json").write_text("{not json", "utf-8")
+        state = VerifyState(tmp_path, on_timeout=noop)
+        assert state.all_pending() == []
+        assert (tmp_path / "pending.json.bad").exists()
+
+    asyncio.run(scenario())
+
+
+def test_recent_set():
+    s = RecentSet(maxsize=3)
+    assert s.add("a") is True
+    assert s.add("a") is False
+    assert s.add("b") is True
+    assert s.add("c") is True
+    assert s.add("d") is True  # 挤掉最旧的 a
+    assert s.add("a") is True  # a 已被淘汰，重新视为首次
+
+
+def test_foreign_click_tracker_counts_and_resets():
+    t = ForeignClickTracker(limit=3, window=1800.0)
+    now = 1000.0
+    assert t.register("k", now) == 1
+    assert t.register("k", now + 10) == 2
+    assert t.register("k", now + 20) == 3  # 到达上限
+    t.reset("k")
+    assert t.register("k", now + 30) == 1  # 处罚后重新计数
+
+
+def test_foreign_click_tracker_window_prunes_old():
+    t = ForeignClickTracker(limit=3, window=60.0)
+    now = 1000.0
+    assert t.register("k", now) == 1
+    assert t.register("k", now + 61) == 1  # 窗口外的第一次点击不计入
+    assert t.register("k", now + 61 + 30) == 2
+
+
+def test_foreign_click_tracker_keys_independent():
+    t = ForeignClickTracker(limit=2)
+    assert t.register("a") == 1
+    assert t.register("b") == 1
+    assert t.register("a") == 2
+
+
+def test_get_by_token():
+    async def scenario():
+        async def noop(record):
+            pass
+
+        state = VerifyState(Path(tempfile.mkdtemp()), on_timeout=noop)
+        rec = make_record(token="tok-xyz")
+        state.put(rec)
+        assert state.get_by_token("tok-xyz").key == rec.key
+        assert state.get_by_token("nope") is None
+        state.pop(rec.key)
+
+    asyncio.run(scenario())

tests/test_wiring.py

@@ -0,0 +1,67 @@
+"""处理器装配与配置装配测试（不发起任何网络请求）。"""
+
+from qqbotsdk import Bot
+
+from authbot.config import AuthConfig
+from authbot.handlers import KICK_NOT_ALLOWED, setup
+
+
+def _bot() -> Bot:
+    import os
+
+    os.environ.setdefault("APPID", "test-appid")
+    os.environ.setdefault("APPSECRET", "test-secret")
+    return Bot(intents=0)  # 掩码直传，避免读 intents.toml
+
+
+def test_config_defaults(monkeypatch, tmp_path):
+    monkeypatch.delenv("VERIFY_TIMEOUT_MINUTES", raising=False)
+    monkeypatch.delenv("TIMEOUT_KICK", raising=False)
+    monkeypatch.delenv("DATA_DIR", raising=False)
+    cfg = AuthConfig.load()
+    assert cfg.verify_timeout_minutes == 10
+    assert cfg.timeout_kick is True
+    assert cfg.timeout_fallback_mute_hours == 72
+    assert cfg.captcha_options == 4
+    assert cfg.verified_trust_days == 0
+
+
+def test_config_env_override(monkeypatch, tmp_path):
+    monkeypatch.setenv("VERIFY_TIMEOUT_MINUTES", "5")
+    monkeypatch.setenv("TIMEOUT_KICK", "false")
+    monkeypatch.setenv("DATA_DIR", str(tmp_path))
+    cfg = AuthConfig.load()
+    assert cfg.verify_timeout_minutes == 5
+    assert cfg.timeout_kick is False
+    assert cfg.data_dir == tmp_path
+
+
+def test_config_rejects_out_of_range(monkeypatch):
+    monkeypatch.setenv("VERIFY_TIMEOUT_MINUTES", "0")
+    try:
+        AuthConfig.load()
+    except ValueError as e:
+        assert "VERIFY_TIMEOUT_MINUTES" in str(e)
+    else:
+        raise AssertionError("应当拒绝越界值")
+
+
+def test_setup_registers_all_handlers(tmp_path, monkeypatch):
+    monkeypatch.setenv("DATA_DIR", str(tmp_path))
+    bot = _bot()
+    cfg = AuthConfig.load()
+    state = setup(bot, cfg)
+    registered = {
+        event
+        for event, handlers in bot._emitter._handlers.items()  # noqa: SLF001
+        if handlers
+    }
+    assert {
+        "READY",
+        "GROUP_MEMBER_ADD",
+        "INTERACTION_CREATE",
+        "GROUP_MEMBER_REMOVE",
+        "GROUP_AT_MESSAGE_CREATE",
+    } <= registered
+    assert state.all_pending() == []
+    assert KICK_NOT_ALLOWED == 11253