HeXi/tests/test_mc_servers.py

"""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