321 lines
16 KiB
Python
321 lines
16 KiB
Python
"""成员绑定、持久记录和插件共用接口验证，不连接微信。"""
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from plugins.member_identity import Plugin
|
|
from wxbot_framework.members import MemberError, MemberStore
|
|
from wxbot_framework.sdk import Message, PluginContext, ServiceFacade
|
|
from wxbot_framework.runtime import BotApplication
|
|
from wxbot_framework.config import ConfigStore
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCRATCH = ROOT / ".runtime/tests"
|
|
ROOM = "12345678@chatroom"
|
|
ROBOT, SMALL, JING = "wxid_robot", "wxid_small", "wxid_jing"
|
|
|
|
|
|
class MemberTests(unittest.TestCase):
|
|
def setUp(self):
|
|
SCRATCH.mkdir(parents=True, exist_ok=True)
|
|
self.temporary = tempfile.TemporaryDirectory(dir=SCRATCH)
|
|
self.root = Path(self.temporary.name).resolve()
|
|
self.assertTrue(self.root.is_relative_to(SCRATCH.resolve()))
|
|
self.options = {"database_file": "data.sqlite3", "history_enabled": True,
|
|
"groups": {ROOM: [
|
|
{"key": "xiaomei", "name": "成员甲", "display_name": "成员甲昵称", "aliases": ["小号"], "wxid": SMALL},
|
|
{"key": "jing", "name": "成员乙", "display_name": "成员乙群名", "aliases": ["成员乙别名"], "wxid": None},
|
|
{"key": "yuting", "name": "成员丁", "display_name": "DemoMemberC", "wxid": None}]}}
|
|
self.config_path = self.root / "members.json"
|
|
self.write_options()
|
|
self.stores = []
|
|
self.store = self.create_store()
|
|
|
|
def write_options(self):
|
|
self.config_path.write_text(json.dumps(self.options, ensure_ascii=False), encoding="utf-8")
|
|
|
|
def create_store(self, account=ROBOT):
|
|
store = MemberStore(self.config_path, account)
|
|
self.stores.append(store)
|
|
return store
|
|
|
|
def tearDown(self):
|
|
for store in self.stores:
|
|
store.close()
|
|
self.assertTrue(self.root.is_relative_to(SCRATCH.resolve()))
|
|
self.temporary.cleanup()
|
|
|
|
def message(self, text="中文消息 🌸", sender=JING, sequence=1, **kwargs):
|
|
return Message(text, ROOM, sender_id=sender, is_group=True, sequence=sequence, session_id="first", **kwargs)
|
|
|
|
def test_configuration_check_does_not_create_database(self):
|
|
self.assertFalse(self.store.path.exists())
|
|
self.store.initialize()
|
|
self.assertTrue(self.store.path.exists())
|
|
|
|
def test_aliases_are_resolved_case_insensitively_and_small_is_bound(self):
|
|
self.assertEqual(self.store.resolve("小号").wxid, SMALL)
|
|
self.assertEqual(self.store.resolve("成员乙群名", ROOM).name, "成员乙")
|
|
self.assertEqual(self.store.resolve("demomemberc", ROOM).name, "成员丁")
|
|
self.assertIsNone(self.store.resolve("成员乙").wxid)
|
|
|
|
def test_unknown_names_are_not_guessed(self):
|
|
with self.assertRaises(MemberError):
|
|
self.store.resolve("未登记的昵称")
|
|
with self.assertRaises(MemberError):
|
|
self.store.bind("未登记的昵称", ROOM, JING)
|
|
|
|
def test_binding_survives_restart_and_null_profile_does_not_clear_it(self):
|
|
self.store.bind("成员乙群名", ROOM, JING)
|
|
self.store.close()
|
|
reopened = self.create_store()
|
|
self.assertEqual(reopened.resolve("成员乙", ROOM).wxid, JING)
|
|
|
|
def test_conflicting_binding_does_not_replace_identity(self):
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
self.assertEqual(self.store.bind("成员乙", ROOM, JING).wxid, JING)
|
|
with self.assertRaises(MemberError):
|
|
self.store.bind("成员乙", ROOM, "wxid_other")
|
|
with self.assertRaises(MemberError):
|
|
self.store.bind("成员丁", ROOM, JING)
|
|
self.assertEqual(self.store.resolve("成员乙", ROOM).wxid, JING)
|
|
|
|
def test_history_before_binding_becomes_queryable_by_alias(self):
|
|
self.store.record(self.message())
|
|
with self.assertRaises(MemberError):
|
|
self.store.history("成员乙")
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
rows = self.store.history("成员乙别名")
|
|
self.assertEqual([row["text"] for row in rows], ["中文消息 🌸"])
|
|
self.assertEqual(rows[0]["sender_id"], JING)
|
|
self.assertEqual(datetime.fromisoformat(rows[0]["received_at"]).utcoffset(), timedelta(0))
|
|
|
|
def test_history_is_isolated_by_robot_account_and_author(self):
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
self.store.record(self.message("成员乙的话"))
|
|
self.store.record(self.message("成员甲的话", sender=SMALL, sequence=2))
|
|
second = self.create_store("wxid_second_robot")
|
|
second.bind("成员乙", ROOM, JING)
|
|
self.assertEqual(second.history("成员乙"), [])
|
|
self.assertEqual([row["text"] for row in self.store.history("成员乙")], ["成员乙的话"])
|
|
|
|
def test_private_other_group_and_wrong_account_are_not_archived(self):
|
|
self.assertFalse(self.store.record(Message("私聊", JING, sender_id=JING, session_id="first")))
|
|
self.assertFalse(self.store.record(Message("另一群", "99999@chatroom", sender_id=JING, is_group=True, session_id="first")))
|
|
self.assertFalse(self.store.record(self.message(data={"self_wxid": "wxid_other_robot"})))
|
|
|
|
def test_sequence_deduplication_and_new_session(self):
|
|
self.assertTrue(self.store.record(self.message()))
|
|
self.assertFalse(self.store.record(self.message()))
|
|
self.assertTrue(self.store.record(Message("重连消息", ROOM, JING, is_group=True, sequence=1, session_id="second")))
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
self.assertEqual(len(self.store.history("成员乙")), 2)
|
|
|
|
def test_limit_and_time_filter_return_chronological_records(self):
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
self.store.record(self.message("第一句", sequence=1))
|
|
since = datetime.now(timezone.utc)
|
|
self.store.record(self.message("第二句", sequence=2))
|
|
self.assertEqual([r["text"] for r in self.store.history("成员乙", limit=1)], ["第二句"])
|
|
self.assertEqual([r["text"] for r in self.store.history("成员乙", since=since)], ["第二句"])
|
|
self.assertEqual([r["text"] for r in self.store.history("成员乙")], ["第一句", "第二句"])
|
|
for limit in (True, 0, 1001):
|
|
with self.assertRaises(MemberError):
|
|
self.store.history("成员乙", limit=limit)
|
|
with self.assertRaises(MemberError):
|
|
self.store.history("成员乙", since=datetime.now())
|
|
|
|
def test_disabled_history_keeps_member_directory(self):
|
|
self.options["history_enabled"] = False
|
|
self.write_options()
|
|
disabled = self.create_store()
|
|
self.assertEqual(disabled.resolve("成员甲").wxid, SMALL)
|
|
self.assertFalse(disabled.record(self.message()))
|
|
|
|
def test_same_alias_in_two_groups_requires_group(self):
|
|
second_room = "99999@chatroom"
|
|
self.options["groups"][second_room] = self.options["groups"][ROOM]
|
|
self.write_options()
|
|
multi = self.create_store()
|
|
with self.assertRaises(MemberError):
|
|
multi.resolve("成员乙")
|
|
self.assertEqual(multi.resolve("成员乙", second_room).room, second_room)
|
|
|
|
def test_configuration_rejects_duplicate_casefolded_aliases(self):
|
|
self.options["groups"][ROOM][1]["aliases"] = ["demomemberc"]
|
|
self.write_options()
|
|
with self.assertRaises(MemberError):
|
|
self.create_store()
|
|
|
|
def test_recording_and_lookup_from_different_threads(self):
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
errors = []
|
|
def write(index):
|
|
try:
|
|
self.store.record(self.message(str(index), sequence=index))
|
|
self.store.resolve("成员乙")
|
|
except Exception as error:
|
|
errors.append(error)
|
|
threads = [threading.Thread(target=write, args=(index,)) for index in range(1, 13)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(3)
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(len(self.store.history("成员乙")), 12)
|
|
|
|
def test_identity_plugin_binds_actual_message_author(self):
|
|
ctx = Mock(members=self.store, logger=logging.getLogger("MembersTest"))
|
|
plugin = Plugin()
|
|
plugin.on_load(ctx)
|
|
plugin.on_message(self.message("/绑定 成员乙"), ctx)
|
|
self.assertEqual(self.store.resolve("成员乙").wxid, JING)
|
|
ctx.reply.assert_called_once_with(unittest.mock.ANY, "已绑定：成员乙。")
|
|
plugin.on_message(self.message("/绑定 成员丁"), ctx)
|
|
self.assertIsNone(self.store.resolve("成员丁").wxid)
|
|
|
|
def test_context_auto_mention_uses_group_name_instead_of_personal_alias(self):
|
|
self.store.bind("成员乙", ROOM, JING)
|
|
backend = Mock(session_id="first")
|
|
service = ServiceFacade(text_delay={"enabled": False})
|
|
service.bind(backend)
|
|
try:
|
|
ctx = PluginContext("example", {}, service, [], member_store=self.store)
|
|
self.assertEqual(ctx.members.resolve("成员乙").wxid, JING)
|
|
ctx.reply_at(self.message(), "回复")
|
|
backend.send_at_text.assert_called_once_with(ROOM, "@成员乙群名\u2005\n回复", (JING,))
|
|
finally:
|
|
service.close()
|
|
|
|
def test_runtime_archives_before_plugin_dispatch(self):
|
|
(self.root / "plugins").mkdir()
|
|
path = self.root / "bot_config.json"
|
|
path.write_text(json.dumps({"backend": "native_text", "wechat_account": ROBOT,
|
|
"members_config_file": self.config_path.name}), encoding="utf-8")
|
|
app = BotApplication(self.root, ConfigStore(path))
|
|
try:
|
|
app.members.bind("成员乙", ROOM, JING)
|
|
def verify(message):
|
|
self.assertEqual(app.members.history("成员乙")[0]["text"], message.text)
|
|
with patch.object(app.plugins, "dispatch", side_effect=verify):
|
|
app._emit(self.message())
|
|
finally:
|
|
app.stop()
|
|
|
|
def test_free_registration_accepts_new_name_and_links_existing_history(self):
|
|
self.store.record(self.message("绑定前说过的话"))
|
|
member = self.store.register("测试成员", ROOM, JING)
|
|
self.assertEqual(member.wxid, JING)
|
|
self.assertEqual(self.store.history("测试成员")[0]["text"], "绑定前说过的话")
|
|
|
|
def test_same_account_can_rename_without_losing_aliases_or_history(self):
|
|
self.store.register("测试成员", ROOM, JING)
|
|
self.store.record(self.message("原有记录"))
|
|
self.store.register("成员丙", ROOM, JING)
|
|
self.assertEqual(self.store.resolve("测试成员").name, "成员丙")
|
|
self.assertEqual(self.store.resolve("成员丙").wxid, JING)
|
|
self.assertEqual(self.store.resolve("成员丙").display_name, "成员丙")
|
|
self.assertEqual(self.store.history("成员丙")[0]["text"], "原有记录")
|
|
self.assertEqual(len([m for m in self.store.list_members() if m.wxid == JING]), 1)
|
|
|
|
def test_another_account_cannot_take_existing_nickname(self):
|
|
self.store.register("测试成员", ROOM, JING)
|
|
with self.assertRaises(MemberError):
|
|
self.store.register("测试成员", ROOM, SMALL)
|
|
self.assertEqual(self.store.resolve("测试成员").wxid, JING)
|
|
|
|
def test_registered_name_survives_config_seeding_on_restart(self):
|
|
self.store.register("成员乙", ROOM, JING)
|
|
self.store.register("测试成员", ROOM, JING)
|
|
self.store.close()
|
|
reopened = self.create_store()
|
|
self.assertEqual(reopened.resolve("成员乙").name, "测试成员")
|
|
self.assertEqual(reopened.resolve("测试成员").wxid, JING)
|
|
|
|
def test_free_registration_still_limits_groups_and_name_shape(self):
|
|
with self.assertRaises(MemberError):
|
|
self.store.register("测试成员", "99999@chatroom", JING)
|
|
for name in ("", "x" * 65, "喵\n喵", "喵\0喵"):
|
|
with self.subTest(name=name), self.assertRaises(MemberError):
|
|
self.store.register(name, ROOM, JING)
|
|
|
|
def test_group_binding_overrides_legacy_display_and_survives_restart(self):
|
|
self.store.register("成员甲", ROOM, SMALL)
|
|
self.assertEqual(self.store.display_name(SMALL, ROOM), "成员甲")
|
|
self.store.close()
|
|
self.store = self.create_store()
|
|
self.assertEqual(self.store.display_name(SMALL, ROOM), "成员甲")
|
|
self.assertTrue(self.store.resolve("成员甲", ROOM).self_named)
|
|
|
|
def test_wechat_default_and_bindings_are_separate_in_each_group(self):
|
|
other_room = "87654321@chatroom"
|
|
self.options["groups"][other_room] = []
|
|
self.write_options()
|
|
store = self.create_store()
|
|
store.register("成员甲", ROOM, SMALL)
|
|
store.remember_wechat_name(SMALL, "DemoUser")
|
|
message = Message("新群的原话", other_room, SMALL, is_group=True, session_id="new", sequence=1)
|
|
store.record(message)
|
|
self.assertEqual(store.display_name(SMALL, other_room), "DemoUser")
|
|
self.assertEqual(store.list_members(other_room), [])
|
|
self.assertEqual(store.today_snapshot("DemoUser", other_room).member.wxid, SMALL)
|
|
backend = Mock(session_id="new")
|
|
service = ServiceFacade(text_delay={"enabled": False})
|
|
service.bind(backend)
|
|
try:
|
|
ctx = PluginContext("test", {}, service, [], member_store=store)
|
|
ctx.reply_at(message, "默认回复")
|
|
self.assertEqual(backend.send_at_text.call_args.args[1], "@DemoUser\u2005\n默认回复")
|
|
store.register("新群名字", other_room, SMALL)
|
|
ctx.reply_at(message, "已绑定", display_name="旧的硬编码名字")
|
|
self.assertEqual(backend.send_at_text.call_args.args[1], "@新群名字\u2005\n已绑定")
|
|
self.assertEqual(store.display_name(SMALL, ROOM), "成员甲")
|
|
self.assertEqual(store.today_snapshot("新群名字", other_room).messages[0]["text"], "新群的原话")
|
|
finally:
|
|
service.close()
|
|
|
|
def test_native_wechat_name_updates_default_without_overwriting_group_binding(self):
|
|
other_room = "87654321@chatroom"
|
|
self.options["groups"][other_room] = []
|
|
self.write_options()
|
|
store = self.create_store()
|
|
store.register("成员甲", ROOM, SMALL)
|
|
for sequence, name in ((1, "DemoUser"), (2, "新的微信名")):
|
|
msg = Message("原文", other_room, SMALL, is_group=True, session_id="new", sequence=sequence,
|
|
data={"sender_wechat_name": name, "self_wxid": ROBOT})
|
|
store.record(msg)
|
|
self.assertEqual(store.display_name(SMALL, other_room), name)
|
|
self.assertEqual(store.display_name(SMALL, ROOM), "成员甲")
|
|
self.assertEqual(store.today_group_snapshot(other_room).messages[-1]["speaker_name"], "新的微信名")
|
|
self.assertEqual(store.conversation_snapshot(msg).speaker_name, "新的微信名")
|
|
store.clear(other_room)
|
|
self.assertEqual(store.today_snapshot("新的微信名", other_room).messages, ())
|
|
|
|
def test_duplicate_default_wechat_names_do_not_mix_member_identities(self):
|
|
self.options["groups"][ROOM] = []
|
|
self.write_options()
|
|
store = self.create_store()
|
|
for index, wxid in enumerate((SMALL, JING), 1):
|
|
store.record(self.message("不同成员", sender=wxid, sequence=index,
|
|
data={"sender_wechat_name": "同名"}))
|
|
with self.assertRaisesRegex(MemberError, "多个成员"):
|
|
store.today_snapshot("同名", ROOM)
|
|
store.register("指定成员", ROOM, SMALL)
|
|
self.assertEqual(store.today_snapshot("指定成员", ROOM).member.wxid, SMALL)
|
|
|
|
def test_plugin_free_registration_and_usage(self):
|
|
ctx = Mock(members=self.store, logger=logging.getLogger("MembersTest"))
|
|
plugin = Plugin()
|
|
plugin.on_load(ctx)
|
|
plugin.on_message(self.message("/绑定 测试成员"), ctx)
|
|
self.assertEqual(self.store.resolve("测试成员").wxid, JING)
|
|
plugin.on_message(self.message("/绑定"), ctx)
|
|
self.assertEqual(ctx.reply.call_args.args[1], "用法：/绑定 昵称")
|
|
plugin.on_message(self.message(" /绑定\t成员丙 "), ctx)
|
|
self.assertEqual(self.store.resolve("成员丙").wxid, JING)
|