From 623e0c1f332bfa3c31ccf1b3fb7e94d9e78c681a Mon Sep 17 00:00:00 2001 From: whatevertogo Date: Fri, 13 Mar 2026 20:26:42 +0800 Subject: [PATCH] feat: Enhance CLI and testing capabilities - Added a new script entry point `astrbot-sdk` in `pyproject.toml`. - Introduced `has_waiter` method in `SessionWaiterManager` to check for existing waiters. - Updated `cli.py` to improve error handling and added context to error messages. - Implemented local development support in `cli.py` with a new `dev` command for running plugins against a mock core. - Created a new testing module `astrbot_sdk.testing` with utilities for local development and plugin testing. - Added comprehensive tests for the new testing module and CLI commands. - Improved compatibility and error messaging for plugin loading failures. --- ARCHITECTURE.md | 31 +- pyproject.toml | 1 + src-new/astrbot/core/platform/register.py | 3 - src-new/astrbot_sdk/_session_waiter.py | 3 + src-new/astrbot_sdk/cli.py | 242 +++++- src-new/astrbot_sdk/testing.py | 855 ++++++++++++++++++++++ tests_v4/test_testing_module.py | 132 ++++ tests_v4/test_top_level_modules.py | 81 ++ 8 files changed, 1342 insertions(+), 6 deletions(-) create mode 100644 src-new/astrbot_sdk/testing.py create mode 100644 tests_v4/test_testing_module.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 336494c48..45288976c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -58,6 +58,8 @@ src-new/ │ ├── events.py # MessageEvent / PlainTextResult │ ├── errors.py # AstrBotError / ErrorCodes │ ├── star.py # Star 基类与 handler 收集 +│ ├── cli.py # astr / astrbot-sdk CLI 入口 +│ ├── testing.py # 本地开发与测试 harness │ ├── compat.py # 旧顶层兼容重导出 │ ├── _legacy_api.py # LegacyContext / LegacyStar / CommandComponent │ ├── _legacy_llm.py # legacy LLM/tool 兼容辅助 @@ -273,7 +275,32 @@ from astrbot.core.utils.session_waiter import session_waiter 只有在需要兼容现有旧插件时才应继续使用这些路径；新插件应直接使用 v4 顶层入口。 -## 9. 测试与维护约定 +## 9. 本地开发与测试 + +当前仓库已经提供一条受控的本地开发路径： + +- CLI：`astr dev --local` 与 `astrbot-sdk dev --local` +- 稳定测试入口：`astrbot_sdk.testing` + +`astrbot_sdk.testing` 当前公开的稳定面包括： + +- `PluginHarness` +- `LocalRuntimeConfig` +- `MockPeer` +- `MockCapabilityRouter` +- `InMemoryDB` +- `InMemoryMemory` +- `StdoutPlatformSink` +- `RecordedSend` + +设计约束： + +- 本地 harness 复用真实的 `load_plugin()`、`HandlerDispatcher`、`CapabilityDispatcher`、`_legacy_runtime.py` 与 `_session_waiter.py` +- `dev --local` 使用进程内 mock core，而不是重新发明一套并行 runtime +- 同一次 `interactive` 会话会复用同一个 dispatcher / waiter manager / in-memory db / in-memory memory +- `astrbot_sdk.testing` 是插件测试依赖的公开 API，minor 版本内保持兼容稳定 + +## 10. 测试与维护约定 - 当前主测试目录是 `tests_v4/`，覆盖 protocol、runtime、clients、compat facade、legacy plugin integration、top-level imports 与 integration flows。 - 文档维护规则： @@ -281,7 +308,7 @@ from astrbot.core.utils.session_waiter import session_waiter - compat 支持级别变化时，同时更新本文档、`CLAUDE.md` / `AGENTS.md` 备注以及相关契约测试。 - `refactor.md` 不再承载现状；出现冲突时，一律以本文档和代码/测试为准。 -## 10. 当前建议的后续演进方向 +## 11. 当前建议的后续演进方向 1. 继续把 runtime 对 compat 的认知收口到 `_legacy_runtime.py`。 2. 继续拆薄 `_legacy_api.py`，让 `LegacyContext` 更偏向 facade 和 orchestration。 diff --git a/pyproject.toml b/pyproject.toml index 146cd1dd9..0dcbfcc6c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ [project.scripts] astr = "astrbot_sdk.cli:cli" +astrbot-sdk = "astrbot_sdk.cli:cli" [tool.setuptools] package-dir = {"" = "src-new"} diff --git a/src-new/astrbot/core/platform/register.py b/src-new/astrbot/core/platform/register.py index 0ea3b88a3..4ce287619 100644 --- a/src-new/astrbot/core/platform/register.py +++ b/src-new/astrbot/core/platform/register.py @@ -2,9 +2,6 @@ 旧版 ``astrbot.core.platform.register`` 导入路径兼容入口。 TODO: 目前仅保留符号以兼容导入，后续如果需要，可以在这里实现一个基于当前平台能力的注册系统适配器。 """ - - - def register_platform_adapter(*args, **kwargs): raise NotImplementedError( "astrbot.core.platform.register_platform_adapter() 尚未在 v4 兼容层实现。" diff --git a/src-new/astrbot_sdk/_session_waiter.py b/src-new/astrbot_sdk/_session_waiter.py index a7e38bdb2..04b7f2ae3 100644 --- a/src-new/astrbot_sdk/_session_waiter.py +++ b/src-new/astrbot_sdk/_session_waiter.py @@ -57,6 +57,9 @@ class SessionWaiterManager: self._waiters[key] = state return key + def has_waiter(self, event: Any) -> bool: + return self.session_key(event) in self._waiters + def unregister(self, key: str, state: _SessionWaitState) -> None: if self._waiters.get(key) is state: self._waiters.pop(key, None) diff --git a/src-new/astrbot_sdk/cli.py b/src-new/astrbot_sdk/cli.py index 20a806836..a513406d0 100644 --- a/src-new/astrbot_sdk/cli.py +++ b/src-new/astrbot_sdk/cli.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import sys +import typing from collections.abc import Coroutine from pathlib import Path from typing import Any @@ -11,7 +12,22 @@ from typing import Any import click from loguru import logger +from .errors import AstrBotError from .runtime.bootstrap import run_plugin_worker, run_supervisor, run_websocket_server +from .testing import ( + LocalRuntimeConfig, + PluginHarness, + StdoutPlatformSink, + _PluginExecutionError, + _PluginLoadError, +) + +EXIT_OK = 0 +EXIT_UNEXPECTED = 1 +EXIT_USAGE = 2 +EXIT_PLUGIN_LOAD = 3 +EXIT_RUNTIME = 4 +EXIT_PLUGIN_EXECUTION = 5 def setup_logger(verbose: bool = False) -> None: @@ -30,10 +46,170 @@ def _run_async_entrypoint( *, log_message: str, log_level: str = "info", + context: dict[str, Any] | None = None, ) -> None: log_method = getattr(logger, log_level) log_method(log_message) - asyncio.run(entrypoint) + try: + asyncio.run(entrypoint) + except Exception as exc: + exit_code, error_code, hint = _classify_cli_exception(exc) + _render_cli_error( + error_code=error_code, + message=str(exc), + hint=hint, + context=context, + ) + if exit_code == EXIT_UNEXPECTED: + logger.exception("CLI 异常退出") + raise SystemExit(exit_code) from exc + + +def _classify_cli_exception(exc: Exception) -> tuple[int, str, str]: + if isinstance(exc, AstrBotError): + return ( + EXIT_RUNTIME, + exc.code, + exc.hint or "请检查本地 mock core 与插件调用参数", + ) + if isinstance( + exc, + (_PluginLoadError, FileNotFoundError, ImportError, ModuleNotFoundError), + ): + return ( + EXIT_PLUGIN_LOAD, + "plugin_load_error", + "请检查插件目录、plugin.yaml、requirements.txt 和导入路径", + ) + if isinstance(exc, LookupError): + return ( + EXIT_RUNTIME, + "dispatch_error", + "请检查 handler 或 capability 是否已正确注册", + ) + if isinstance(exc, _PluginExecutionError): + return ( + EXIT_PLUGIN_EXECUTION, + "plugin_execution_error", + "请检查插件生命周期、handler 或 capability 的实现", + ) + return ( + EXIT_UNEXPECTED, + "unexpected_error", + "请查看详细日志，必要时使用 --verbose 重试", + ) + + +def _render_cli_error( + *, + error_code: str, + message: str, + hint: str = "", + context: dict[str, Any] | None = None, +) -> None: + click.echo(f"Error[{error_code}]: {message}", err=True) + if hint: + click.echo(f"Suggestion: {hint}", err=True) + if not context: + return + for key, value in context.items(): + click.echo(f"{key}: {value}", err=True) + + +async def _run_local_dev( + *, + plugin_dir: Path, + event_text: str | None, + interactive: bool, + session_id: str, + user_id: str, + platform: str, + group_id: str | None, + event_type: str, +) -> None: + sink = StdoutPlatformSink(stream=sys.stdout) + harness = PluginHarness( + LocalRuntimeConfig( + plugin_dir=plugin_dir, + session_id=session_id, + user_id=user_id, + platform=platform, + group_id=group_id, + event_type=event_type, + ), + platform_sink=sink, + ) + state = { + "session_id": session_id, + "user_id": user_id, + "platform": platform, + "group_id": group_id, + "event_type": event_type, + } + async with harness: + if interactive: + click.echo( + "本地交互模式已启动。可用命令：/session /user /platform /group /private /event /exit" + ) + while True: + line = await asyncio.to_thread(sys.stdin.readline) + if not line: + break + text = line.strip() + if not text: + continue + if _handle_dev_meta_command(text, state): + if text in {"/exit", "/quit"}: + break + continue + await harness.dispatch_text( + text, + session_id=str(state["session_id"]), + user_id=str(state["user_id"]), + platform=str(state["platform"]), + group_id=typing.cast(str | None, state["group_id"]), + event_type=str(state["event_type"]), + ) + return + assert event_text is not None + await harness.dispatch_text( + event_text, + session_id=session_id, + user_id=user_id, + platform=platform, + group_id=group_id, + event_type=event_type, + ) + + +def _handle_dev_meta_command(command: str, state: dict[str, Any]) -> bool: + if command in {"/exit", "/quit"}: + return True + if command.startswith("/session "): + state["session_id"] = command.split(" ", 1)[1].strip() + click.echo(f"切换 session_id -> {state['session_id']}") + return True + if command.startswith("/user "): + state["user_id"] = command.split(" ", 1)[1].strip() + click.echo(f"切换 user_id -> {state['user_id']}") + return True + if command.startswith("/platform "): + state["platform"] = command.split(" ", 1)[1].strip() + click.echo(f"切换 platform -> {state['platform']}") + return True + if command.startswith("/group "): + state["group_id"] = command.split(" ", 1)[1].strip() + click.echo(f"切换 group_id -> {state['group_id']}") + return True + if command == "/private": + state["group_id"] = None + click.echo("已切换为私聊上下文") + return True + if command.startswith("/event "): + state["event_type"] = command.split(" ", 1)[1].strip() + click.echo(f"切换 event_type -> {state['event_type']}") + return True + return False @click.group() @@ -58,6 +234,68 @@ def run(plugins_dir: Path) -> None: _run_async_entrypoint( run_supervisor(plugins_dir=plugins_dir), log_message=f"启动插件主管进程，插件目录：{plugins_dir}", + context={"plugins_dir": plugins_dir}, + ) + + +@cli.command() +@click.option( + "--plugin-dir", + required=True, + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + help="Plugin directory to run locally", +) +@click.option("--local", "local_mode", is_flag=True, help="Run against local mock core") +@click.option( + "--standalone", + "standalone_mode", + is_flag=True, + help="Alias of --local for compatibility", +) +@click.option("--event-text", type=str, help="Single message text to dispatch") +@click.option("--interactive", is_flag=True, help="Read follow-up messages from stdin") +@click.option("--session-id", default="local-session", show_default=True) +@click.option("--user-id", default="local-user", show_default=True) +@click.option("--platform", "platform_name", default="test", show_default=True) +@click.option("--group-id", default=None) +@click.option("--event-type", default="message", show_default=True) +def dev( + plugin_dir: Path, + local_mode: bool, + standalone_mode: bool, + event_text: str | None, + interactive: bool, + session_id: str, + user_id: str, + platform_name: str, + group_id: str | None, + event_type: str, +) -> None: + """Run a plugin against the local mock core for development.""" + if not (local_mode or standalone_mode): + raise click.BadParameter("当前 dev 只支持 --local/--standalone 模式") + if interactive and event_text: + raise click.BadParameter("--interactive 与 --event-text 不能同时使用") + if not interactive and not event_text: + raise click.BadParameter("请提供 --event-text，或改用 --interactive") + _run_async_entrypoint( + _run_local_dev( + plugin_dir=plugin_dir, + event_text=event_text, + interactive=interactive, + session_id=session_id, + user_id=user_id, + platform=platform_name, + group_id=group_id, + event_type=event_type, + ), + log_message=f"启动本地开发模式：{plugin_dir}", + context={ + "plugin_dir": plugin_dir, + "session_id": session_id, + "platform": platform_name, + "event_type": event_type, + }, ) @@ -73,6 +311,7 @@ def worker(plugin_dir: Path) -> None: run_plugin_worker(plugin_dir=plugin_dir), log_message=f"启动插件工作进程：{plugin_dir}", log_level="debug", + context={"plugin_dir": plugin_dir}, ) @@ -83,4 +322,5 @@ def websocket(port: int) -> None: _run_async_entrypoint( run_websocket_server(port=port), log_message=f"启动 WebSocket 服务器，端口：{port}", + context={"port": port}, ) diff --git a/src-new/astrbot_sdk/testing.py b/src-new/astrbot_sdk/testing.py new file mode 100644 index 000000000..054817cf4 --- /dev/null +++ b/src-new/astrbot_sdk/testing.py @@ -0,0 +1,855 @@ +"""本地开发与插件测试辅助。 + +`astrbot_sdk.testing` 是面向插件作者的稳定开发入口： + +- `PluginHarness` 负责复用现有 loader / dispatcher / compat 执行链 +- `MockCapabilityRouter` 提供进程内 mock core 能力 +- `MockPeer` 让 `Context` 客户端继续走真实的 capability 调用路径 +- `StdoutPlatformSink` / `RecordedSend` 提供可观测的发送记录 + +这个模块刻意不暴露 runtime 内部编排数据结构，只封装本地开发/测试真正 +需要的最小稳定面。 +""" + +from __future__ import annotations + +import asyncio +import inspect +import re +import shlex +import typing +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, TextIO, get_type_hints + +from ._legacy_runtime import ( + bind_legacy_runtime_contexts, + run_legacy_worker_shutdown_hooks, + run_legacy_worker_startup_hooks, +) +from .context import CancelToken, Context as RuntimeContext +from .errors import AstrBotError +from .events import MessageEvent +from .protocol.descriptors import ( + CommandTrigger, + EventTrigger, + MessageTrigger, + ScheduleTrigger, +) +from .protocol.messages import EventMessage, InvokeMessage, PeerInfo +from .runtime.capability_router import CapabilityRouter, StreamExecution +from .runtime.handler_dispatcher import CapabilityDispatcher, HandlerDispatcher +from .runtime.loader import ( + LoadedHandler, + LoadedPlugin, + PluginSpec, + load_plugin, + load_plugin_spec, +) +from .star import Star + + +class _PluginLoadError(RuntimeError): + """本地 harness 初始化阶段的已知插件加载失败。""" + + +class _PluginExecutionError(RuntimeError): + """本地 harness 执行插件代码时的已知插件异常。""" + + +@dataclass(slots=True) +class RecordedSend: + """结构化发送记录，供断言和本地调试输出复用。""" + + kind: str + message_id: str + session_id: str + text: str | None = None + image_url: str | None = None + chain: list[dict[str, Any]] | None = None + target: dict[str, Any] | None = None + raw: dict[str, Any] = field(default_factory=dict) + + @property + def session(self) -> str: + return self.session_id + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> "RecordedSend": + if "text" in payload: + kind = "text" + elif "image_url" in payload: + kind = "image" + elif "chain" in payload: + kind = "chain" + else: + kind = "unknown" + return cls( + kind=kind, + message_id=str(payload.get("message_id", "")), + session_id=str(payload.get("session", "")), + text=payload.get("text") if isinstance(payload.get("text"), str) else None, + image_url=( + payload.get("image_url") + if isinstance(payload.get("image_url"), str) + else None + ), + chain=( + [dict(item) for item in payload.get("chain", [])] + if isinstance(payload.get("chain"), list) + else None + ), + target=( + dict(payload.get("target")) + if isinstance(payload.get("target"), dict) + else None + ), + raw=dict(payload), + ) + + +class StdoutPlatformSink: + """把 platform.* 的发送结果同时写到终端与内存记录。""" + + def __init__(self, stream: TextIO | None = None) -> None: + self._stream = stream + self.records: list[RecordedSend] = [] + + def record(self, item: RecordedSend) -> None: + self.records.append(item) + if self._stream is None: + return + self._stream.write(self._format(item) + "\n") + self._stream.flush() + + def clear(self) -> None: + self.records.clear() + + def _format(self, item: RecordedSend) -> str: + if item.kind == "text": + return f"[text][{item.session_id}] {item.text or ''}" + if item.kind == "image": + return f"[image][{item.session_id}] {item.image_url or ''}" + if item.kind == "chain": + count = len(item.chain or []) + return f"[chain][{item.session_id}] {count} components" + return f"[send][{item.session_id}] {item.raw}" + + +class InMemoryDB: + """测试友好的 KV 视图，直接绑定到 mock router 的内存存储。""" + + def __init__(self, store: dict[str, Any]) -> None: + self._store = store + + def get(self, key: str, default: Any = None) -> Any: + return self._store.get(key, default) + + def set(self, key: str, value: Any) -> None: + self._store[key] = value + + def delete(self, key: str) -> None: + self._store.pop(key, None) + + def list(self, prefix: str | None = None) -> list[str]: + keys = sorted(self._store.keys()) + if prefix is None: + return keys + return [key for key in keys if key.startswith(prefix)] + + def get_many(self, keys: list[str]) -> list[dict[str, Any]]: + return [{"key": key, "value": self._store.get(key)} for key in keys] + + def set_many(self, items: list[dict[str, Any]]) -> None: + for item in items: + self.set(str(item.get("key", "")), item.get("value")) + + +class InMemoryMemory: + """测试友好的 memory 视图，保持与 mock router 同步。""" + + def __init__(self, store: dict[str, dict[str, Any]]) -> None: + self._store = store + + def get(self, key: str, default: Any = None) -> Any: + return self._store.get(key, default) + + def save(self, key: str, value: dict[str, Any]) -> None: + self._store[key] = dict(value) + + def delete(self, key: str) -> None: + self._store.pop(key, None) + + def search(self, query: str) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + for key, value in self._store.items(): + if query in key or query in str(value): + results.append({"key": key, "value": value}) + return results + + +class MockCapabilityRouter(CapabilityRouter): + """本地 mock core，直接复用已有的内建 capability 实现。""" + + def __init__(self, *, platform_sink: StdoutPlatformSink | None = None) -> None: + self.platform_sink = platform_sink or StdoutPlatformSink() + super().__init__() + self.db = InMemoryDB(self.db_store) + self.memory = InMemoryMemory(self.memory_store) + + async def execute( + self, + capability: str, + payload: dict[str, Any], + *, + stream: bool, + cancel_token, + request_id: str, + ) -> dict[str, Any] | StreamExecution: + before = len(self.sent_messages) + result = await super().execute( + capability, + payload, + stream=stream, + cancel_token=cancel_token, + request_id=request_id, + ) + self._flush_platform_records(before) + return result + + def _flush_platform_records(self, start_index: int) -> None: + for payload in self.sent_messages[start_index:]: + self.platform_sink.record(RecordedSend.from_payload(payload)) + + +class MockPeer: + """满足 `Context`/`CapabilityProxy` 需要的最小 peer。""" + + def __init__(self, router: MockCapabilityRouter) -> None: + self._router = router + self._counter = 0 + self.remote_peer = PeerInfo( + name="astrbot-local-core", + role="core", + version="local", + ) + self.remote_capabilities = list(router.descriptors()) + self.remote_capability_map = { + item.name: item for item in self.remote_capabilities + } + self.remote_handlers: list[Any] = [] + self.remote_provided_capabilities: list[Any] = [] + self.remote_metadata = {"mode": "local"} + + async def invoke( + self, + capability: str, + payload: dict[str, Any], + *, + stream: bool = False, + request_id: str | None = None, + ) -> dict[str, Any]: + if stream: + raise ValueError("stream=True 请使用 invoke_stream()") + return typing.cast( + dict[str, Any], + await self._router.execute( + capability, + payload, + stream=False, + cancel_token=CancelToken(), + request_id=request_id or self._next_id(), + ), + ) + + async def invoke_stream( + self, + capability: str, + payload: dict[str, Any], + *, + request_id: str | None = None, + include_completed: bool = False, + ): + request_id = request_id or self._next_id() + execution = typing.cast( + StreamExecution, + await self._router.execute( + capability, + payload, + stream=True, + cancel_token=CancelToken(), + request_id=request_id, + ), + ) + + async def iterator(): + yield EventMessage(id=request_id, phase="started") + chunks: list[dict[str, Any]] = [] + async for chunk in execution.iterator: + if execution.collect_chunks: + chunks.append(chunk) + yield EventMessage(id=request_id, phase="delta", data=chunk) + output = execution.finalize(chunks) + if include_completed: + yield EventMessage(id=request_id, phase="completed", output=output) + + return iterator() + + def _next_id(self) -> str: + self._counter += 1 + return f"local_{self._counter:04d}" + + +@dataclass(slots=True) +class LocalRuntimeConfig: + """本地 harness 的稳定配置对象。""" + + plugin_dir: Path + session_id: str = "local-session" + user_id: str = "local-user" + platform: str = "test" + group_id: str | None = None + event_type: str = "message" + + +class PluginHarness: + """本地插件消息泵。 + + 这里复用真实的 loader / dispatcher / compat 执行链，只负责： + - 在同一个事件循环里装配单插件运行时 + - 维持本地 mock core 与发送记录 + - 把后续消息持续送入同一个 dispatcher/session_waiter 图 + """ + + def __init__( + self, + config: LocalRuntimeConfig, + *, + platform_sink: StdoutPlatformSink | None = None, + ) -> None: + self.config = config + self.platform_sink = platform_sink or StdoutPlatformSink() + self.router = MockCapabilityRouter(platform_sink=self.platform_sink) + self.peer = MockPeer(self.router) + self.plugin: PluginSpec | None = None + self.loaded_plugin: LoadedPlugin | None = None + self.dispatcher: HandlerDispatcher | None = None + self.capability_dispatcher: CapabilityDispatcher | None = None + self.lifecycle_context: RuntimeContext | None = None + self._request_counter = 0 + self._started = False + + async def __aenter__(self) -> "PluginHarness": + await self.start() + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.stop() + + @property + def sent_messages(self) -> list[RecordedSend]: + return list(self.platform_sink.records) + + def clear_sent_messages(self) -> None: + self.platform_sink.clear() + + async def start(self) -> None: + if self._started: + return + try: + self.plugin = load_plugin_spec(self.config.plugin_dir) + self.loaded_plugin = load_plugin(self.plugin) + except Exception as exc: # pragma: no cover - 由 CLI/集成测试覆盖 + raise _PluginLoadError(str(exc)) from exc + self.dispatcher = HandlerDispatcher( + plugin_id=self.plugin.name, + peer=self.peer, + handlers=self.loaded_plugin.handlers, + ) + self.capability_dispatcher = CapabilityDispatcher( + plugin_id=self.plugin.name, + peer=self.peer, + capabilities=self.loaded_plugin.capabilities, + ) + self.lifecycle_context = RuntimeContext( + peer=self.peer, + plugin_id=self.plugin.name, + ) + bind_legacy_runtime_contexts( + [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities], + self.lifecycle_context, + ) + try: + await self._run_lifecycle("on_start") + await run_legacy_worker_startup_hooks( + [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities], + context=self.lifecycle_context, + metadata=dict(self.plugin.manifest_data), + ) + except AstrBotError: + raise + except Exception as exc: # pragma: no cover - 由 CLI/集成测试覆盖 + raise _PluginExecutionError(str(exc)) from exc + self._started = True + + async def stop(self) -> None: + if ( + not self._started + or self.loaded_plugin is None + or self.lifecycle_context is None + ): + return + try: + await run_legacy_worker_shutdown_hooks( + [*self.loaded_plugin.handlers, *self.loaded_plugin.capabilities], + context=self.lifecycle_context, + metadata=dict(self.plugin.manifest_data), + ) + await self._run_lifecycle("on_stop") + finally: + self._started = False + + async def dispatch_text( + self, + text: str, + *, + session_id: str | None = None, + user_id: str | None = None, + platform: str | None = None, + group_id: str | None = None, + event_type: str | None = None, + request_id: str | None = None, + ) -> list[RecordedSend]: + payload = self.build_event_payload( + text=text, + session_id=session_id, + user_id=user_id, + platform=platform, + group_id=group_id, + event_type=event_type, + request_id=request_id, + ) + return await self.dispatch_event(payload, request_id=request_id) + + async def dispatch_event( + self, + event_payload: dict[str, Any], + *, + request_id: str | None = None, + ) -> list[RecordedSend]: + await self.start() + assert self.loaded_plugin is not None + assert self.dispatcher is not None + + start_index = len(self.platform_sink.records) + if self._has_waiter_for_event(event_payload): + carrier = ( + self.loaded_plugin.handlers[0] if self.loaded_plugin.handlers else None + ) + if carrier is None: + raise AstrBotError.invalid_input( + "当前没有可用于承接 session_waiter 的 handler" + ) + await self._invoke_handler( + carrier, + event_payload, + args={}, + request_id=request_id, + ) + await self._wait_for_followup_side_effects( + start_index=start_index, + event_payload=event_payload, + ) + return self.platform_sink.records[start_index:] + + matches = self._match_handlers(event_payload) + if not matches: + raise AstrBotError.invalid_input("未找到匹配的 handler") + for loaded, args in matches: + await self._invoke_handler( + loaded, + event_payload, + args=args, + request_id=request_id, + ) + return self.platform_sink.records[start_index:] + + async def invoke_capability( + self, + capability: str, + payload: dict[str, Any], + *, + request_id: str | None = None, + stream: bool = False, + ) -> dict[str, Any] | StreamExecution: + await self.start() + assert self.capability_dispatcher is not None + message = InvokeMessage( + id=request_id or self._next_request_id("cap"), + capability=capability, + input=dict(payload), + stream=stream, + ) + try: + return await self.capability_dispatcher.invoke(message, CancelToken()) + except AstrBotError: + raise + except Exception as exc: # pragma: no cover - 由 CLI/集成测试覆盖 + raise _PluginExecutionError(str(exc)) from exc + + def build_event_payload( + self, + *, + text: str, + session_id: str | None = None, + user_id: str | None = None, + platform: str | None = None, + group_id: str | None = None, + event_type: str | None = None, + request_id: str | None = None, + ) -> dict[str, Any]: + session_value = session_id or self.config.session_id + group_value = group_id if group_id is not None else self.config.group_id + event_type_value = event_type or self.config.event_type + payload = { + "type": event_type_value, + "event_type": event_type_value, + "text": text, + "session_id": session_value, + "user_id": user_id or self.config.user_id, + "platform": platform or self.config.platform, + "group_id": group_value, + "raw": { + "trace_id": request_id or self._next_request_id("trace"), + "event_type": event_type_value, + }, + } + if group_value: + payload["message_type"] = "group" + elif payload["user_id"]: + payload["message_type"] = "private" + else: + payload["message_type"] = "other" + return payload + + async def _invoke_handler( + self, + loaded: LoadedHandler, + event_payload: dict[str, Any], + *, + args: dict[str, Any], + request_id: str | None = None, + ) -> None: + assert self.dispatcher is not None + message = InvokeMessage( + id=request_id or self._next_request_id("msg"), + capability="handler.invoke", + input={ + "handler_id": loaded.descriptor.id, + "event": dict(event_payload), + "args": dict(args), + }, + ) + try: + await self.dispatcher.invoke(message, CancelToken()) + except AstrBotError: + raise + except Exception as exc: # pragma: no cover - 由 CLI/集成测试覆盖 + raise _PluginExecutionError(str(exc)) from exc + + async def _wait_for_followup_side_effects( + self, + *, + start_index: int, + event_payload: dict[str, Any], + ) -> None: + for _ in range(20): + if len(self.platform_sink.records) > start_index: + return + await asyncio.sleep(0) + if not self._has_waiter_for_event(event_payload): + return + + async def _run_lifecycle(self, method_name: str) -> None: + assert self.loaded_plugin is not None + assert self.lifecycle_context is not None + + for instance in self.loaded_plugin.instances: + hook = self._resolve_lifecycle_hook(instance, method_name) + if hook is None: + continue + args: list[Any] = [] + try: + signature = inspect.signature(hook) + except (TypeError, ValueError): + signature = None + if signature is not None: + positional_params = [ + parameter + for parameter in signature.parameters.values() + if parameter.kind + in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ) + ] + if positional_params: + args.append(self.lifecycle_context) + result = hook(*args) + if inspect.isawaitable(result): + await result + + def _match_handlers( + self, + event_payload: dict[str, Any], + ) -> list[tuple[LoadedHandler, dict[str, Any]]]: + assert self.loaded_plugin is not None + ranked: list[tuple[int, int, LoadedHandler, dict[str, Any]]] = [] + for index, loaded in enumerate(self.loaded_plugin.handlers): + args = self._match_handler(loaded, event_payload) + if args is None: + continue + ranked.append((loaded.descriptor.priority, index, loaded, args)) + ranked.sort(key=lambda item: (-item[0], item[1])) + return [(loaded, args) for _priority, _index, loaded, args in ranked] + + def _match_handler( + self, + loaded: LoadedHandler, + event_payload: dict[str, Any], + ) -> dict[str, Any] | None: + trigger = loaded.descriptor.trigger + if isinstance(trigger, CommandTrigger): + return self._match_command_trigger(loaded, trigger, event_payload) + if isinstance(trigger, MessageTrigger): + return self._match_message_trigger(loaded, trigger, event_payload) + if isinstance(trigger, EventTrigger): + current_type = str( + event_payload.get("event_type") + or event_payload.get("type") + or "message" + ) + if current_type != trigger.event_type: + return None + return {} + if isinstance(trigger, ScheduleTrigger): + return None + return None + + def _match_command_trigger( + self, + loaded: LoadedHandler, + trigger: CommandTrigger, + event_payload: dict[str, Any], + ) -> dict[str, Any] | None: + if not self._passes_trigger_constraints( + trigger.platforms, trigger.message_types, event_payload + ): + return None + text = str(event_payload.get("text", "")).strip() + for command_name in [trigger.command, *trigger.aliases]: + if not command_name: + continue + match = self._match_command_name(text, command_name) + if match is None: + continue + return self._build_command_args(loaded.callable, match) + return None + + def _match_message_trigger( + self, + loaded: LoadedHandler, + trigger: MessageTrigger, + event_payload: dict[str, Any], + ) -> dict[str, Any] | None: + if not self._passes_trigger_constraints( + trigger.platforms, trigger.message_types, event_payload + ): + return None + text = str(event_payload.get("text", "")) + if trigger.regex: + match = re.search(trigger.regex, text) + if match is None: + return None + return self._build_regex_args(loaded.callable, match) + if trigger.keywords and not any( + keyword in text for keyword in trigger.keywords + ): + return None + return {} + + def _passes_trigger_constraints( + self, + platforms: list[str], + message_types: list[str], + event_payload: dict[str, Any], + ) -> bool: + platform = str(event_payload.get("platform", "")) + if platforms and platform not in platforms: + return False + if not message_types: + return True + current_message_type = self._message_type_name(event_payload) + return current_message_type in message_types + + def _has_waiter_for_event(self, event_payload: dict[str, Any]) -> bool: + assert self.dispatcher is not None + probe_event = MessageEvent.from_payload( + event_payload, + context=self.lifecycle_context, + ) + return self.dispatcher._session_waiters.has_waiter(probe_event) + + @staticmethod + def _message_type_name(event_payload: dict[str, Any]) -> str: + explicit = str(event_payload.get("message_type", "")).lower() + if explicit in {"group", "private", "other"}: + return explicit + if event_payload.get("group_id"): + return "group" + if event_payload.get("user_id"): + return "private" + return "other" + + @staticmethod + def _match_command_name(text: str, command_name: str) -> str | None: + if text == command_name: + return "" + if text.startswith(f"{command_name} "): + return text[len(command_name) :].strip() + return None + + def _build_command_args(self, handler, remainder: str) -> dict[str, Any]: + names = self._legacy_arg_parameter_names(handler) + if not names or not remainder: + return {} + if len(names) == 1: + return {names[0]: remainder} + tokens = self._split_command_remainder(remainder) + if not tokens: + return {} + values: dict[str, Any] = {} + for index, name in enumerate(names): + if index >= len(tokens): + break + if index == len(names) - 1: + values[name] = " ".join(tokens[index:]) + break + values[name] = tokens[index] + return values + + def _build_regex_args(self, handler, match: re.Match[str]) -> dict[str, Any]: + named = { + key: value for key, value in match.groupdict().items() if value is not None + } + names = [ + name + for name in self._legacy_arg_parameter_names(handler) + if name not in named + ] + positional = [value for value in match.groups() if value is not None] + for index, value in enumerate(positional): + if index >= len(names): + break + named[names[index]] = value + return named + + @staticmethod + def _split_command_remainder(remainder: str) -> list[str]: + try: + return shlex.split(remainder) + except ValueError: + return remainder.split() + + @staticmethod + def _resolve_lifecycle_hook(instance: Any, method_name: str): + hook = getattr(instance, method_name, None) + marker = getattr(instance.__class__, "__astrbot_is_new_star__", None) + is_new_star = True + if callable(marker): + is_new_star = bool(marker()) + + if hook is not None and callable(hook): + bound_func = getattr(hook, "__func__", hook) + star_default = getattr(Star, method_name, None) + if star_default is None or bound_func is not star_default: + return hook + + if not is_new_star: + alias = {"on_start": "initialize", "on_stop": "terminate"}.get(method_name) + if alias is not None: + legacy_hook = getattr(instance, alias, None) + if legacy_hook is not None and callable(legacy_hook): + return legacy_hook + + if hook is not None and callable(hook): + return hook + return None + + def _legacy_arg_parameter_names(self, handler) -> list[str]: + try: + signature = inspect.signature(handler) + except (TypeError, ValueError): + return [] + try: + type_hints = get_type_hints(handler) + except Exception: + type_hints = {} + names: list[str] = [] + for parameter in signature.parameters.values(): + if parameter.kind not in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ): + continue + if self._is_injected_parameter( + parameter.name, type_hints.get(parameter.name) + ): + continue + names.append(parameter.name) + return names + + def _is_injected_parameter(self, name: str, annotation: Any) -> bool: + if name in {"event", "ctx", "context"}: + return True + normalized = self._unwrap_optional(annotation) + if normalized is None: + return False + if normalized is RuntimeContext: + return True + if normalized is MessageEvent: + return True + if isinstance(normalized, type) and issubclass( + normalized, (RuntimeContext, MessageEvent) + ): + return True + return False + + @staticmethod + def _unwrap_optional(annotation: Any) -> Any: + if annotation is None: + return None + origin = typing.get_origin(annotation) + if origin is typing.Union: + options = [ + item for item in typing.get_args(annotation) if item is not type(None) + ] + if len(options) == 1: + return options[0] + return annotation + + def _next_request_id(self, prefix: str) -> str: + self._request_counter += 1 + return f"{prefix}_{self._request_counter:04d}" + + +__all__ = [ + "InMemoryDB", + "InMemoryMemory", + "LocalRuntimeConfig", + "MockCapabilityRouter", + "MockPeer", + "PluginHarness", + "RecordedSend", + "StdoutPlatformSink", +] diff --git a/tests_v4/test_testing_module.py b/tests_v4/test_testing_module.py new file mode 100644 index 000000000..38da1f911 --- /dev/null +++ b/tests_v4/test_testing_module.py @@ -0,0 +1,132 @@ +"""Tests for the public local-dev/testing helpers.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest + +import astrbot_sdk.testing as testing_module +from astrbot_sdk.testing import ( + LocalRuntimeConfig, + MockCapabilityRouter, + MockPeer, + PluginHarness, +) + + +class TestTestingModule: + """Tests for `astrbot_sdk.testing` exports and behavior.""" + + def test_public_all_matches_stable_testing_surface(self): + """testing.__all__ should stay aligned with the documented stable helper API.""" + assert testing_module.__all__ == [ + "InMemoryDB", + "InMemoryMemory", + "LocalRuntimeConfig", + "MockCapabilityRouter", + "MockPeer", + "PluginHarness", + "RecordedSend", + "StdoutPlatformSink", + ] + + @pytest.mark.asyncio + async def test_mock_peer_stream_emits_event_messages(self): + """MockPeer.invoke_stream should behave like a peer-level event stream.""" + router = MockCapabilityRouter() + peer = MockPeer(router) + + stream = await peer.invoke_stream( + "llm.stream_chat", + {"prompt": "hi"}, + include_completed=True, + ) + phases = [] + chunks = [] + async for event in stream: + phases.append(event.phase) + if event.phase == "delta": + chunks.append(event.data["text"]) + + assert phases[0] == "started" + assert phases[-1] == "completed" + assert "".join(chunks) == "Echo: hi" + + @pytest.mark.asyncio + async def test_plugin_harness_dispatches_v4_sample_plugin(self): + """PluginHarness should run the maintained v4 sample against the local mock core.""" + harness = PluginHarness( + LocalRuntimeConfig(plugin_dir=Path("test_plugin/new")), + ) + + async with harness: + records = await harness.dispatch_text("hello") + + assert [item.text for item in records if item.kind == "text"] == [ + "Echo: hello", + "Echo: stream", + ] + + @pytest.mark.asyncio + async def test_plugin_harness_can_invoke_plugin_capabilities(self): + """Harness should expose plugin-provided capabilities for local assertions.""" + harness = PluginHarness( + LocalRuntimeConfig(plugin_dir=Path("test_plugin/new")), + ) + + async with harness: + result = await harness.invoke_capability("demo.echo", {"text": "abc"}) + + assert result == { + "echo": "abc", + "plugin_id": "astrbot_plugin_v4demo", + } + + @pytest.mark.asyncio + async def test_plugin_harness_reuses_session_waiter_across_followups( + self, + tmp_path: Path, + ): + """Follow-up messages from the same session should be routed into the active waiter.""" + plugin_dir = tmp_path / "legacy_waiter" + plugin_dir.mkdir() + (plugin_dir / "main.py").write_text( + """ +from astrbot.core.utils.session_waiter import SessionController, session_waiter +from astrbot_sdk.api.components.command import CommandComponent +from astrbot_sdk.api.event import AstrMessageEvent, filter +from astrbot_sdk.api.message import MessageChain + + +class WaiterPlugin(CommandComponent): + @filter.command("ask") + async def ask(self, event: AstrMessageEvent): + await event.send(MessageChain().message("请输入确认内容")) + + @session_waiter(timeout=0.2) + async def waiter(controller: SessionController, ev: AstrMessageEvent): + await ev.send(MessageChain().message(f"收到:{ev.message_str}")) + controller.stop() + + await waiter(event) +""".strip(), + encoding="utf-8", + ) + + harness = PluginHarness( + LocalRuntimeConfig(plugin_dir=plugin_dir, platform="test"), + ) + + async with harness: + first = asyncio.create_task(harness.dispatch_text("ask")) + await asyncio.sleep(0.05) + follow_up = await harness.dispatch_text("确认") + await first + + assert [item.text for item in follow_up if item.kind == "text"] == ["收到:确认"] + assert [item.text for item in harness.sent_messages if item.kind == "text"] == [ + "请输入确认内容", + "收到:确认", + ] diff --git a/tests_v4/test_top_level_modules.py b/tests_v4/test_top_level_modules.py index 77479a0d7..f29cc3e34 100644 --- a/tests_v4/test_top_level_modules.py +++ b/tests_v4/test_top_level_modules.py @@ -37,6 +37,7 @@ from astrbot_sdk.runtime.transport import ( WebSocketServerTransport, ) from astrbot_sdk.star import Star +from astrbot_sdk.testing import _PluginLoadError TOP_LEVEL_MODULES = [ "astrbot_sdk", @@ -49,6 +50,7 @@ TOP_LEVEL_MODULES = [ "astrbot_sdk.events", "astrbot_sdk.runtime", "astrbot_sdk.star", + "astrbot_sdk.testing", ] @@ -214,6 +216,85 @@ class TestCliModule: entrypoint_mock.assert_called_once_with(**kwargs) asyncio_run_mock.assert_called_once_with(sentinel) + def test_dev_command_delegates_to_local_runtime(self): + """dev --local should delegate to the local harness entrypoint.""" + runner = CliRunner() + sentinel = object() + + with ( + patch( + "astrbot_sdk.cli._run_local_dev", + new=Mock(return_value=sentinel), + ) as dev_mock, + patch("astrbot_sdk.cli.asyncio.run") as asyncio_run_mock, + ): + result = runner.invoke( + cli, + [ + "dev", + "--plugin-dir", + "test_plugin/new", + "--local", + "--event-text", + "hello", + ], + ) + + assert result.exit_code == 0 + dev_mock.assert_called_once_with( + plugin_dir=Path("test_plugin/new"), + event_text="hello", + interactive=False, + session_id="local-session", + user_id="local-user", + platform="test", + group_id=None, + event_type="message", + ) + asyncio_run_mock.assert_called_once_with(sentinel) + + def test_dev_command_requires_local_mode(self): + """dev should reject invocations that do not opt into local mode.""" + runner = CliRunner() + + result = runner.invoke( + cli, + [ + "dev", + "--plugin-dir", + "test_plugin/new", + "--event-text", + "hello", + ], + ) + + assert result.exit_code == 2 + assert "--local/--standalone" in result.output + + def test_dev_command_maps_plugin_load_errors_to_exit_code_3(self): + """Known plugin load failures should render a friendly error and exit code 3.""" + runner = CliRunner() + + async def fail(*args, **kwargs): + raise _PluginLoadError("missing plugin") + + with patch("astrbot_sdk.cli._run_local_dev", new=fail): + result = runner.invoke( + cli, + [ + "dev", + "--plugin-dir", + "missing-plugin", + "--local", + "--event-text", + "hello", + ], + ) + + assert result.exit_code == 3 + assert "Error[plugin_load_error]" in result.output + assert "Suggestion:" in result.output + def test_main_module_invokes_cli_entrypoint(self): """Running astrbot_sdk.__main__ as a script should call cli().""" cli_mock = Mock()