From d07e8f90ef3458419ad37d2c132a638f71771a81 Mon Sep 17 00:00:00 2001 From: UnCLAS-Prommer Date: Sun, 22 Mar 2026 18:32:14 +0800 Subject: [PATCH] fix: remove nc ada pytest --- pytests/test_napcat_adapter_codec.py | 151 -------------------------- pytests/test_napcat_adapter_config.py | 91 ---------------- pytests/test_napcat_adapter_plugin.py | 60 ---------- 3 files changed, 302 deletions(-) delete mode 100644 pytests/test_napcat_adapter_codec.py delete mode 100644 pytests/test_napcat_adapter_config.py delete mode 100644 pytests/test_napcat_adapter_plugin.py diff --git a/pytests/test_napcat_adapter_codec.py b/pytests/test_napcat_adapter_codec.py deleted file mode 100644 index 97ed1d9e..00000000 --- a/pytests/test_napcat_adapter_codec.py +++ /dev/null @@ -1,151 +0,0 @@ -from pathlib import Path -from typing import Any, Dict - -import importlib -import sys -from types import SimpleNamespace - -import pytest - - -BUILT_IN_PLUGIN_ROOT = Path(__file__).resolve().parents[1] / "src" / "plugins" / "built_in" -if str(BUILT_IN_PLUGIN_ROOT) not in sys.path: - sys.path.insert(0, str(BUILT_IN_PLUGIN_ROOT)) - -NapCatInboundCodec = importlib.import_module("napcat_adapter.codec_inbound").NapCatInboundCodec -NapCatOutboundCodec = importlib.import_module("napcat_adapter.codec_outbound").NapCatOutboundCodec - - -def test_napcat_outbound_codec_supports_binary_and_forward_segments() -> None: - codec = NapCatOutboundCodec() - raw_message = [ - {"type": "text", "data": "hello"}, - {"type": "image", "data": "", "hash": "h1", "binary_data_base64": "aW1hZ2U="}, - {"type": "emoji", "data": "", "hash": "h2", "binary_data_base64": "ZW1vamk="}, - {"type": "voice", "data": "", "hash": "h3", "binary_data_base64": "dm9pY2U="}, - { - "type": "reply", - "data": { - "target_message_id": "origin-1", - "target_message_content": "origin text", - }, - }, - { - "type": "forward", - "data": [ - { - "user_id": "42", - "user_nickname": "alice", - "user_cardname": "Alice", - "message_id": "fwd-1", - "content": [{"type": "text", "data": "node-text"}], - } - ], - }, - ] - - converted = codec.convert_segments(raw_message) - - assert converted[0] == {"type": "text", "data": {"text": "hello"}} - assert converted[1]["type"] == "image" - assert converted[1]["data"]["file"] == "base64://aW1hZ2U=" - assert converted[2]["type"] == "image" - assert converted[2]["data"]["subtype"] == 1 - assert converted[3] == {"type": "record", "data": {"file": "base64://dm9pY2U="}} - assert converted[4] == {"type": "reply", "data": {"id": "origin-1"}} - assert converted[5]["type"] == "node" - assert converted[5]["data"]["name"] == "alice" - assert converted[5]["data"]["content"] == [{"type": "text", "data": {"text": "node-text"}}] - - -def test_napcat_outbound_codec_builds_private_action_from_route_metadata() -> None: - codec = NapCatOutboundCodec() - message: Dict[str, Any] = { - "message_info": { - "user_info": {"user_id": "10001", "user_nickname": "tester"}, - "additional_config": {}, - }, - "raw_message": [{"type": "text", "data": "hello"}], - } - - action_name, params = codec.build_outbound_action(message, {"target_user_id": "30001"}) - - assert action_name == "send_private_msg" - assert params == {"message": [{"type": "text", "data": {"text": "hello"}}], "user_id": "30001"} - - -class DummyQueryService: - """用于测试的轻量查询服务。""" - - async def download_binary(self, url: str) -> bytes: - """返回固定图片二进制。 - - Args: - url: 图片地址。 - - Returns: - bytes: 固定测试图片二进制。 - """ - if url: - return b"image-bytes" - return b"" - - async def get_message_detail(self, message_id: str) -> Dict[str, Any] | None: - """返回空消息详情。 - - Args: - message_id: 目标消息 ID。 - - Returns: - Dict[str, Any] | None: 固定空结果。 - """ - del message_id - return None - - async def get_record_detail(self, file_name: str, file_id: str | None = None) -> Dict[str, Any] | None: - """返回空语音详情。 - - Args: - file_name: 语音文件名。 - file_id: 可选文件 ID。 - - Returns: - Dict[str, Any] | None: 固定空结果。 - """ - del file_name - del file_id - return None - - async def get_forward_message(self, message_id: str) -> Dict[str, Any] | None: - """返回空转发详情。 - - Args: - message_id: 转发消息 ID。 - - Returns: - Dict[str, Any] | None: 固定空结果。 - """ - del message_id - return None - - -@pytest.mark.asyncio -async def test_napcat_inbound_codec_parses_cq_string_image_segments() -> None: - codec = NapCatInboundCodec(SimpleNamespace(debug=lambda message: None), DummyQueryService()) - payload = { - "message": "[CQ:image,file=test.png,sub_type=0,url=https://example.com/test.png][CQ:at,qq=10001] 看到是国人直接给你封了", - } - - raw_message, is_at = await codec.convert_segments(payload, "10001") - - assert raw_message[0]["type"] == "image" - assert raw_message[1] == { - "type": "at", - "data": { - "target_user_id": "10001", - "target_user_nickname": None, - "target_user_cardname": None, - }, - } - assert raw_message[2] == {"type": "text", "data": " 看到是国人直接给你封了"} - assert is_at is True diff --git a/pytests/test_napcat_adapter_config.py b/pytests/test_napcat_adapter_config.py deleted file mode 100644 index 688b1a48..00000000 --- a/pytests/test_napcat_adapter_config.py +++ /dev/null @@ -1,91 +0,0 @@ -from pathlib import Path -from typing import List - -import importlib -import sys - - -BUILT_IN_PLUGIN_ROOT = Path(__file__).resolve().parents[1] / "src" / "plugins" / "built_in" -if str(BUILT_IN_PLUGIN_ROOT) not in sys.path: - sys.path.insert(0, str(BUILT_IN_PLUGIN_ROOT)) - -NapCatPluginSettings = importlib.import_module("napcat_adapter.config").NapCatPluginSettings - - -class DummyLogger: - """用于测试的轻量日志对象。""" - - def __init__(self) -> None: - """初始化测试日志对象。""" - self.warnings: List[str] = [] - self.errors: List[str] = [] - - def warning(self, message: str) -> None: - """记录警告日志。 - - Args: - message: 待记录的日志内容。 - """ - self.warnings.append(message) - - def error(self, message: str) -> None: - """记录错误日志。 - - Args: - message: 待记录的日志内容。 - """ - self.errors.append(message) - - -def test_parse_new_napcat_server_config() -> None: - logger = DummyLogger() - settings = NapCatPluginSettings.from_mapping( - { - "plugin": {"enabled": True, "config_version": "0.1.0"}, - "napcat_server": { - "host": "localhost", - "port": 8095, - "token": "secret", - "heartbeat_interval": 45, - "reconnect_delay_sec": 7, - "action_timeout_sec": 18, - "connection_id": "main", - }, - }, - logger, - ) - - assert settings.should_connect() is True - assert settings.napcat_server.host == "localhost" - assert settings.napcat_server.port == 8095 - assert settings.napcat_server.token == "secret" - assert settings.napcat_server.heartbeat_interval == 45.0 - assert settings.napcat_server.reconnect_delay_sec == 7.0 - assert settings.napcat_server.action_timeout_sec == 18.0 - assert settings.napcat_server.connection_id == "main" - assert settings.napcat_server.build_ws_url() == "ws://localhost:8095" - assert settings.validate(logger) is True - - -def test_parse_legacy_connection_ws_url_fallback() -> None: - logger = DummyLogger() - settings = NapCatPluginSettings.from_mapping( - { - "plugin": {"enabled": True, "config_version": "0.1.0"}, - "connection": { - "ws_url": "ws://127.0.0.1:3001", - "access_token": "legacy-token", - "heartbeat_sec": 35, - "action_timeout_sec": 12, - }, - }, - logger, - ) - - assert settings.napcat_server.host == "127.0.0.1" - assert settings.napcat_server.port == 3001 - assert settings.napcat_server.token == "legacy-token" - assert settings.napcat_server.heartbeat_interval == 35.0 - assert settings.napcat_server.action_timeout_sec == 12.0 - assert settings.validate(logger) is True - assert logger.warnings diff --git a/pytests/test_napcat_adapter_plugin.py b/pytests/test_napcat_adapter_plugin.py deleted file mode 100644 index ca550a39..00000000 --- a/pytests/test_napcat_adapter_plugin.py +++ /dev/null @@ -1,60 +0,0 @@ -"""NapCat 插件入口行为测试。""" - -from pathlib import Path -from typing import List -from types import SimpleNamespace - -import importlib -import sys - -import pytest - - -BUILT_IN_PLUGIN_ROOT = Path(__file__).resolve().parents[1] / "src" / "plugins" / "built_in" -if str(BUILT_IN_PLUGIN_ROOT) not in sys.path: - sys.path.insert(0, str(BUILT_IN_PLUGIN_ROOT)) - -NapCatAdapterPlugin = importlib.import_module("napcat_adapter.plugin").NapCatAdapterPlugin - - -class DummyLogger: - """用于测试的轻量日志对象。""" - - def __init__(self) -> None: - """初始化测试日志对象。""" - self.debug_messages: List[str] = [] - - def debug(self, message: str) -> None: - """记录调试日志。 - - Args: - message: 待记录的日志内容。 - """ - self.debug_messages.append(message) - - -@pytest.mark.asyncio -async def test_on_config_update_refreshes_settings_and_restarts(monkeypatch: pytest.MonkeyPatch) -> None: - """配置更新时应刷新插件配置、清空旧 settings，并触发连接重启。""" - plugin = NapCatAdapterPlugin() - plugin._ctx = SimpleNamespace(logger=DummyLogger()) - plugin._settings = object() - - restart_calls: List[dict] = [] - - async def fake_restart() -> None: - """记录一次重启调用。""" - restart_calls.append(dict(plugin._plugin_config)) - - monkeypatch.setattr(plugin, "_restart_connection_if_needed", fake_restart) - - new_config = { - "plugin": {"enabled": True, "config_version": "0.1.0"}, - "napcat_server": {"host": "127.0.0.1", "port": 3001}, - } - await plugin.on_config_update(new_config, "v2") - - assert plugin._plugin_config == new_config - assert plugin._settings is None - assert restart_calls == [new_config] - assert plugin.ctx.logger.debug_messages == ["NapCat 适配器收到配置更新通知: v2"]