This commit is contained in:
DrSmoothl
2026-04-03 15:24:53 +08:00
parent 6c720e0403
commit a2fd0a1750

View File

@@ -1,9 +1,9 @@
"""NapCat 插件与新 SDK 对接测试。""" """NapCat 插件与新 SDK 对接测试。"""
from importlib import import_module, util
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List from typing import Any, Dict, List, Tuple
import importlib
import logging import logging
import sys import sys
@@ -12,8 +12,10 @@ import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = Path(__file__).resolve().parents[1]
PLUGINS_ROOT = PROJECT_ROOT / "plugins" PLUGINS_ROOT = PROJECT_ROOT / "plugins"
SDK_ROOT = PROJECT_ROOT / "packages" / "maibot-plugin-sdk" SDK_ROOT = PROJECT_ROOT / "packages" / "maibot-plugin-sdk"
NAPCAT_PLUGIN_DIR = PLUGINS_ROOT / "MaiBot-Napcat-Adapter"
NAPCAT_TEST_MODULE = "_test_napcat_adapter"
for import_path in (str(PLUGINS_ROOT), str(SDK_ROOT)): for import_path in (str(SDK_ROOT),):
if import_path not in sys.path: if import_path not in sys.path:
sys.path.insert(0, import_path) sys.path.insert(0, import_path)
@@ -63,7 +65,41 @@ class _FakeGatewayCapability:
return True return True
def _load_napcat_sdk_symbols() -> tuple[Any, Any, Any, Any]: def _load_napcat_sdk_modules() -> Tuple[Any, Any, Any, Any]:
"""动态加载 NapCat 插件测试所需的模块。
Returns:
tuple[Any, Any, Any, Any]:
依次返回常量模块、配置模块、插件模块和运行时状态模块。
"""
if NAPCAT_TEST_MODULE not in sys.modules:
plugin_path = NAPCAT_PLUGIN_DIR / "plugin.py"
spec = util.spec_from_file_location(
NAPCAT_TEST_MODULE,
plugin_path,
submodule_search_locations=[str(NAPCAT_PLUGIN_DIR)],
)
if spec is None or spec.loader is None:
raise ImportError(f"无法为 NapCat 插件创建模块规格: {plugin_path}")
module = util.module_from_spec(spec)
sys.modules[NAPCAT_TEST_MODULE] = module
try:
spec.loader.exec_module(module)
except Exception:
sys.modules.pop(NAPCAT_TEST_MODULE, None)
raise
return (
import_module(f"{NAPCAT_TEST_MODULE}.constants"),
import_module(f"{NAPCAT_TEST_MODULE}.config"),
import_module(f"{NAPCAT_TEST_MODULE}.plugin"),
import_module(f"{NAPCAT_TEST_MODULE}.runtime_state"),
)
def _load_napcat_sdk_symbols() -> Tuple[Any, Any, Any, Any]:
"""动态加载 NapCat 插件测试所需的符号。 """动态加载 NapCat 插件测试所需的符号。
Returns: Returns:
@@ -71,10 +107,7 @@ def _load_napcat_sdk_symbols() -> tuple[Any, Any, Any, Any]:
依次返回网关名常量、配置类、插件类和运行时状态管理器类。 依次返回网关名常量、配置类、插件类和运行时状态管理器类。
""" """
constants_module = importlib.import_module("napcat_adapter.constants") constants_module, config_module, plugin_module, runtime_state_module = _load_napcat_sdk_modules()
config_module = importlib.import_module("napcat_adapter.config")
plugin_module = importlib.import_module("napcat_adapter.plugin")
runtime_state_module = importlib.import_module("napcat_adapter.runtime_state")
return ( return (
constants_module.NAPCAT_GATEWAY_NAME, constants_module.NAPCAT_GATEWAY_NAME,
config_module.NapCatServerConfig, config_module.NapCatServerConfig,
@@ -103,6 +136,63 @@ def test_napcat_plugin_collects_duplex_message_gateway() -> None:
assert gateway_component["metadata"]["protocol"] == "napcat" assert gateway_component["metadata"]["protocol"] == "napcat"
def test_napcat_plugin_uses_sdk_config_model() -> None:
"""NapCat 插件应声明 SDK 配置模型并暴露默认配置与 Schema。"""
constants_module, _config_module, plugin_module, _runtime_state_module = _load_napcat_sdk_modules()
plugin = plugin_module.NapCatAdapterPlugin()
default_config = plugin.get_default_config()
schema = plugin.get_webui_config_schema(plugin_id="maibot-team.napcat-adapter")
assert default_config["plugin"]["config_version"] == constants_module.SUPPORTED_CONFIG_VERSION
assert default_config["chat"]["ban_qq_bot"] is False
assert default_config["filters"]["ignore_self_message"] is True
assert schema["plugin_id"] == "maibot-team.napcat-adapter"
assert schema["sections"]["chat"]["fields"]["group_list"]["type"] == "array"
assert schema["sections"]["chat"]["fields"]["group_list_type"]["choices"] == ["whitelist", "blacklist"]
def test_napcat_plugin_normalizes_legacy_config_values() -> None:
"""NapCat 插件应兼容旧配置字段并输出规范化结果。"""
constants_module, _config_module, plugin_module, _runtime_state_module = _load_napcat_sdk_modules()
plugin = plugin_module.NapCatAdapterPlugin()
plugin.set_plugin_config(
{
"plugin": {"enabled": True, "config_version": ""},
"connection": {
"access_token": "secret-token",
"heartbeat_sec": "45",
"ws_url": "ws://10.0.0.8:3012/onebot/v11/ws",
},
"chat": {
"ban_qq_bot": True,
"ban_user_id": ["42", 42, ""],
"group_list": [123, " 456 ", None, "123"],
"group_list_type": "whitelist",
"private_list": "invalid",
"private_list_type": "unexpected",
},
"filters": {"ignore_self_message": True},
}
)
config_data = plugin.get_plugin_config_data()
assert "connection" not in config_data
assert config_data["plugin"]["config_version"] == constants_module.SUPPORTED_CONFIG_VERSION
assert config_data["napcat_server"]["host"] == "10.0.0.8"
assert config_data["napcat_server"]["port"] == 3012
assert config_data["napcat_server"]["token"] == "secret-token"
assert config_data["napcat_server"]["heartbeat_interval"] == 45.0
assert config_data["chat"]["group_list"] == ["123", "456"]
assert config_data["chat"]["private_list"] == []
assert config_data["chat"]["private_list_type"] == constants_module.DEFAULT_CHAT_LIST_TYPE
assert plugin.config.napcat_server.build_ws_url() == "ws://10.0.0.8:3012"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runtime_state_reports_via_gateway_capability() -> None: async def test_runtime_state_reports_via_gateway_capability() -> None:
"""NapCat 运行时状态应通过新的消息网关能力上报。""" """NapCat 运行时状态应通过新的消息网关能力上报。"""