From 945bc3bc349bb7c71820bd4cf4d2502293e9ccec Mon Sep 17 00:00:00 2001 From: Lishiling Date: Fri, 13 Mar 2026 11:02:31 +0800 Subject: [PATCH 1/2] =?UTF-8?q?add:=E6=95=B0=E6=8D=AE=E5=BA=93=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E6=89=B9=E9=87=8F=E6=93=8D=E4=BD=9C=E6=94=AF=E6=8C=81?= =?UTF-8?q?,=E9=99=84=E6=B5=8B=E8=AF=95=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 8 ++ src-new/astrbot_sdk/clients/db.py | 69 +++++++++++- src-new/astrbot_sdk/protocol/descriptors.py | 49 +++++++++ .../astrbot_sdk/runtime/capability_router.py | 94 +++++++++++++++- src-new/astrbot_sdk/runtime/peer.py | 4 +- tests_v4/test_capability_router.py | 90 ++++++++++++++++ tests_v4/test_db_client.py | 100 ++++++++++++++++++ 7 files changed, 407 insertions(+), 7 deletions(-) diff --git a/.gitignore b/.gitignore index 4ddd0641f..35c92ee93 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ __pycache__/ *.pyd *.so .pytest_cache/ +pytest-cache-files-*/ .mypy_cache/ .ruff_cache/ .coverage @@ -22,6 +23,13 @@ wheels/ .eggs/ pip-wheel-metadata/ +# +fork-docs/ +tmp/ +openspec/ +scripts/ +docs/zh/reference + # Virtual environments .venv/ venv/ diff --git a/src-new/astrbot_sdk/clients/db.py b/src-new/astrbot_sdk/clients/db.py index 7c270d23d..161c0c8bc 100644 --- a/src-new/astrbot_sdk/clients/db.py +++ b/src-new/astrbot_sdk/clients/db.py @@ -18,14 +18,11 @@ - 数据永久存储，除非用户显式删除 - 值类型支持任意 JSON 数据 - 支持前缀查询键列表 - -TODO: - - 缺少批量操作支持 (set_many, get_many) - - 缺少数据变更事件通知 """ from __future__ import annotations +from collections.abc import AsyncIterator, Mapping, Sequence from typing import Any from ._proxy import CapabilityProxy @@ -108,3 +105,67 @@ class DBClient: if not isinstance(keys, (list, tuple)): return [] return [str(item) for item in keys] + + async def get_many(self, keys: Sequence[str]) -> dict[str, Any | None]: + """批量获取多个键的值。 + + Args: + keys: 要读取的键列表 + + Returns: + 一个 dict，key 为键名，value 为对应值（不存在则为 None） + + 示例: + values = await ctx.db.get_many(["user:1", "user:2"]) + if values["user:1"] is None: + print("user:1 missing") + """ + output = await self._proxy.call("db.get_many", {"keys": list(keys)}) + items = output.get("items") + if not isinstance(items, (list, tuple)): + return {} + result: dict[str, Any | None] = {} + for item in items: + if not isinstance(item, dict): + continue + key = item.get("key") + if not isinstance(key, str): + continue + result[key] = item.get("value") + return result + + async def set_many( + self, items: Mapping[str, Any] | Sequence[tuple[str, Any]] + ) -> None: + """批量写入多个键值对。 + + Args: + items: 键值对集合（dict 或二元组序列） + + 示例: + await ctx.db.set_many({"user:1": {"name": "a"}, "user:2": {"name": "b"}}) + """ + if isinstance(items, Mapping): + pairs = list(items.items()) + else: + pairs = list(items) + + payload_items: list[dict[str, Any]] = [ + {"key": str(key), "value": value} for key, value in pairs + ] + await self._proxy.call("db.set_many", {"items": payload_items}) + + def watch(self, prefix: str | None = None) -> AsyncIterator[dict[str, Any]]: + """订阅 KV 变更事件（流式）。 + + Args: + prefix: 键前缀过滤；None 表示订阅所有键 + + Yields: + 变更事件 dict：{"op": "set"|"delete", "key": str, "value": Any|None} + + 示例: + async for event in ctx.db.watch("user:"): + print(event["op"], event["key"]) + """ + return self._proxy.stream("db.watch", {"prefix": prefix}) diff --git a/src-new/astrbot_sdk/protocol/descriptors.py b/src-new/astrbot_sdk/protocol/descriptors.py index 7df6017a1..bdeca089c 100644 --- a/src-new/astrbot_sdk/protocol/descriptors.py +++ b/src-new/astrbot_sdk/protocol/descriptors.py @@ -127,6 +127,37 @@ DB_LIST_OUTPUT_SCHEMA = _object_schema( required=("keys",), keys={"type": "array", "items": {"type": "string"}}, ) +DB_GET_MANY_INPUT_SCHEMA = _object_schema( + required=("keys",), + keys={"type": "array", "items": {"type": "string"}}, +) +DB_GET_MANY_OUTPUT_SCHEMA = _object_schema( + required=("items",), + items={ + "type": "array", + "items": _object_schema( + required=("key", "value"), + key={"type": "string"}, + value=_nullable({}), + ), + }, +) +DB_SET_MANY_INPUT_SCHEMA = _object_schema( + required=("items",), + items={ + "type": "array", + "items": _object_schema( + required=("key", "value"), + key={"type": "string"}, + value={}, + ), + }, +) +DB_SET_MANY_OUTPUT_SCHEMA = _object_schema() +DB_WATCH_INPUT_SCHEMA = _object_schema( + prefix=_nullable({"type": "string"}), +) +DB_WATCH_OUTPUT_SCHEMA = _object_schema() SESSION_REF_SCHEMA = _object_schema( required=("conversation_id",), conversation_id={"type": "string"}, @@ -218,6 +249,18 @@ BUILTIN_CAPABILITY_SCHEMAS: dict[str, dict[str, JSONSchema]] = { "input": DB_LIST_INPUT_SCHEMA, "output": DB_LIST_OUTPUT_SCHEMA, }, + "db.get_many": { + "input": DB_GET_MANY_INPUT_SCHEMA, + "output": DB_GET_MANY_OUTPUT_SCHEMA, + }, + "db.set_many": { + "input": DB_SET_MANY_INPUT_SCHEMA, + "output": DB_SET_MANY_OUTPUT_SCHEMA, + }, + "db.watch": { + "input": DB_WATCH_INPUT_SCHEMA, + "output": DB_WATCH_OUTPUT_SCHEMA, + }, "platform.send": { "input": PLATFORM_SEND_INPUT_SCHEMA, "output": PLATFORM_SEND_OUTPUT_SCHEMA, @@ -491,10 +534,16 @@ __all__ = [ "DB_DELETE_OUTPUT_SCHEMA", "DB_GET_INPUT_SCHEMA", "DB_GET_OUTPUT_SCHEMA", + "DB_GET_MANY_INPUT_SCHEMA", + "DB_GET_MANY_OUTPUT_SCHEMA", "DB_LIST_INPUT_SCHEMA", "DB_LIST_OUTPUT_SCHEMA", "DB_SET_INPUT_SCHEMA", "DB_SET_OUTPUT_SCHEMA", + "DB_SET_MANY_INPUT_SCHEMA", + "DB_SET_MANY_OUTPUT_SCHEMA", + "DB_WATCH_INPUT_SCHEMA", + "DB_WATCH_OUTPUT_SCHEMA", "EventTrigger", "HandlerDescriptor", "JSONSchema", diff --git a/src-new/astrbot_sdk/runtime/capability_router.py b/src-new/astrbot_sdk/runtime/capability_router.py index 32028d9dd..de33aa696 100644 --- a/src-new/astrbot_sdk/runtime/capability_router.py +++ b/src-new/astrbot_sdk/runtime/capability_router.py @@ -109,6 +109,7 @@ CAPABILITY_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$") class StreamExecution: iterator: AsyncIterator[dict[str, Any]] finalize: FinalizeHandler + collect_chunks: bool = True StreamHandler = Callable[ @@ -134,8 +135,18 @@ class CapabilityRouter: self.db_store: dict[str, Any] = {} self.memory_store: dict[str, dict[str, Any]] = {} self.sent_messages: list[dict[str, Any]] = [] + self._db_watch_subscriptions: dict[ + str, tuple[str | None, asyncio.Queue[dict[str, Any]]] + ] = {} self._register_builtin_capabilities() + def _emit_db_change(self, *, op: str, key: str, value: Any | None) -> None: + event = {"op": op, "key": key, "value": value} + for prefix, queue in list(self._db_watch_subscriptions.values()): + if prefix is not None and not key.startswith(prefix): + continue + queue.put_nowait(event) + def descriptors(self) -> list[CapabilityDescriptor]: return [ entry.descriptor for entry in self._registrations.values() if entry.exposed @@ -227,6 +238,7 @@ class CapabilityRouter: return StreamExecution( iterator=execution.iterator, finalize=validated_finalize, + collect_chunks=execution.collect_chunks, ) def _register_builtin_capabilities(self) -> None: @@ -329,13 +341,17 @@ class CapabilityRouter: _request_id: str, payload: dict[str, Any], _token ) -> dict[str, Any]: key = str(payload.get("key", "")) - self.db_store[key] = payload.get("value") + value = payload.get("value") + self.db_store[key] = value + self._emit_db_change(op="set", key=key, value=value) return {} async def db_delete( _request_id: str, payload: dict[str, Any], _token ) -> dict[str, Any]: - self.db_store.pop(str(payload.get("key", "")), None) + key = str(payload.get("key", "")) + self.db_store.pop(key, None) + self._emit_db_change(op="delete", key=key, value=None) return {} async def db_list( @@ -347,6 +363,63 @@ class CapabilityRouter: keys = [item for item in keys if item.startswith(prefix)] return {"keys": keys} + async def db_get_many( + _request_id: str, payload: dict[str, Any], _token + ) -> dict[str, Any]: + keys_payload = payload.get("keys") + if not isinstance(keys_payload, (list, tuple)): + raise AstrBotError.invalid_input("db.get_many 的 keys 必须是数组") + keys = [str(item) for item in keys_payload] + items = [{"key": key, "value": self.db_store.get(key)} for key in keys] + return {"items": items} + + async def db_set_many( + _request_id: str, payload: dict[str, Any], _token + ) -> dict[str, Any]: + items_payload = payload.get("items") + if not isinstance(items_payload, (list, tuple)): + raise AstrBotError.invalid_input("db.set_many 的 items 必须是数组") + for entry in items_payload: + if not isinstance(entry, dict): + raise AstrBotError.invalid_input( + "db.set_many 的 items 必须是 object 数组" + ) + key = str(entry.get("key", "")) + value = entry.get("value") + self.db_store[key] = value + self._emit_db_change(op="set", key=key, value=value) + return {} + + async def db_watch( + request_id: str, payload: dict[str, Any], _token + ) -> StreamExecution: + prefix = payload.get("prefix") + prefix_value: str | None + if isinstance(prefix, str): + prefix_value = prefix + elif prefix is None: + prefix_value = None + else: + raise AstrBotError.invalid_input( + "db.watch 的 prefix 必须是 string 或 null" + ) + + queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() + self._db_watch_subscriptions[request_id] = (prefix_value, queue) + + async def iterator() -> AsyncIterator[dict[str, Any]]: + try: + while True: + yield await queue.get() + finally: + self._db_watch_subscriptions.pop(request_id, None) + + return StreamExecution( + iterator=iterator(), + finalize=lambda _chunks: {}, + collect_chunks=False, + ) + async def platform_send( _request_id: str, payload: dict[str, Any], _token ) -> dict[str, Any]: @@ -460,6 +533,23 @@ class CapabilityRouter: builtin_descriptor("db.list", "列出 KV"), call_handler=db_list, ) + self.register( + builtin_descriptor("db.get_many", "批量读取 KV"), + call_handler=db_get_many, + ) + self.register( + builtin_descriptor("db.set_many", "批量写入 KV"), + call_handler=db_set_many, + ) + self.register( + builtin_descriptor( + "db.watch", + "订阅 KV 变更", + supports_stream=True, + cancelable=True, + ), + stream_handler=db_watch, + ) self.register( builtin_descriptor("platform.send", "发送消息"), call_handler=platform_send, diff --git a/src-new/astrbot_sdk/runtime/peer.py b/src-new/astrbot_sdk/runtime/peer.py index d737c9f57..f501c420d 100644 --- a/src-new/astrbot_sdk/runtime/peer.py +++ b/src-new/astrbot_sdk/runtime/peer.py @@ -507,9 +507,11 @@ class Peer: "stream=true 必须返回 StreamExecution" ) await self._send(EventMessage(id=message.id, phase="started")) + collect_chunks = execution.collect_chunks chunks: list[dict[str, Any]] = [] async for chunk in execution.iterator: - chunks.append(chunk) + if collect_chunks: + chunks.append(chunk) await self._send( EventMessage(id=message.id, phase="delta", data=chunk) ) diff --git a/tests_v4/test_capability_router.py b/tests_v4/test_capability_router.py index 759010aa8..de8a9118c 100644 --- a/tests_v4/test_capability_router.py +++ b/tests_v4/test_capability_router.py @@ -164,6 +164,9 @@ class TestCapabilityRouterInit: assert "db.set" in capability_names assert "db.delete" in capability_names assert "db.list" in capability_names + assert "db.get_many" in capability_names + assert "db.set_many" in capability_names + assert "db.watch" in capability_names # Platform capabilities assert "platform.send" in capability_names @@ -182,6 +185,14 @@ class TestCapabilityRouterInit: assert descriptors[name].input_schema == schema["input"] assert descriptors[name].output_schema == schema["output"] + def test_db_watch_descriptor_supports_stream(self): + router = CapabilityRouter() + descriptors = { + descriptor.name: descriptor for descriptor in router.descriptors() + } + + assert descriptors["db.watch"].supports_stream is True + class TestCapabilityRouterRegister: """Tests for CapabilityRouter.register method.""" @@ -350,6 +361,85 @@ class TestCapabilityRouterExecute: request_id="req-1", ) + +class TestCapabilityRouterDBWatch: + """Router-level tests for db.watch behavior.""" + + @pytest.mark.asyncio + async def test_db_watch_receives_set_and_delete_events(self): + router = CapabilityRouter() + token = CancelToken() + + execution = await router.execute( + "db.watch", + {"prefix": None}, + stream=True, + cancel_token=token, + request_id="watch-1", + ) + assert isinstance(execution, StreamExecution) + assert execution.collect_chunks is False + + await router.execute( + "db.set", + {"key": "a", "value": 1}, + stream=False, + cancel_token=token, + request_id="set-1", + ) + await router.execute( + "db.delete", + {"key": "a"}, + stream=False, + cancel_token=token, + request_id="del-1", + ) + + event1 = await anext(execution.iterator) + event2 = await anext(execution.iterator) + assert event1 == {"op": "set", "key": "a", "value": 1} + assert event2 == {"op": "delete", "key": "a", "value": None} + + close = getattr(execution.iterator, "aclose", None) + if close is not None: + await close() + + @pytest.mark.asyncio + async def test_db_watch_prefix_filters_events(self): + router = CapabilityRouter() + token = CancelToken() + + execution = await router.execute( + "db.watch", + {"prefix": "user:"}, + stream=True, + cancel_token=token, + request_id="watch-2", + ) + assert isinstance(execution, StreamExecution) + + await router.execute( + "db.set", + {"key": "sys:1", "value": 1}, + stream=False, + cancel_token=token, + request_id="set-2", + ) + await router.execute( + "db.set", + {"key": "user:1", "value": {"ok": True}}, + stream=False, + cancel_token=token, + request_id="set-3", + ) + + event = await anext(execution.iterator) + assert event == {"op": "set", "key": "user:1", "value": {"ok": True}} + + close = getattr(execution.iterator, "aclose", None) + if close is not None: + await close() + @pytest.mark.asyncio async def test_execute_missing_capability_raises(self): """execute should raise for unknown capability.""" diff --git a/tests_v4/test_db_client.py b/tests_v4/test_db_client.py index 8023b53b8..8d15c1310 100644 --- a/tests_v4/test_db_client.py +++ b/tests_v4/test_db_client.py @@ -237,3 +237,103 @@ class TestDBClientList: proxy.call.assert_called_once_with("db.list", {"prefix": None}) assert result == [] + + +class TestDBClientGetMany: + """Tests for DBClient.get_many() method.""" + + @pytest.mark.asyncio + async def test_get_many_returns_mapping(self): + proxy = MagicMock(spec=CapabilityProxy) + proxy.call = AsyncMock( + return_value={ + "items": [ + {"key": "a", "value": 1}, + {"key": "b", "value": None}, + ] + } + ) + client = DBClient(proxy) + + result = await client.get_many(["a", "b"]) + + proxy.call.assert_called_once_with("db.get_many", {"keys": ["a", "b"]}) + assert result == {"a": 1, "b": None} + + @pytest.mark.asyncio + async def test_get_many_returns_empty_dict_for_malformed_items(self): + proxy = MagicMock(spec=CapabilityProxy) + proxy.call = AsyncMock(return_value={"items": "not-a-list"}) + client = DBClient(proxy) + + result = await client.get_many(["a"]) + + assert result == {} + + +class TestDBClientSetMany: + """Tests for DBClient.set_many() method.""" + + @pytest.mark.asyncio + async def test_set_many_accepts_mapping(self): + proxy = MagicMock(spec=CapabilityProxy) + proxy.call = AsyncMock(return_value={}) + client = DBClient(proxy) + + await client.set_many({"a": 1, "b": 2}) + + proxy.call.assert_called_once_with( + "db.set_many", + {"items": [{"key": "a", "value": 1}, {"key": "b", "value": 2}]}, + ) + + @pytest.mark.asyncio + async def test_set_many_accepts_sequence_pairs(self): + proxy = MagicMock(spec=CapabilityProxy) + proxy.call = AsyncMock(return_value={}) + client = DBClient(proxy) + + await client.set_many([("a", True), ("b", {"x": 1})]) + + proxy.call.assert_called_once_with( + "db.set_many", + {"items": [{"key": "a", "value": True}, {"key": "b", "value": {"x": 1}}]}, + ) + + +class TestDBClientWatch: + """Tests for DBClient.watch() method.""" + + @pytest.mark.asyncio + async def test_watch_calls_proxy_stream_and_yields_events(self): + async def gen(): + yield {"op": "set", "key": "a", "value": 1} + yield {"op": "delete", "key": "a", "value": None} + + proxy = MagicMock(spec=CapabilityProxy) + proxy.stream = MagicMock(return_value=gen()) + client = DBClient(proxy) + + iterator = client.watch() + + proxy.stream.assert_called_once_with("db.watch", {"prefix": None}) + events = [event async for event in iterator] + assert events == [ + {"op": "set", "key": "a", "value": 1}, + {"op": "delete", "key": "a", "value": None}, + ] + + @pytest.mark.asyncio + async def test_watch_with_prefix(self): + async def gen(): + yield {"op": "set", "key": "user:1", "value": {"ok": True}} + + proxy = MagicMock(spec=CapabilityProxy) + proxy.stream = MagicMock(return_value=gen()) + client = DBClient(proxy) + + iterator = client.watch(prefix="user:") + + proxy.stream.assert_called_once_with("db.watch", {"prefix": "user:"}) + events = [event async for event in iterator] + assert events == [{"op": "set", "key": "user:1", "value": {"ok": True}}] From 19064ef8eb86fd6ef75bb61b23b301181e6adacc Mon Sep 17 00:00:00 2001 From: Lishiling Date: Fri, 13 Mar 2026 11:46:33 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:conftest=E5=92=8Chelpers=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests_v4/conftest.py | 201 +++++++++++++++++++++++++++++++++---------- tests_v4/helpers.py | 79 ++++++++++++++--- 2 files changed, 224 insertions(+), 56 deletions(-) diff --git a/tests_v4/conftest.py b/tests_v4/conftest.py index 5f653e724..e213f288b 100644 --- a/tests_v4/conftest.py +++ b/tests_v4/conftest.py @@ -1,8 +1,8 @@ -"""Pytest shared fixtures and test bootstrap helpers.""" +"""Pytest共享的fixture和测试引导辅助函数。""" -# ruff: noqa: E402 +# ruff: noqa: E402 # 忽略E402（模块导入顺序）警告 -# Test configuration +# 测试配置 import asyncio import sys import tempfile @@ -13,89 +13,144 @@ from typing import Any import pytest -# 将 src-new 加入路径 - 这使得测试可以运行，但不算"已安装" +# 将src-new目录添加到Python路径 - 这使得测试可以运行，但不算作"已安装"的包 SRC_NEW_PATH = str(Path(__file__).parent.parent / "src-new") sys.path.insert(0, SRC_NEW_PATH) # ============================================================ -# Async Configuration +# 异步测试配置 # ============================================================ @pytest.fixture(scope="session") def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]: - """Create an event loop for async tests.""" - policy = asyncio.get_event_loop_policy() - loop = policy.new_event_loop() - yield loop - loop.close() + """为异步测试创建事件循环。 + + 这是一个会话级别的fixture，在整个测试会话期间只创建一次事件循环。 + """ + policy = asyncio.get_event_loop_policy() # 获取当前事件循环策略 + loop = policy.new_event_loop() # 创建新的事件循环 + yield loop # 提供事件循环给测试使用 + loop.close() # 测试结束后关闭事件循环 # ============================================================ -# Transport Fixtures +# 传输层Fixture（用于模拟网络通信） # ============================================================ class MemoryTransport: - """In-memory transport for testing peer communication.""" + """用于测试对等通信的内存传输模拟。 + + 这个类模拟了两个对等方之间的通信通道，所有消息都在内存中传递， + 无需实际网络连接。 + """ def __init__(self) -> None: - self._closed = asyncio.Event() - self._message_handler = None - self.partner: MemoryTransport | None = None + self._closed = asyncio.Event() # 用于跟踪传输是否已关闭 + self._message_handler = None # 消息处理函数 + self.partner: MemoryTransport | None = None # 通信伙伴 def set_message_handler(self, handler) -> None: + """设置消息处理函数。 + + Args: + handler: 接收消息的异步函数 + """ self._message_handler = handler async def start(self) -> None: + """启动传输。 + + 清除关闭状态，使传输可用。 + """ self._closed.clear() async def stop(self) -> None: + """停止传输。 + + 设置关闭事件，表示传输已停止。 + """ self._closed.set() async def wait_closed(self) -> None: + """等待传输关闭。 + + 阻塞直到传输完全关闭。 + """ await self._closed.wait() async def send(self, payload: str) -> None: + """发送消息给伙伴。 + + Args: + payload: 要发送的消息内容 + + Raises: + RuntimeError: 如果没有设置伙伴传输 + """ if self.partner is None: raise RuntimeError("MemoryTransport 未连接 partner") if self.partner._message_handler is not None: - await self.partner._message_handler(payload) + await self.partner._message_handler(payload) # 将消息分发给伙伴的处理函数 async def _dispatch(self, payload: str) -> None: + """内部方法：将消息分发给本地处理函数。 + + Args: + payload: 接收到的消息内容 + """ if self._message_handler is not None: await self._message_handler(payload) @pytest.fixture def transport_pair() -> tuple[MemoryTransport, MemoryTransport]: - """Create a connected pair of in-memory transports.""" + """创建一对相互连接的内存传输实例。 + + 返回的左右两个传输实例互为伙伴，可用于模拟两个对等方之间的通信。 + + Returns: + (left_transport, right_transport) 的元组 + """ left = MemoryTransport() right = MemoryTransport() - left.partner = right - right.partner = left + left.partner = right # 左传输的伙伴是右传输 + right.partner = left # 右传输的伙伴是左传输 return left, right # ============================================================ -# Mock/Fake Fixtures +# 模拟/Fake Fixture（用于测试的假对象） # ============================================================ class FakeEnvManager: - """Fake environment manager for testing.""" + """用于测试的虚假环境管理器。 + + 模拟真实的环境管理器行为，但不执行实际的环境准备操作。 + """ def prepare_environment(self, _plugin: Any) -> Path: + """模拟准备插件环境。 + + Args: + _plugin: 插件对象（未使用） + + Returns: + 返回当前Python解释器路径作为模拟的环境路径 + """ return Path(sys.executable) @pytest.fixture def fake_env_manager() -> FakeEnvManager: - """Provide a fake environment manager.""" + """提供一个虚假的环境管理器fixture。""" return FakeEnvManager() +# 导入需要使用的类型 from astrbot_sdk.protocol.messages import InitializeOutput, PeerInfo from astrbot_sdk.runtime.capability_router import CapabilityRouter from astrbot_sdk.runtime.peer import Peer @@ -103,66 +158,111 @@ from astrbot_sdk.runtime.peer import Peer @pytest.fixture async def core_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer: - """Create a core peer with default handlers.""" - left, _ = transport_pair - router = CapabilityRouter() + """创建一个配置了默认处理函数的核心对等方。 + 这个fixture创建并启动一个核心角色的对等方，设置了初始化和调用处理函数。 + + Args: + transport_pair: 传输对fixture + + Returns: + 已启动的核心对等方实例 + """ + left, _ = transport_pair # 使用传输对中的左传输 + router = CapabilityRouter() # 创建能力路由器 + + # 创建核心对等方 peer = Peer( transport=left, peer_info=PeerInfo(name="core", role="core", version="v4"), ) + # 定义初始化处理函数 async def init_handler(_message) -> InitializeOutput: return InitializeOutput( peer=PeerInfo(name="core", role="core", version="v4"), - capabilities=router.descriptors(), + capabilities=router.descriptors(), # 获取路由器描述的能力列表 metadata={}, ) + # 定义调用处理函数 async def invoke_handler(message, token): return await router.execute( - message.capability, - message.input, - stream=message.stream, - cancel_token=token, - request_id=message.id, + message.capability, # 要执行的能力 + message.input, # 输入参数 + stream=message.stream, # 是否流式输出 + cancel_token=token, # 取消令牌 + request_id=message.id, # 请求ID ) + # 设置处理函数 peer.set_initialize_handler(init_handler) peer.set_invoke_handler(invoke_handler) - await peer.start() + await peer.start() # 启动对等方 yield peer - await peer.stop() + await peer.stop() # 测试结束后停止 @pytest.fixture async def plugin_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer: - """Create a plugin peer connected to core.""" - _, right = transport_pair + """创建一个连接到核心的插件对等方。 + 这个fixture创建并启动一个插件角色的对等方。 + + Args: + transport_pair: 传输对fixture + + Returns: + 已启动的插件对等方实例 + """ + _, right = transport_pair # 使用传输对中的右传输 + + # 创建插件对等方 peer = Peer( transport=right, peer_info=PeerInfo(name="plugin", role="plugin", version="v4"), ) - await peer.start() + await peer.start() # 启动对等方 yield peer - await peer.stop() + await peer.stop() # 测试结束后停止 @pytest.fixture def temp_plugin_dir() -> Generator[Path, None, None]: - """Create a temporary directory for plugin testing.""" + """创建用于插件测试的临时目录。 + + 这个fixture创建一个临时目录，并在测试结束后自动清理。 + + Yields: + 临时目录的Path对象 + """ with tempfile.TemporaryDirectory() as temp_dir: yield Path(temp_dir) def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None: - """Helper to create a minimal test plugin.""" + """辅助函数：创建一个最小的测试插件。 + + 在指定目录创建插件所需的基本文件结构： + - commands/__init__.py + - commands/sample.py（包含测试命令） + - requirements.txt（空文件） + - plugin.yaml（插件配置文件） + + Args: + plugin_root: 插件根目录 + name: 插件名称，默认为"test_plugin" + """ + # 创建commands目录和__init__.py文件 (plugin_root / "commands").mkdir(parents=True, exist_ok=True) (plugin_root / "commands" / "__init__.py").write_text("", encoding="utf-8") + + # 创建空的requirements.txt (plugin_root / "requirements.txt").write_text("", encoding="utf-8") + + # 创建插件配置文件plugin.yaml (plugin_root / "plugin.yaml").write_text( textwrap.dedent( f"""\ @@ -173,7 +273,7 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None: author: tester version: 0.1.0 runtime: - python: "{sys.version_info.major}.{sys.version_info.minor}" + python: "{sys.version_info.major}.{sys.version_info.minor}" # 使用当前Python版本 components: - class: commands.sample:TestPlugin type: command @@ -183,6 +283,8 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None: ), encoding="utf-8", ) + + # 创建测试命令文件sample.py (plugin_root / "commands" / "sample.py").write_text( textwrap.dedent( """\ @@ -190,9 +292,9 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None: class TestPlugin(Star): - @on_command("test") + @on_command("test") # 注册test命令 async def test_cmd(self, event: MessageEvent, ctx: Context): - await event.reply("test ok") + await event.reply("test ok") # 回复消息 """ ), encoding="utf-8", @@ -201,7 +303,16 @@ def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None: @pytest.fixture def test_plugin(temp_plugin_dir: Path) -> Path: - """Create a test plugin and return its root directory.""" + """创建一个测试插件并返回其根目录。 + + 这个fixture使用create_test_plugin函数创建插件，并返回插件目录路径。 + + Args: + temp_plugin_dir: 临时插件目录fixture + + Returns: + 包含插件的目录路径 + """ plugin_root = temp_plugin_dir / "plugins" / "test_plugin" - create_test_plugin(plugin_root) - return temp_plugin_dir / "plugins" + create_test_plugin(plugin_root) # 创建测试插件 + return temp_plugin_dir / "plugins" \ No newline at end of file diff --git a/tests_v4/helpers.py b/tests_v4/helpers.py index 528231f69..be91ef17f 100644 --- a/tests_v4/helpers.py +++ b/tests_v4/helpers.py @@ -1,4 +1,4 @@ -from __future__ import annotations +from __future__ import annotations # 启用延迟类型注解求值，避免循环引用 import asyncio from pathlib import Path @@ -7,34 +7,91 @@ from astrbot_sdk.runtime.transport import Transport class MemoryTransport(Transport): + """基于内存的传输层实现，用于测试场景。 + + 继承自Transport基类，模拟两个对等方之间的通信，所有消息在内存中传递， + 无需实际网络连接。主要用于单元测试。 + """ + def __init__(self) -> None: - super().__init__() - self.partner: "MemoryTransport | None" = None + """初始化内存传输实例。""" + super().__init__() # 调用父类初始化方法 + self.partner: "MemoryTransport | None" = None # 通信伙伴，可以是对等的另一个MemoryTransport实例 async def start(self) -> None: - self._closed.clear() + """启动传输。 + + 通过清除关闭标志使传输变为可用状态。 + """ + self._closed.clear() # 清除关闭事件 async def stop(self) -> None: - self._closed.set() + """停止传输。 + + 设置关闭标志，表示传输已停止。 + """ + self._closed.set() # 设置关闭事件 async def send(self, payload: str) -> None: + """发送消息给伙伴。 + + Args: + payload: 要发送的消息内容字符串 + + Raises: + RuntimeError: 如果没有设置伙伴传输（即self.partner为None） + """ if self.partner is None: raise RuntimeError("MemoryTransport 未连接 partner") + # 将消息转发给伙伴的_dispatch方法进行处理 await self.partner._dispatch(payload) def make_transport_pair() -> tuple[MemoryTransport, MemoryTransport]: - left = MemoryTransport() - right = MemoryTransport() - left.partner = right - right.partner = left - return left, right + """创建一对相互连接的内存传输实例。 + + 工厂函数，用于创建两个互为通信伙伴的MemoryTransport实例， + 简化测试设置过程。 + + Returns: + tuple[MemoryTransport, MemoryTransport]: 返回左右两个传输实例的元组， + 它们已互相设置为伙伴关系 + """ + left = MemoryTransport() # 创建左侧传输实例 + right = MemoryTransport() # 创建右侧传输实例 + left.partner = right # 设置左侧的伙伴为右侧 + right.partner = left # 设置右侧的伙伴为左侧 + return left, right # 返回配对的传输实例 class FakeEnvManager: + """虚假的环境管理器，用于测试。 + + 模拟真实环境管理器的行为，但不执行实际的环境准备操作， + 主要用于需要环境管理器但又不希望产生副作用的测试场景。 + """ + def prepare_environment(self, _plugin) -> Path: + """模拟准备插件环境的方法。 + + 不实际创建虚拟环境，而是返回当前Python解释器路径作为模拟的环境路径。 + + Args: + _plugin: 插件对象参数，在模拟实现中未使用 + + Returns: + Path: 当前Python解释器的路径 + """ + # 动态导入sys模块并返回可执行文件路径 return Path(__import__("sys").executable) async def drain_loop() -> None: - await asyncio.sleep(0.05) + """清空事件循环的辅助函数。 + + 通过短暂的异步睡眠，让事件循环有机会处理所有已排队的任务和回调。 + 常用于测试中等待异步操作完成，确保所有待处理的事件被处理。 + + 睡眠时间(0.05秒)足够短不会明显减慢测试，又足够长让事件循环有机会处理任务。 + """ + await asyncio.sleep(0.05) # 暂停当前协程50毫秒，让出控制权给事件循环 \ No newline at end of file