"""BF Bot 绑定库 (repository.py) 单测：建库幂等 / 列类型比较 / UserManager 回环。 repository.py 顶层只依赖标准库 + 同包 paths.py，用 importlib 造一次性包骨架 （与 test_mc_servers.py 同法）加载，避免触发插件包 __init__ 的 NoneBot 初始化。 **所有用例都显式传 tmp_path 下的库文件**，绝不碰 hexi/data 下的真实绑定库。 """ import importlib.util import sqlite3 import sys import types from pathlib import Path import pytest _PLUGIN_DIR = ( Path(__file__).resolve().parents[1] / "hexi" / "plugins" / "nonebot_plugin_bf_bot" ) _PKG = "bf_repo_testpkg" def _load_by_path(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module @pytest.fixture def repo(tmp_path): """一次性包骨架 + 全新 repository 模块 + tmp_path 下的空库文件。""" pkg = types.ModuleType(_PKG) pkg.__path__ = [str(_PLUGIN_DIR)] # type: ignore[attr-defined] sys.modules[_PKG] = pkg _load_by_path(f"{_PKG}.paths", _PLUGIN_DIR / "paths.py") module = _load_by_path(f"{_PKG}.repository", _PLUGIN_DIR / "repository.py") return module, str(tmp_path / "user_data.db") def test_paths_defaults_point_into_hexi_data(tmp_path, repo): """默认库路径/头像/缓存必须都在 hexi/data 下，不再落插件源码目录。""" module, _ = repo paths = sys.modules[f"{_PKG}.paths"] hexi_root = _PLUGIN_DIR.parents[1] assert paths.DATA_DIR == hexi_root / "data" / "nonebot_plugin_bf_bot" assert paths.DB_PATH.parent == paths.DATA_DIR assert paths.RES_DIR == _PLUGIN_DIR / "res" assert module.TableManager().db_path == str(paths.DB_PATH) def test_initialize_database_idempotent(repo): module, db_path = repo manager = module.DatabaseManager(db_path=db_path) assert manager.initialize_database() is True assert manager.initialize_database() is True # 第二次建表仍是幂等成功 conn = sqlite3.connect(db_path) tables = { row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") } # 四张业务表齐全；sqlite_sequence 属内部表，完整性校验会忽略 assert { "users", "designation", "medals", "query_record", } <= tables assert manager._verify_database_integrity(conn) is True conn.close() def test_initialize_database_failure_keeps_file(repo, tmp_path): """建库失败只返回 False，不得删库/抛异常（删库会连带丢绑定）。""" module, _ = repo bad_path = str(tmp_path / "不存在的目录" / "x.db") # 父目录不存在，打不开 assert module.DatabaseManager(db_path=bad_path).initialize_database() is False def test_type_head(repo): module, _ = repo type_head = module.DatabaseManager._type_head assert type_head("INTEGER PRIMARY KEY AUTOINCREMENT") == "INTEGER" assert type_head("text") == "TEXT" assert type_head("") == "" assert type_head(None) == "" def test_compare_columns(repo): module, _ = repo manager = module.DatabaseManager(db_path=":memory:") # PRAGMA 返回值不带约束，只比类型首词 assert manager._compare_columns( [("id", "INTEGER"), ("qq_id", "TEXT")], [("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("qq_id", "TEXT NOT NULL")], ) # 列名不同 / 列数不同都要判 False assert not manager._compare_columns([("id", "INTEGER")], [("idx", "INTEGER")]) assert not manager._compare_columns( [("id", "INTEGER")], [("id", "INTEGER"), ("qq_id", "TEXT")] ) # 类型首词不同也判 False assert not manager._compare_columns([("id", "INTEGER")], [("id", "TEXT")]) def test_user_roundtrip(repo): module, db_path = repo module.DatabaseManager(db_path=db_path).initialize_database() manager = module.UserManager(db_path=db_path) assert manager.get_user_by_qq("10001") is None row_id = manager.add_user("10001", "PlayerA", "pid-1", "oid-1", "[]", "[]") assert row_id > 0 record = manager.get_user_by_qq("10001") assert record is not None assert record["ea_player_name"] == "PlayerA" assert record["ea_player_id"] == "pid-1" assert manager.get_user_by_ea_id("oid-1")["qq_id"] == "10001" # 换绑 assert manager.update_user("10001", "PlayerB", "pid-2", "oid-2", "[]", "[]") == 1 assert manager.get_user_by_qq("10001")["ea_player_name"] == "PlayerB" assert manager.get_user_by_ea_id("oid-1") is None assert manager.get_user_by_ea_id("oid-2")["qq_id"] == "10001" # 解绑：删到 0 行；重复删除返回 0 assert manager.delete_user_by_qq("10001") == 1 assert manager.get_user_by_qq("10001") is None assert manager.delete_user_by_qq("10001") == 0 def test_user_ea_unique_constraint(repo): """EA UNIQUE 是最终的并发兜底；绑定入口需先 get_user_by_ea_id 预检。""" module, db_path = repo module.DatabaseManager(db_path=db_path).initialize_database() manager = module.UserManager(db_path=db_path) manager.add_user("10001", "PlayerA", "pid-1", "oid-1", "[]", "[]") with pytest.raises(sqlite3.IntegrityError): manager.add_user("10002", "PlayerA", "pid-9", "oid-1", "[]", "[]") def test_list_users_newest_first(repo): module, db_path = repo module.DatabaseManager(db_path=db_path).initialize_database() manager = module.UserManager(db_path=db_path) manager.add_user("10001", "PlayerA", "pid-1", "oid-1", "[]", "[]") manager.add_user("10002", "PlayerB", "pid-2", "oid-2", "[]", "[]") users = manager.list_users() assert [u["qq_id"] for u in users] == ["10002", "10001"] assert manager.count_users() == 2 # Web 管理页需要的字段都在 for key in ( "qq_id", "ea_player_name", "ea_player_id", "ea_user_id", "designation", "create_time", ): assert key in users[0] def test_list_users_pagination(repo): """Web 页分页：LIMIT/OFFSET 翻页不重不漏，顺序与全量一致（时间倒序 + id 兜底）。""" module, db_path = repo module.DatabaseManager(db_path=db_path).initialize_database() manager = module.UserManager(db_path=db_path) for i in range(7): manager.add_user(f"1000{i}", f"Player{i}", f"pid-{i}", f"oid-{i}", "[]", "[]") assert manager.count_users() == 7 assert [u["qq_id"] for u in manager.list_users(limit=3, offset=0)] == [ "10006", "10005", "10004", ] assert [u["qq_id"] for u in manager.list_users(limit=3, offset=3)] == [ "10003", "10002", "10001", ] assert [u["qq_id"] for u in manager.list_users(limit=3, offset=6)] == ["10000"] # 越界页返回空；不带 limit 仍是全量 assert manager.list_users(limit=3, offset=9) == [] assert len(manager.list_users()) == 7