sansenhoshiandClaude Code
814c1994b6
- services/servers.py：校验 + 读写 + 增删改收敛成一份数据层（顶层不 import nonebot， 可脱离运行时单测），群命令与 Web API 都只经它改数据，校验不再各写一套 - web_hub.py + /hub/mc_server_status 页面：按群列出服务器，抽屉新增/编辑（改名）/ 测试连接，行内删除（删空一台连带删群）；每条路由自带 require_admin - handlers/server.py 改为薄封装，只在需要命令特有提示时自己判断 - services/mc.py 的图标属性名兼容 mcstatus ≥13 的 icon（旧版叫 favicon， 读错会把整条查询打成"查询失败"） - tests/test_mc_servers.py：数据层归一/改名/删群/坏文件兜底 + 图标解码两版属性名 Co-Authored-By: Claude Code <noreply@anthropic.com>
341 lines
14 KiB
Python
341 lines
14 KiB
Python
"""MC 服务器数据层 (services/servers.py) 单元测试
|
||
|
||
servers.py 顶层不 import nonebot（运行态 config 是运行时才取），但 config.py
|
||
里有 `from .services import ...` 这类相对导入，所以这里**按包结构**用 importlib
|
||
加载（而不是像 test_video_policy.py 那样裸加载单文件）：造一棵一次性的包骨架
|
||
塞进 sys.modules，真正的插件包 `nonebot_plugin_mc_server_status` 全程没被
|
||
NoneBot 加载过。
|
||
|
||
**每个用例重建一份 servers/config 模块**：增删改落在 `config.var.group_list`
|
||
上，直接复用进程里已有的 config 会把正在跑的 bot 的内存态改写掉（测试污染运行
|
||
数据），所以每次都用全新的模块对象 + tmp_path 下的数据文件。
|
||
"""
|
||
|
||
import base64
|
||
import importlib.util
|
||
import json
|
||
import sys
|
||
import types
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
_PLUGIN_DIR = (
|
||
Path(__file__).resolve().parents[1]
|
||
/ "hexi"
|
||
/ "plugins"
|
||
/ "nonebot_plugin_mc_server_status"
|
||
)
|
||
pytestmark = pytest.mark.skipif(
|
||
not (_PLUGIN_DIR / "config.py").exists(), reason="MC 插件不存在"
|
||
)
|
||
|
||
_PKG = "mc_status_testpkg"
|
||
_SERVERS = f"{_PKG}.services.servers"
|
||
_CONFIG = f"{_PKG}.config"
|
||
|
||
|
||
def _load_by_path(name: str, path: Path):
|
||
spec = importlib.util.spec_from_file_location(
|
||
name, path, submodule_search_locations=[] if path.is_dir() else None
|
||
)
|
||
module = importlib.util.module_from_spec(spec)
|
||
sys.modules[name] = module
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
class _StubDriver:
|
||
"""config.py 的 @driver.on_startup / on_bot_connect 只登记回调，测试里不触发。"""
|
||
|
||
def __getattr__(self, _name):
|
||
def _decorator(*_args, **_kwargs):
|
||
return lambda func: func
|
||
|
||
return _decorator
|
||
|
||
|
||
def _init_nonebot_stubs():
|
||
"""config.py 顶层要 get_driver()/get_plugin_config()/nonebot.log，没有运行时就得打桩。
|
||
|
||
先真 `import nonebot` 再把这几个入口换掉（而不是塞一个假包进 sys.modules）：
|
||
假包会让 python 用它的 `__path__` 去找 `nonebot.adapters`，反而找不到真子模块。
|
||
"""
|
||
import nonebot
|
||
|
||
pkg = types.ModuleType("nonebot.log")
|
||
pkg.logger = types.SimpleNamespace( # type: ignore[attr-defined]
|
||
debug=lambda *a, **k: None,
|
||
info=lambda *a, **k: None,
|
||
warning=lambda *a, **k: None,
|
||
error=lambda *a, **k: None,
|
||
)
|
||
saved = {
|
||
"get_driver": nonebot.get_driver, # type: ignore[attr-defined]
|
||
"get_plugin_config": nonebot.get_plugin_config, # type: ignore[attr-defined]
|
||
"get_bot": nonebot.get_bot, # type: ignore[attr-defined]
|
||
"get_bots": nonebot.get_bots, # type: ignore[attr-defined]
|
||
"log": sys.modules.get("nonebot.log"),
|
||
}
|
||
nonebot.get_driver = lambda: _StubDriver() # type: ignore[attr-defined]
|
||
nonebot.get_plugin_config = lambda *_a, **_k: types.SimpleNamespace( # type: ignore[attr-defined]
|
||
mc_status_data_filename="mc_status_data.json"
|
||
)
|
||
nonebot.get_bot = lambda *_a, **_k: None # type: ignore[attr-defined]
|
||
nonebot.get_bots = lambda: {} # type: ignore[attr-defined]
|
||
sys.modules["nonebot.log"] = pkg
|
||
|
||
def restore() -> None:
|
||
nonebot.get_driver = saved["get_driver"] # type: ignore[attr-defined]
|
||
nonebot.get_plugin_config = saved["get_plugin_config"] # type: ignore[attr-defined]
|
||
nonebot.get_bot = saved["get_bot"] # type: ignore[attr-defined]
|
||
nonebot.get_bots = saved["get_bots"] # type: ignore[attr-defined]
|
||
if saved["log"] is None:
|
||
sys.modules.pop("nonebot.log", None)
|
||
else:
|
||
sys.modules["nonebot.log"] = saved["log"]
|
||
|
||
return restore
|
||
|
||
|
||
@pytest.fixture
|
||
def mcs(tmp_path):
|
||
"""一套干净的数据层：一次性包骨架 + 全新 servers/config 模块 + 临时数据文件。"""
|
||
pkg = types.ModuleType(_PKG)
|
||
pkg.__path__ = [str(_PLUGIN_DIR)] # type: ignore[attr-defined]
|
||
sys.modules[_PKG] = pkg
|
||
|
||
services = types.ModuleType(f"{_PKG}.services")
|
||
services.__path__ = [str(_PLUGIN_DIR / "services")] # type: ignore[attr-defined]
|
||
sys.modules[f"{_PKG}.services"] = services
|
||
|
||
restore = _init_nonebot_stubs()
|
||
try:
|
||
module = _load_by_path(_SERVERS, _PLUGIN_DIR / "services" / "servers.py")
|
||
runtime = _load_by_path(_CONFIG, _PLUGIN_DIR / "config.py")
|
||
finally:
|
||
restore()
|
||
|
||
# config.py 的 on_startup 不在测试里跑，内存态与数据文件名手动准备
|
||
runtime.var.group_list.clear()
|
||
runtime.pc.mc_status_data_filename = str(tmp_path / "mc_status_data.json")
|
||
try:
|
||
yield module
|
||
finally:
|
||
for name in (_SERVERS, _CONFIG, f"{_PKG}.services", _PKG):
|
||
sys.modules.pop(name, None)
|
||
|
||
|
||
@pytest.fixture
|
||
def groups(mcs):
|
||
return mcs._runtime().var.group_list
|
||
|
||
|
||
@pytest.fixture
|
||
def mc_module():
|
||
"""加载 services/mc.py（只测纯函数 _server_icon，不联网）。
|
||
|
||
mc.py 在模块顶层 import 了 OneBot 的消息类型，这里给个占位桩 —— 测试只关心
|
||
图标解码，不会真的构造 Message。
|
||
"""
|
||
pkg = types.ModuleType(_PKG)
|
||
pkg.__path__ = [str(_PLUGIN_DIR)] # type: ignore[attr-defined]
|
||
sys.modules[_PKG] = pkg
|
||
services = types.ModuleType(f"{_PKG}.services")
|
||
services.__path__ = [str(_PLUGIN_DIR / "services")] # type: ignore[attr-defined]
|
||
sys.modules[f"{_PKG}.services"] = services
|
||
|
||
adapter = types.ModuleType("nonebot.adapters.onebot.v11")
|
||
adapter.Message = type("Message", (), {})
|
||
adapter.MessageSegment = types.SimpleNamespace(image=lambda *_a, **_k: object())
|
||
saved = sys.modules.get("nonebot.adapters.onebot.v11")
|
||
sys.modules["nonebot.adapters.onebot.v11"] = adapter
|
||
try:
|
||
module = _load_by_path(
|
||
f"{_PKG}.services.mc", _PLUGIN_DIR / "services" / "mc.py"
|
||
)
|
||
finally:
|
||
if saved is None:
|
||
sys.modules.pop("nonebot.adapters.onebot.v11", None)
|
||
else:
|
||
sys.modules["nonebot.adapters.onebot.v11"] = saved
|
||
try:
|
||
yield module
|
||
finally:
|
||
for name in (f"{_PKG}.services.mc", f"{_PKG}.services", _PKG):
|
||
sys.modules.pop(name, None)
|
||
|
||
|
||
# ───────────────────────── 服务器图标（mcstatus 跨版本） ─────────────────────────
|
||
|
||
_PNG = b"\x89PNG\r\n\x1a\n" + b"fake-image-bytes"
|
||
|
||
|
||
def test_server_icon_accepts_new_and_old_attribute_names(mc_module):
|
||
"""mcstatus ≥13 是 icon，11/12 是 favicon —— 两种都要认（线上 13.0.1 踩过一次）。"""
|
||
b64 = base64.b64encode(_PNG).decode()
|
||
new_status = types.SimpleNamespace(icon=f"data:image/png;base64,{b64}")
|
||
old_status = types.SimpleNamespace(favicon=f"data:image/png;base64,{b64}")
|
||
assert mc_module._server_icon(new_status) == _PNG
|
||
assert mc_module._server_icon(old_status) == _PNG
|
||
|
||
|
||
def test_server_icon_handles_bare_base64_and_missing(mc_module):
|
||
b64 = base64.b64encode(_PNG).decode()
|
||
assert mc_module._server_icon(types.SimpleNamespace(icon=b64)) == _PNG # 无 data URI 前缀
|
||
assert mc_module._server_icon(types.SimpleNamespace(icon=None)) is None
|
||
assert mc_module._server_icon(types.SimpleNamespace()) is None # 两个属性都没有
|
||
assert mc_module._server_icon(types.SimpleNamespace(icon="不是 base64!!")) is None
|
||
|
||
|
||
def _file(mcs) -> dict:
|
||
return json.loads(Path(mcs._runtime().pc.mc_status_data_filename).read_text(encoding="utf-8"))
|
||
|
||
|
||
# ───────────────────────── 校验 / 归一化 ─────────────────────────
|
||
|
||
|
||
def test_normalize_type_aliases_and_errors(mcs):
|
||
assert mcs.normalize_type("js") == "js"
|
||
assert mcs.normalize_type("JS") == "js"
|
||
assert mcs.normalize_type(" Java ") == "js"
|
||
assert mcs.normalize_type("bds") == "bds"
|
||
assert mcs.normalize_type("bedrock") == "bds"
|
||
assert mcs.normalize_type("") == "js" # 空值取默认
|
||
assert mcs.normalize_type(None, "bds") == "bds"
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.normalize_type("java版")
|
||
|
||
|
||
def test_normalize_address_strips_scheme_and_rejects_junk(mcs):
|
||
assert mcs.normalize_address(" mc.example.com:25565 ") == "mc.example.com:25565"
|
||
assert mcs.normalize_address("minecraft://mc.example.com") == "mc.example.com"
|
||
assert mcs.normalize_address("http://mc.example.com/") == "mc.example.com"
|
||
for bad in ("", " ", "mc example.com", "mc.example.com/foo", ":25565", "host:"):
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.normalize_address(bad)
|
||
|
||
|
||
def test_normalize_name_keeps_spaces_and_strips_quotes(mcs):
|
||
assert mcs.normalize_name("亡者世界") == "亡者世界"
|
||
assert mcs.normalize_name(" 我 的 服 ") == "我 的 服"
|
||
assert mcs.normalize_name('"EMDS"') == "EMDS"
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.normalize_name(" ")
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.normalize_name("x" * (mcs.MAX_NAME_LEN + 2))
|
||
|
||
|
||
def test_parse_group_id(mcs):
|
||
assert mcs.parse_group_id("872490448") == 872490448
|
||
assert mcs.parse_group_id(304484502) == 304484502
|
||
for bad in ("", " ", "abc", "872490448.0", "0", 1234567890123):
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.parse_group_id(bad)
|
||
|
||
|
||
# ───────────────────────── 增 / 改 ─────────────────────────
|
||
|
||
|
||
def test_add_writes_file_and_memory(mcs, groups):
|
||
assert mcs.set_server(872490448, "EMDS", "sansenhoshi.top:29985", "js") is True
|
||
assert groups[872490448]["EMDS"] == ["sansenhoshi.top:29985", "js"]
|
||
assert _file(mcs) == {"872490448": {"EMDS": ["sansenhoshi.top:29985", "js"]}}
|
||
|
||
|
||
def test_update_same_name_keeps_position(mcs, groups):
|
||
mcs.set_server(872490448, "EMDS", "a.example.com:1", "js")
|
||
mcs.set_server(872490448, "星门", "b.example.com:2", "bds")
|
||
assert mcs.set_server(872490448, "EMDS", "c.example.com:3", "js") is False # 同名 = 更新
|
||
assert list(groups[872490448]) == ["EMDS", "星门"] # 顺序不变
|
||
assert groups[872490448]["EMDS"] == ["c.example.com:3", "js"]
|
||
|
||
|
||
def test_rename_moves_entry(mcs, groups):
|
||
mcs.set_server(872490448, "旧名", "mc.example.com:25565", "js")
|
||
assert mcs.set_server(872490448, "新名", "mc.example.com:25565", "js", old_name="旧名") is True
|
||
assert list(groups[872490448]) == ["新名"]
|
||
assert _file(mcs)["872490448"]["新名"] == ["mc.example.com:25565", "js"]
|
||
|
||
|
||
def test_rename_conflict_is_rejected_without_touching_data(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
mcs.set_server(872490448, "B", "b.example.com:2", "js")
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.set_server(872490448, "B", "c.example.com:3", "js", old_name="A")
|
||
assert groups[872490448] == {"A": ["a.example.com:1", "js"], "B": ["b.example.com:2", "js"]}
|
||
|
||
|
||
def test_rename_missing_source_is_rejected(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.set_server(872490448, "B", "b.example.com:2", "js", old_name="不存在")
|
||
assert list(groups[872490448]) == ["A"]
|
||
|
||
|
||
def test_invalid_input_does_not_write_file(mcs, groups, tmp_path):
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "不是类型")
|
||
assert groups == {}
|
||
assert not (tmp_path / "mc_status_data.json").exists()
|
||
|
||
|
||
# ───────────────────────── 删 ─────────────────────────
|
||
|
||
|
||
def test_delete_server_and_cleanup_empty_group(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
mcs.set_server(872490448, "B", "b.example.com:2", "js")
|
||
mcs.set_server(304484502, "C", "c.example.com:3", "bds")
|
||
|
||
mcs.delete_server(872490448, "A")
|
||
assert list(groups[872490448]) == ["B"]
|
||
mcs.delete_server(872490448, "B")
|
||
assert 872490448 not in groups # 删空连带删群
|
||
assert _file(mcs) == {"304484502": {"C": ["c.example.com:3", "bds"]}}
|
||
|
||
|
||
def test_delete_unknown_targets(mcs, groups):
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.delete_server(872490448, "没有这个群")
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.delete_server(872490448, "没有这台")
|
||
|
||
|
||
def test_delete_group(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
mcs.delete_group(872490448)
|
||
assert groups == {}
|
||
with pytest.raises(mcs.ServerError):
|
||
mcs.delete_group(872490448)
|
||
|
||
|
||
# ───────────────────────── 读 ─────────────────────────
|
||
|
||
|
||
def test_load_servers_handles_missing_broken_and_partial(mcs, tmp_path):
|
||
path = tmp_path / "mc_status_data.json"
|
||
assert mcs.load_servers(str(path)) == {} # 文件不存在
|
||
|
||
path.write_text("{ 坏掉的 json", encoding="utf-8")
|
||
assert mcs.load_servers(str(path)) == {} # 坏文件不炸
|
||
|
||
path.write_text(json.dumps({"8123": {"A": ["a:1", "js"], "B": "不是数组"}}), encoding="utf-8")
|
||
assert mcs.load_servers(str(path)) == {8123: {"A": ["a:1", "js"]}}
|
||
|
||
|
||
def test_snapshot_is_a_copy(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
snap = mcs.snapshot()
|
||
snap[872490448]["A"][0] = "改坏了"
|
||
snap[872490448]["新"] = ["x:1", "js"]
|
||
assert groups[872490448] == {"A": ["a.example.com:1", "js"]}
|
||
|
||
|
||
def test_lookup(mcs, groups):
|
||
mcs.set_server(872490448, "A", "a.example.com:1", "js")
|
||
assert mcs.lookup(872490448, "A") == ["a.example.com:1", "js"]
|
||
assert mcs.lookup(872490448, "没有这台") is None
|
||
assert mcs.lookup(999, "A") is None
|