fix: remove nc ada pytest
This commit is contained in:
UnCLAS-Prommer
committed by
DrSmoothl
DrSmoothl
parent
0066224251
commit
d07e8f90ef
@@ -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
|
||||
@@ -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
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user