"""群分析本地消息源 (history_store) 单元测试 history_store 只依赖标准库,故用 importlib 按文件路径裸加载 (与 test_rate_limit.py 同法),避免触发插件包 __init__ 的 NoneBot 初始化。 """ import importlib.util import json import sqlite3 import sys from pathlib import Path import pytest _MODULE_PATH = ( Path(__file__).resolve().parents[1] / "hexi" / "plugins" / "nonebot_plugin_group_daily_analysis" / "history_store.py" ) GROUP_ID = 872490448 BASE_TS = 1_800_000_000 @pytest.fixture(scope="module") def hs(): """以独立模块名加载 history_store.py,避免与插件包 __init__ 冲突""" spec = importlib.util.spec_from_file_location( "history_store_under_test", _MODULE_PATH ) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) yield module sys.modules.pop(spec.name, None) def _create_db(path: Path, *, with_uninfo: bool = True) -> sqlite3.Connection: """按真实列名建 learning_chat_message(可选带 uninfo 三表)""" con = sqlite3.connect(path) con.execute( """ CREATE TABLE learning_chat_message ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id BIGINT NOT NULL, user_id BIGINT NOT NULL, message_id BIGINT NOT NULL, message TEXT NOT NULL, raw_message TEXT NOT NULL, plain_text TEXT NOT NULL, time INTEGER NOT NULL ) """ ) con.execute( "CREATE INDEX ix_message_group_time" " ON learning_chat_message (group_id, time)" ) if with_uninfo: con.execute( "CREATE TABLE nonebot_plugin_uninfo_scenemodel" " (id INTEGER PRIMARY KEY, scene_id VARCHAR(64), scene_type INTEGER)" ) con.execute( "CREATE TABLE nonebot_plugin_uninfo_sessionmodel" " (id INTEGER PRIMARY KEY, scene_persist_id INTEGER," " user_persist_id INTEGER, member_data JSON)" ) con.execute( "CREATE TABLE nonebot_plugin_uninfo_usermodel" " (id INTEGER PRIMARY KEY, user_id VARCHAR(64), user_data JSON)" ) return con def _add_message( con: sqlite3.Connection, *, message_id: int, user_id: int = 1001, ts: int = BASE_TS, raw: str = "hi", plain_text: str | None = None, group_id: int = GROUP_ID, ) -> None: con.execute( "INSERT INTO learning_chat_message" " (group_id, user_id, message_id, message, raw_message, plain_text, time)" " VALUES (?, ?, ?, ?, ?, ?, ?)", ( group_id, user_id, message_id, raw, raw, raw if plain_text is None else plain_text, ts, ), ) def _add_member(con, user_id: str, nickname: str, card: str | None) -> None: """写入 uninfo 场景/用户/会话三条记录(群场景 scene_type=1)""" con.execute( "INSERT OR IGNORE INTO nonebot_plugin_uninfo_scenemodel" " (id, scene_id, scene_type) VALUES (1, ?, 1)", (str(GROUP_ID),), ) cur = con.execute( "INSERT INTO nonebot_plugin_uninfo_usermodel (user_id, user_data)" " VALUES (?, ?)", (user_id, json.dumps({"name": nickname, "nick": ""}, ensure_ascii=False)), ) con.execute( "INSERT INTO nonebot_plugin_uninfo_sessionmodel" " (scene_persist_id, user_persist_id, member_data) VALUES (1, ?, ?)", (cur.lastrowid, json.dumps({"nick": card or ""}, ensure_ascii=False)), ) def _build(path: Path, rows: list[dict], *, with_uninfo: bool = True) -> None: con = _create_db(path, with_uninfo=with_uninfo) for row in rows: _add_message(con, **row) con.commit() con.close() def test_cq_parsed_into_onebot_segments(hs, tmp_path): db = tmp_path / "cq.db" _build( db, [ { "message_id": 1, "raw": "[CQ:reply,id=5][CQ:at,qq=2]hi" "[CQ:image,file=a.jpg,url=http://x?a=1&b=2]", } ], ) res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 1, 100, db_path=db) assert res and res.error is None segs = res.messages[0]["message"] assert [s["type"] for s in segs] == ["reply", "at", "text", "image"] assert segs[0]["data"] == {"id": "5"} assert segs[1]["data"] == {"qq": "2"} assert segs[2]["data"] == {"text": "hi"} assert segs[3]["data"]["url"] == "http://x?a=1&b=2" # CQ 反转义 # 交给 adapter._convert_message 的字段形状 assert res.messages[0]["sender"]["user_id"] == "1001" def test_window_is_inclusive(hs, tmp_path): db = tmp_path / "window.db" _build( db, [ {"message_id": 1, "ts": BASE_TS - 1}, {"message_id": 2, "ts": BASE_TS}, {"message_id": 3, "ts": BASE_TS + 1}, ], ) res = hs.fetch_group_messages(GROUP_ID, BASE_TS, BASE_TS, 100, db_path=db) # 两端闭区间:起始时间戳上的消息必须被取到(增量水位依赖这一点) assert [m["message_id"] for m in res.messages] == [2] def test_same_second_sorted_by_id_ascending(hs, tmp_path): db = tmp_path / "order.db" _build( db, [ {"message_id": 11, "ts": BASE_TS}, {"message_id": 12, "ts": BASE_TS}, {"message_id": 13, "ts": BASE_TS}, ], ) res = hs.fetch_group_messages(GROUP_ID, BASE_TS, BASE_TS, 100, db_path=db) assert [m["message_id"] for m in res.messages] == [11, 12, 13] def test_duplicate_message_id_deduped(hs, tmp_path): db = tmp_path / "dup.db" _build(db, [{"message_id": 7}, {"message_id": 7}]) res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 1, 100, db_path=db) assert len(res.messages) == 1 assert res.duplicates == 1 def test_limit_truncates_and_keeps_newest(hs, tmp_path): db = tmp_path / "trunc.db" _build( db, [ {"message_id": 1, "ts": BASE_TS}, {"message_id": 2, "ts": BASE_TS + 1}, {"message_id": 3, "ts": BASE_TS + 2}, ], ) res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 5, 2, db_path=db) assert res.truncated is True assert res.window_total == 3 assert [m["message_id"] for m in res.messages] == [2, 3] # 丢最旧的 def test_names_resolved_from_uninfo(hs, tmp_path): db = tmp_path / "names.db" con = _create_db(db) _add_member(con, "1001", "昵称甲", "群名片甲") _add_member(con, "1002", "昵称乙", "") _add_message(con, message_id=1, user_id=1001) _add_message(con, message_id=2, user_id=1002) _add_message(con, message_id=3, user_id=1003) # 未在 uninfo 中 con.commit() con.close() res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 1, 100, db_path=db) senders = {m["message_id"]: m["sender"] for m in res.messages} assert senders[1]["nickname"] == "昵称甲" assert senders[1]["card"] == "群名片甲" assert senders[2]["nickname"] == "昵称乙" assert senders[2]["card"] == "" # 空群名片交给 _convert_message 归一为 None assert senders[3]["nickname"] == "" and senders[3]["card"] == "" assert res.names_resolved == 2 # 群名片空串在源数据里就归一为 None con = hs._open_readonly(db) try: assert hs._load_display_names(con, GROUP_ID)["1002"] == ("昵称乙", None) finally: con.close() def test_missing_db_reports_error_without_creating(hs, tmp_path): missing = tmp_path / "nope.sqlite3" res = hs.fetch_group_messages( GROUP_ID, BASE_TS - 1, BASE_TS + 1, 10, db_path=missing ) assert not res assert res.error and "不存在" in res.error assert not missing.exists() # mode=ro 不得建库 def test_missing_table_is_not_fatal(hs, tmp_path): db = tmp_path / "empty.db" con = sqlite3.connect(db) con.execute("CREATE TABLE unrelated (x INTEGER)") con.commit() con.close() res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 1, 10, db_path=db) assert not res assert res.error and "learning_chat_message" in res.error def test_other_group_filtered_out(hs, tmp_path): db = tmp_path / "other.db" _build( db, [ {"message_id": 1}, {"message_id": 2, "group_id": GROUP_ID + 1}, ], ) res = hs.fetch_group_messages(GROUP_ID, BASE_TS - 1, BASE_TS + 1, 100, db_path=db) assert [m["message_id"] for m in res.messages] == [1]