From 1c4c9677bbf92b823888be5031471be2b6b7c481 Mon Sep 17 00:00:00 2001 From: whatevertogo Date: Fri, 13 Mar 2026 20:55:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=E3=80=81=E9=AA=8C=E8=AF=81=E5=92=8C?= =?UTF-8?q?=E6=9E=84=E5=BB=BA=E5=91=BD=E4=BB=A4=EF=BC=8C=E5=A2=9E=E5=BC=BA?= =?UTF-8?q?=20CLI=20=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ARCHITECTURE.md | 10 +- docs/v4/quickstart.md | 150 +++++++++++ src-new/astrbot/core/platform/register.py | 2 + src-new/astrbot_sdk/cli.py | 299 +++++++++++++++++++++- src-new/astrbot_sdk/runtime/loader.py | 76 +++--- src-new/astrbot_sdk/testing.py | 186 ++++++++++++++ tests_v4/test_testing_module.py | 24 ++ tests_v4/test_top_level_modules.py | 101 ++++++++ 8 files changed, 817 insertions(+), 31 deletions(-) create mode 100644 docs/v4/quickstart.md diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 45288976c..685790b77 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -279,13 +279,21 @@ from astrbot.core.utils.session_waiter import session_waiter 当前仓库已经提供一条受控的本地开发路径： -- CLI：`astr dev --local` 与 `astrbot-sdk dev --local` +- CLI： + - `astr dev --local` / `astrbot-sdk dev --local` + - `astrbot-sdk init` + - `astrbot-sdk validate` + - `astrbot-sdk build` - 稳定测试入口：`astrbot_sdk.testing` `astrbot_sdk.testing` 当前公开的稳定面包括： - `PluginHarness` - `LocalRuntimeConfig` +- `MockContext` +- `MockMessageEvent` +- `MockLLMClient` +- `MockPlatformClient` - `MockPeer` - `MockCapabilityRouter` - `InMemoryDB` diff --git a/docs/v4/quickstart.md b/docs/v4/quickstart.md new file mode 100644 index 000000000..62c118860 --- /dev/null +++ b/docs/v4/quickstart.md @@ -0,0 +1,150 @@ +# AstrBot SDK v4 Quickstart + +这份 quickstart 只覆盖当前已经落地的能力：`Star`、`Context`、`MessageEvent`、`astr dev --local`、`astrbot_sdk.testing`。 + +## 1. 创建插件目录 + +现在可以直接生成一个符合当前 loader 契约的骨架： + +```bash +astrbot-sdk init my-plugin +``` + +生成结果大致是： + +```text +my-plugin/ +├── plugin.yaml +├── requirements.txt +├── main.py +└── tests/ + └── test_plugin.py +``` + +如果你想手动创建，目录结构也至少应包含这些文件。`requirements.txt` 可以先留空。 + +## 2. 编写 `plugin.yaml` + +```yaml +name: my_plugin +display_name: My Plugin +desc: 我的第一个 AstrBot SDK v4 插件 +author: you +version: 0.1.0 +runtime: + python: "3.12" +components: + - class: main:MyPlugin +``` + +## 3. 编写 `main.py` + +```python +from astrbot_sdk import Context, MessageEvent, Star, on_command + + +class MyPlugin(Star): + @on_command("hello") + async def hello(self, event: MessageEvent, ctx: Context) -> None: + reply = await ctx.llm.chat("say hello") + await event.reply(reply) +``` + +## 4. 本地运行 + +安装当前仓库后，可以直接用本地 mock core 跑插件： + +```bash +astr dev --local --plugin-dir my-plugin --event-text "hello" +``` + +或者： + +```bash +astrbot-sdk dev --local --plugin-dir my-plugin --event-text "hello" +``` + +进入交互模式： + +```bash +astr dev --local --plugin-dir my-plugin --interactive +``` + +交互模式下支持这些元命令： + +- `/session ` 切换 session +- `/user ` 切换 user +- `/platform ` 切换 platform +- `/group ` 切换为群消息 +- `/private` 切回私聊 +- `/event ` 切换事件类型 +- `/exit` 退出 + +## 4.1 校验与打包 + +本地写完插件后，可以先做静态校验，再构建 zip 包： + +```bash +astrbot-sdk validate --plugin-dir my-plugin +astrbot-sdk build --plugin-dir my-plugin +``` + +默认构建产物会写到 `my-plugin/dist/`。 + +## 5. 直接写 handler 单元测试 + +如果你不想每次都起完整 harness，可以直接用 `MockContext` 和 `MockMessageEvent`： + +```python +import pytest + +from astrbot_sdk.testing import MockContext, MockMessageEvent + + +@pytest.mark.asyncio +async def test_hello_handler(): + ctx = MockContext(plugin_id="demo") + event = MockMessageEvent(text="hello", context=ctx) + ctx.llm.mock_response("你好！") + + async def handler(event, ctx): + text = await ctx.llm.chat("hello") + await event.reply(text) + + await handler(event, ctx) + + assert event.replies == ["你好！"] + ctx.platform.assert_sent("你好！") +``` + +## 6. 用 `PluginHarness` 跑真实插件 + +如果你想复用真实的 `loader` / `HandlerDispatcher` / compat 链路，用 `PluginHarness`： + +```python +import pytest +from pathlib import Path + +from astrbot_sdk.testing import LocalRuntimeConfig, PluginHarness + + +@pytest.mark.asyncio +async def test_plugin_directory(): + harness = PluginHarness( + LocalRuntimeConfig(plugin_dir=Path("my-plugin")), + ) + + async with harness: + records = await harness.dispatch_text("hello") + + assert any(item.text for item in records) +``` + +## 7. 当前边界 + +当前 quickstart 对应的是已经存在的能力，不包含这些后续项： + +- `ctx.http` / `ctx.cache` / `ctx.storage` / `ctx.i18n` +- 完整宿主调度下的 schedule 执行器 + +如果你需要查看当前架构与兼容边界，请看 [ARCHITECTURE.md](../../ARCHITECTURE.md)。 diff --git a/src-new/astrbot/core/platform/register.py b/src-new/astrbot/core/platform/register.py index 4ce287619..ba31e0e84 100644 --- a/src-new/astrbot/core/platform/register.py +++ b/src-new/astrbot/core/platform/register.py @@ -2,6 +2,8 @@ 旧版 ``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/cli.py b/src-new/astrbot_sdk/cli.py index a513406d0..05bc53e3e 100644 --- a/src-new/astrbot_sdk/cli.py +++ b/src-new/astrbot_sdk/cli.py @@ -3,10 +3,13 @@ from __future__ import annotations import asyncio +import re import sys import typing +import zipfile from collections.abc import Coroutine from pathlib import Path +from textwrap import dedent from typing import Any import click @@ -14,6 +17,7 @@ from loguru import logger from .errors import AstrBotError from .runtime.bootstrap import run_plugin_worker, run_supervisor, run_websocket_server +from .runtime.loader import load_plugin, load_plugin_spec, validate_plugin_spec from .testing import ( LocalRuntimeConfig, PluginHarness, @@ -28,6 +32,23 @@ EXIT_USAGE = 2 EXIT_PLUGIN_LOAD = 3 EXIT_RUNTIME = 4 EXIT_PLUGIN_EXECUTION = 5 +BUILD_EXCLUDED_DIRS = { + ".git", + ".idea", + ".mypy_cache", + ".pytest_cache", + ".ruff_cache", + ".venv", + "__pycache__", + "dist", +} +BUILD_EXCLUDED_FILES = { + ".astrbot-worker-state.json", +} + + +class _CliPluginValidationError(RuntimeError): + """CLI 侧的插件结构或打包校验失败。""" def setup_logger(verbose: bool = False) -> None: @@ -65,6 +86,30 @@ def _run_async_entrypoint( raise SystemExit(exit_code) from exc +def _run_sync_entrypoint( + entrypoint: typing.Callable[[], object], + *, + log_message: str, + log_level: str = "info", + context: dict[str, Any] | None = None, +) -> None: + log_method = getattr(logger, log_level) + log_method(log_message) + try: + 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 ( @@ -74,7 +119,13 @@ def _classify_cli_exception(exc: Exception) -> tuple[int, str, str]: ) if isinstance( exc, - (_PluginLoadError, FileNotFoundError, ImportError, ModuleNotFoundError), + ( + _CliPluginValidationError, + _PluginLoadError, + FileNotFoundError, + ImportError, + ModuleNotFoundError, + ), ): return ( EXIT_PLUGIN_LOAD, @@ -212,6 +263,201 @@ def _handle_dev_meta_command(command: str, state: dict[str, Any]) -> bool: return False +def _slugify_plugin_name(value: str) -> str: + slug = re.sub(r"[^a-zA-Z0-9]+", "_", value).strip("_").lower() + return slug or "my_plugin" + + +def _class_name_for_plugin(value: str) -> str: + parts = [part for part in re.split(r"[^a-zA-Z0-9]+", value) if part] + if not parts: + return "MyPlugin" + return "".join(part[:1].upper() + part[1:] for part in parts) + + +def _sanitize_build_part(value: str) -> str: + sanitized = re.sub(r"[^a-zA-Z0-9._-]+", "_", value).strip("._-") + return sanitized or "artifact" + + +def _render_init_plugin_yaml(*, plugin_name: str, display_name: str) -> str: + python_version = f"{sys.version_info.major}.{sys.version_info.minor}" + class_name = _class_name_for_plugin(plugin_name) + return dedent( + f"""\ + name: {plugin_name} + display_name: {display_name} + desc: 使用 AstrBot SDK 创建的插件 + author: your-name + version: 0.1.0 + runtime: + python: "{python_version}" + components: + - class: main:{class_name} + """ + ) + + +def _render_init_main_py(*, plugin_name: str) -> str: + class_name = _class_name_for_plugin(plugin_name) + return dedent( + f"""\ + from astrbot_sdk import Context, MessageEvent, Star, on_command + + + class {class_name}(Star): + @on_command("hello") + async def hello(self, event: MessageEvent, ctx: Context) -> None: + await event.reply("Hello, World!") + """ + ) + + +def _render_init_test_py(*, plugin_name: str) -> str: + class_name = _class_name_for_plugin(plugin_name) + return dedent( + f'''\ + import pytest + + from astrbot_sdk.testing import MockContext, MockMessageEvent + from main import {class_name} + + + @pytest.mark.asyncio + async def test_hello_handler(): + plugin = {class_name}() + ctx = MockContext(plugin_id="{plugin_name}") + event = MockMessageEvent(text="/hello", context=ctx) + + await plugin.hello(event, ctx) + + assert event.replies == ["Hello, World!"] + ctx.platform.assert_sent("Hello, World!") + ''' + ) + + +def _ensure_plugin_dir_exists(plugin_dir: Path) -> Path: + resolved = plugin_dir.resolve() + if not resolved.exists() or not resolved.is_dir(): + raise _CliPluginValidationError(f"插件目录不存在：{plugin_dir}") + return resolved + + +def _load_validated_plugin(plugin_dir: Path) -> tuple[Any, Any]: + resolved_dir = _ensure_plugin_dir_exists(plugin_dir) + plugin = load_plugin_spec(resolved_dir) + try: + validate_plugin_spec(plugin) + except ValueError as exc: + raise _CliPluginValidationError(str(exc)) from exc + + loaded = load_plugin(plugin) + if not loaded.instances: + raise _CliPluginValidationError( + "未找到可加载的组件，请检查 plugin.yaml 中的 components" + ) + return plugin, loaded + + +def _build_kind(plugin: Any) -> str: + return ( + "legacy-main" + if bool(plugin.manifest_data.get("__legacy_main__")) + else "plugin-yaml" + ) + + +def _path_is_within(path: Path, root: Path) -> bool: + try: + path.resolve().relative_to(root.resolve()) + except ValueError: + return False + return True + + +def _iter_build_files(plugin_dir: Path, output_dir: Path) -> list[Path]: + files: list[Path] = [] + for path in sorted(plugin_dir.rglob("*")): + if path.is_dir(): + continue + if _path_is_within(path, output_dir): + continue + relative = path.relative_to(plugin_dir) + if any(part in BUILD_EXCLUDED_DIRS for part in relative.parts[:-1]): + continue + if relative.name in BUILD_EXCLUDED_FILES: + continue + if path.suffix in {".pyc", ".pyo"}: + continue + files.append(path) + return files + + +def _init_plugin(name: str) -> None: + target_dir = Path(name) + if target_dir.exists(): + raise _CliPluginValidationError(f"目标目录已存在：{target_dir}") + + plugin_name = _slugify_plugin_name(target_dir.name) + display_name = target_dir.name + target_dir.mkdir(parents=True, exist_ok=False) + (target_dir / "tests").mkdir() + (target_dir / "plugin.yaml").write_text( + _render_init_plugin_yaml( + plugin_name=plugin_name, + display_name=display_name, + ), + encoding="utf-8", + ) + (target_dir / "requirements.txt").write_text("", encoding="utf-8") + (target_dir / "main.py").write_text( + _render_init_main_py(plugin_name=plugin_name), + encoding="utf-8", + ) + (target_dir / "tests" / "test_plugin.py").write_text( + _render_init_test_py(plugin_name=plugin_name), + encoding="utf-8", + ) + click.echo(f"已创建插件骨架：{target_dir}") + click.echo("后续命令：") + click.echo(f" astrbot-sdk validate --plugin-dir {target_dir}") + click.echo( + f" astrbot-sdk dev --local --plugin-dir {target_dir} --event-text hello" + ) + + +def _validate_plugin(plugin_dir: Path) -> None: + plugin, loaded = _load_validated_plugin(plugin_dir) + click.echo(f"校验通过：{plugin.name}") + click.echo(f"kind: {_build_kind(plugin)}") + click.echo(f"plugin_dir: {plugin.plugin_dir}") + click.echo(f"handlers: {len(loaded.handlers)}") + click.echo(f"capabilities: {len(loaded.capabilities)}") + click.echo(f"instances: {len(loaded.instances)}") + + +def _build_plugin(plugin_dir: Path, output_dir: Path | None) -> None: + plugin, _ = _load_validated_plugin(plugin_dir) + build_dir = (output_dir or (plugin.plugin_dir / "dist")).resolve() + build_dir.mkdir(parents=True, exist_ok=True) + + version = _sanitize_build_part(str(plugin.manifest_data.get("version") or "0.0.0")) + archive_name = f"{_sanitize_build_part(plugin.name)}-{version}.zip" + archive_path = build_dir / archive_name + + with zipfile.ZipFile( + archive_path, + mode="w", + compression=zipfile.ZIP_DEFLATED, + ) as archive: + for path in _iter_build_files(plugin.plugin_dir, build_dir): + archive.write(path, arcname=path.relative_to(plugin.plugin_dir)) + + click.echo(f"构建完成：{archive_path}") + click.echo(f"artifact: {archive_path}") + + @click.group() @click.option("-v", "--verbose", is_flag=True, help="Enable verbose output") @click.pass_context @@ -238,6 +484,57 @@ def run(plugins_dir: Path) -> None: ) +@cli.command() +@click.argument("name", type=str) +def init(name: str) -> None: + """Create a new plugin skeleton in the target directory.""" + _run_sync_entrypoint( + lambda: _init_plugin(name), + log_message=f"创建插件骨架：{name}", + context={"target": Path(name)}, + ) + + +@cli.command() +@click.option( + "--plugin-dir", + default=".", + show_default=True, + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + help="Plugin directory to validate", +) +def validate(plugin_dir: Path) -> None: + """Validate plugin manifest, imports and handler discovery.""" + _run_sync_entrypoint( + lambda: _validate_plugin(plugin_dir), + log_message=f"校验插件目录：{plugin_dir}", + context={"plugin_dir": plugin_dir}, + ) + + +@cli.command() +@click.option( + "--plugin-dir", + default=".", + show_default=True, + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + help="Plugin directory to package", +) +@click.option( + "--output-dir", + default=None, + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + help="Directory for the build artifact, defaults to /dist", +) +def build(plugin_dir: Path, output_dir: Path | None) -> None: + """Validate and package a plugin into a zip artifact.""" + _run_sync_entrypoint( + lambda: _build_plugin(plugin_dir, output_dir), + log_message=f"构建插件包：{plugin_dir}", + context={"plugin_dir": plugin_dir, "output_dir": output_dir}, + ) + + @cli.command() @click.option( "--plugin-dir", diff --git a/src-new/astrbot_sdk/runtime/loader.py b/src-new/astrbot_sdk/runtime/loader.py index 9c41a7117..fd012305a 100644 --- a/src-new/astrbot_sdk/runtime/loader.py +++ b/src-new/astrbot_sdk/runtime/loader.py @@ -525,6 +525,38 @@ def load_plugin_spec(plugin_dir: Path) -> PluginSpec: ) +def validate_plugin_spec(plugin: PluginSpec) -> None: + """校验单个插件规范，供 CLI 和发现流程复用。""" + manifest_data = plugin.manifest_data + is_legacy_main = bool(manifest_data.get(LEGACY_MAIN_MANIFEST_KEY)) + + if not is_legacy_main and not plugin.requirements_path.exists(): + raise ValueError("missing requirements.txt") + + raw_name = manifest_data.get("name") + if not is_legacy_main and (not isinstance(raw_name, str) or not raw_name): + raise ValueError("plugin name is required") + + raw_runtime = manifest_data.get("runtime") or {} + raw_python = raw_runtime.get("python") + if not is_legacy_main and (not isinstance(raw_python, str) or not raw_python): + raise ValueError("runtime.python is required") + + components = manifest_data.get("components") + if not isinstance(components, list): + raise ValueError("components must be a list") + + if is_legacy_main: + return + + for index, component in enumerate(components): + if not isinstance(component, dict): + raise ValueError(f"components[{index}] must be an object") + class_path = component.get("class") + if not isinstance(class_path, str) or ":" not in class_path: + raise ValueError(f"components[{index}].class must be ':'") + + def discover_plugins(plugins_dir: Path) -> PluginDiscoveryResult: plugins_root = plugins_dir.resolve() skipped_plugins: dict[str, str] = {} @@ -538,47 +570,33 @@ def discover_plugins(plugins_dir: Path) -> PluginDiscoveryResult: if not entry.is_dir() or entry.name.startswith("."): continue manifest_path = entry / PLUGIN_MANIFEST_FILE - requirements_path = entry / "requirements.txt" if not manifest_path.exists() and not _looks_like_legacy_plugin(entry): continue - if manifest_path.exists() and not requirements_path.exists(): - skipped_plugins[entry.name] = "missing requirements.txt" - continue + plugin: PluginSpec | None = None try: - if manifest_path.exists(): - manifest_data = _read_yaml(manifest_path) - else: - manifest_path, manifest_data = _build_legacy_manifest(entry) + plugin = load_plugin_spec(entry) + validate_plugin_spec(plugin) except Exception as exc: - skipped_plugins[entry.name] = f"failed to parse plugin manifest: {exc}" + skip_key = entry.name + if plugin is not None: + raw_name = plugin.manifest_data.get("name") + if ( + isinstance(raw_name, str) + and raw_name + and str(exc) != "missing requirements.txt" + ): + skip_key = raw_name + skipped_plugins[skip_key] = f"failed to parse plugin manifest: {exc}" continue - plugin_name = manifest_data.get("name") - runtime = manifest_data.get("runtime") or {} - python_version = runtime.get("python") - components = manifest_data.get("components") + plugin_name = plugin.name if not isinstance(plugin_name, str) or not plugin_name: skipped_plugins[entry.name] = "plugin name is required" continue if plugin_name in seen_names: skipped_plugins[plugin_name] = "duplicate plugin name" continue - if not isinstance(components, list): - skipped_plugins[plugin_name] = "components must be a list" - continue - if not isinstance(python_version, str) or not python_version: - skipped_plugins[plugin_name] = "runtime.python is required" - continue seen_names.add(plugin_name) - plugins.append( - PluginSpec( - name=plugin_name, - plugin_dir=entry.resolve(), - manifest_path=manifest_path.resolve(), - requirements_path=requirements_path.resolve(), - python_version=python_version, - manifest_data=manifest_data, - ) - ) + plugins.append(plugin) return PluginDiscoveryResult(plugins=plugins, skipped_plugins=skipped_plugins) diff --git a/src-new/astrbot_sdk/testing.py b/src-new/astrbot_sdk/testing.py index 054817cf4..42c39aee0 100644 --- a/src-new/astrbot_sdk/testing.py +++ b/src-new/astrbot_sdk/testing.py @@ -188,15 +188,83 @@ class InMemoryMemory: return results +class MockLLMClient: + """在真实 LLMClient 之上补一层测试控制能力。""" + + def __init__(self, client: Any, router: "MockCapabilityRouter") -> None: + self._client = client + self._router = router + + def mock_response(self, text: str) -> None: + self._router.enqueue_llm_response(text) + + def mock_stream_response(self, text: str) -> None: + self._router.enqueue_llm_stream_response(text) + + def clear_mock_responses(self) -> None: + self._router.clear_llm_responses() + + def __getattr__(self, name: str) -> Any: + return getattr(self._client, name) + + +class MockPlatformClient: + """在真实 PlatformClient 之上补一层断言入口。""" + + def __init__(self, client: Any, sink: StdoutPlatformSink) -> None: + self._client = client + self._sink = sink + + @property + def records(self) -> list[RecordedSend]: + return list(self._sink.records) + + def assert_sent( + self, + expected_text: str | None = None, + *, + kind: str = "text", + count: int | None = None, + ) -> None: + matched = [item for item in self._sink.records if item.kind == kind] + if expected_text is not None: + matched = [item for item in matched if item.text == expected_text] + if count is not None: + if len(matched) != count: + raise AssertionError( + f"expected {count} sent records, got {len(matched)}: {matched}" + ) + return + if not matched: + raise AssertionError( + f"expected sent record kind={kind!r} text={expected_text!r}, got {self._sink.records}" + ) + + def __getattr__(self, name: str) -> Any: + return getattr(self._client, name) + + class MockCapabilityRouter(CapabilityRouter): """本地 mock core，直接复用已有的内建 capability 实现。""" def __init__(self, *, platform_sink: StdoutPlatformSink | None = None) -> None: self.platform_sink = platform_sink or StdoutPlatformSink() + self._llm_responses: list[str] = [] + self._llm_stream_responses: list[str] = [] super().__init__() self.db = InMemoryDB(self.db_store) self.memory = InMemoryMemory(self.memory_store) + def enqueue_llm_response(self, text: str) -> None: + self._llm_responses.append(text) + + def enqueue_llm_stream_response(self, text: str) -> None: + self._llm_stream_responses.append(text) + + def clear_llm_responses(self) -> None: + self._llm_responses.clear() + self._llm_stream_responses.clear() + async def execute( self, capability: str, @@ -206,6 +274,34 @@ class MockCapabilityRouter(CapabilityRouter): cancel_token, request_id: str, ) -> dict[str, Any] | StreamExecution: + if capability == "llm.chat": + return {"text": self._take_llm_response(str(payload.get("prompt", "")))} + if capability == "llm.chat_raw": + text = self._take_llm_response(str(payload.get("prompt", ""))) + return { + "text": text, + "usage": { + "input_tokens": len(str(payload.get("prompt", ""))), + "output_tokens": len(text), + }, + "finish_reason": "stop", + "tool_calls": [], + } + if capability == "llm.stream_chat": + text = self._take_llm_stream_response(str(payload.get("prompt", ""))) + + async def iterator() -> typing.AsyncIterator[dict[str, Any]]: + for char in text: + cancel_token.raise_if_cancelled() + await asyncio.sleep(0) + yield {"text": char} + + return StreamExecution( + iterator=iterator(), + finalize=lambda chunks: { + "text": "".join(item.get("text", "") for item in chunks) + }, + ) before = len(self.sent_messages) result = await super().execute( capability, @@ -221,6 +317,18 @@ class MockCapabilityRouter(CapabilityRouter): for payload in self.sent_messages[start_index:]: self.platform_sink.record(RecordedSend.from_payload(payload)) + def _take_llm_response(self, prompt: str) -> str: + if self._llm_responses: + return self._llm_responses.pop(0) + return f"Echo: {prompt}" + + def _take_llm_stream_response(self, prompt: str) -> str: + if self._llm_stream_responses: + return self._llm_stream_responses.pop(0) + if self._llm_responses: + return self._llm_responses.pop(0) + return f"Echo: {prompt}" + class MockPeer: """满足 `Context`/`CapabilityProxy` 需要的最小 peer。""" @@ -300,6 +408,80 @@ class MockPeer: return f"local_{self._counter:04d}" +class MockContext(RuntimeContext): + """直接用于 handler 单元测试的轻量运行时上下文。""" + + def __init__( + self, + *, + plugin_id: str = "test-plugin", + logger: Any | None = None, + cancel_token: CancelToken | None = None, + platform_sink: StdoutPlatformSink | None = None, + ) -> None: + self.platform_sink = platform_sink or StdoutPlatformSink() + self.router = MockCapabilityRouter(platform_sink=self.platform_sink) + self.mock_peer = MockPeer(self.router) + super().__init__( + peer=self.mock_peer, + plugin_id=plugin_id, + cancel_token=cancel_token, + logger=logger, + ) + self.llm = MockLLMClient(self.llm, self.router) + self.platform = MockPlatformClient(self.platform, self.platform_sink) + + @property + def sent_messages(self) -> list[RecordedSend]: + return list(self.platform_sink.records) + + +class MockMessageEvent(MessageEvent): + """直接用于 handler 单元测试的轻量消息事件。""" + + def __init__( + self, + *, + text: str = "", + user_id: str | None = "test-user", + group_id: str | None = None, + platform: str | None = "test", + session_id: str | None = "test-session", + raw: dict[str, Any] | None = None, + context: MockContext | None = None, + ) -> None: + self.replies: list[str] = [] + super().__init__( + text=text, + user_id=user_id, + group_id=group_id, + platform=platform, + session_id=session_id, + raw=raw, + context=context, + ) + if context is not None: + self.bind_runtime_reply(context) + elif self._reply_handler is None: + self.bind_reply_handler(self._capture_reply) + + @property + def is_private(self) -> bool: + return self.group_id is None + + def bind_runtime_reply(self, context: MockContext) -> None: + self._context = context + + async def reply(text: str) -> None: + self.replies.append(text) + await context.platform.send(self.session_ref or self.session_id, text) + + self.bind_reply_handler(reply) + + async def _capture_reply(self, text: str) -> None: + self.replies.append(text) + + @dataclass(slots=True) class LocalRuntimeConfig: """本地 harness 的稳定配置对象。""" @@ -848,7 +1030,11 @@ __all__ = [ "InMemoryMemory", "LocalRuntimeConfig", "MockCapabilityRouter", + "MockContext", + "MockLLMClient", + "MockMessageEvent", "MockPeer", + "MockPlatformClient", "PluginHarness", "RecordedSend", "StdoutPlatformSink", diff --git a/tests_v4/test_testing_module.py b/tests_v4/test_testing_module.py index 38da1f911..19870656a 100644 --- a/tests_v4/test_testing_module.py +++ b/tests_v4/test_testing_module.py @@ -11,6 +11,8 @@ import astrbot_sdk.testing as testing_module from astrbot_sdk.testing import ( LocalRuntimeConfig, MockCapabilityRouter, + MockContext, + MockMessageEvent, MockPeer, PluginHarness, ) @@ -26,7 +28,11 @@ class TestTestingModule: "InMemoryMemory", "LocalRuntimeConfig", "MockCapabilityRouter", + "MockContext", + "MockLLMClient", + "MockMessageEvent", "MockPeer", + "MockPlatformClient", "PluginHarness", "RecordedSend", "StdoutPlatformSink", @@ -84,6 +90,24 @@ class TestTestingModule: "plugin_id": "astrbot_plugin_v4demo", } + @pytest.mark.asyncio + async def test_mock_context_and_event_support_direct_handler_unit_tests(self): + """MockContext/MockMessageEvent should support direct handler tests without a full harness.""" + ctx = MockContext(plugin_id="demo-test") + event = MockMessageEvent(text="hello", context=ctx) + ctx.llm.mock_response("你好！") + + async def handler(mock_event, mock_ctx): + reply = await mock_ctx.llm.chat("hello") + await mock_event.reply(reply) + return mock_event.plain_result("done") + + result = await handler(event, ctx) + + assert result.text == "done" + assert event.replies == ["你好！"] + ctx.platform.assert_sent("你好！") + @pytest.mark.asyncio async def test_plugin_harness_reuses_session_waiter_across_followups( self, diff --git a/tests_v4/test_top_level_modules.py b/tests_v4/test_top_level_modules.py index f29cc3e34..0d465d86d 100644 --- a/tests_v4/test_top_level_modules.py +++ b/tests_v4/test_top_level_modules.py @@ -7,6 +7,7 @@ from __future__ import annotations import importlib import runpy import sys +import zipfile from pathlib import Path from unittest.mock import AsyncMock, Mock, patch @@ -39,6 +40,7 @@ from astrbot_sdk.runtime.transport import ( from astrbot_sdk.star import Star from astrbot_sdk.testing import _PluginLoadError +REPO_ROOT = Path(__file__).resolve().parents[1] TOP_LEVEL_MODULES = [ "astrbot_sdk", "astrbot_sdk._legacy_api", @@ -295,6 +297,105 @@ class TestCliModule: assert "Error[plugin_load_error]" in result.output assert "Suggestion:" in result.output + def test_init_command_creates_plugin_skeleton(self): + """init should generate a loader-compatible plugin skeleton.""" + runner = CliRunner() + + with runner.isolated_filesystem(): + result = runner.invoke(cli, ["init", "demo-plugin"]) + + plugin_dir = Path("demo-plugin") + manifest = (plugin_dir / "plugin.yaml").read_text(encoding="utf-8") + main_file = (plugin_dir / "main.py").read_text(encoding="utf-8") + test_file = (plugin_dir / "tests" / "test_plugin.py").read_text( + encoding="utf-8" + ) + + assert result.exit_code == 0 + assert "已创建插件骨架" in result.output + assert "name: demo_plugin" in manifest + assert ( + f'python: "{sys.version_info.major}.{sys.version_info.minor}"' in manifest + ) + assert "class DemoPlugin(Star):" in main_file + assert "MockContext" in test_file + assert "MockMessageEvent" in test_file + + def test_validate_command_checks_real_plugin_fixture(self): + """validate should reuse loader-based discovery against a real v4 fixture.""" + runner = CliRunner() + + result = runner.invoke( + cli, + [ + "validate", + "--plugin-dir", + str(REPO_ROOT / "test_plugin" / "new"), + ], + ) + + assert result.exit_code == 0 + assert "校验通过：astrbot_plugin_v4demo" in result.output + assert "handlers:" in result.output + assert "capabilities:" in result.output + + def test_validate_command_maps_invalid_component_to_exit_code_3(self): + """validate should fail with a friendly plugin-load error on broken manifests.""" + runner = CliRunner() + + with runner.isolated_filesystem(): + plugin_dir = Path("broken-plugin") + plugin_dir.mkdir() + (plugin_dir / "requirements.txt").write_text("", encoding="utf-8") + (plugin_dir / "plugin.yaml").write_text( + "\n".join( + [ + "name: broken_plugin", + "runtime:", + ' python: "3.12"', + "components:", + " - class: broken", + ] + ), + encoding="utf-8", + ) + + result = runner.invoke(cli, ["validate", "--plugin-dir", str(plugin_dir)]) + + assert result.exit_code == 3 + assert "Error[plugin_load_error]" in result.output + assert "components[0].class" in result.output + + def test_build_command_creates_zip_artifact(self): + """build should validate first and then package the plugin directory into a zip.""" + runner = CliRunner() + + with runner.isolated_filesystem(): + init_result = runner.invoke(cli, ["init", "buildable-plugin"]) + assert init_result.exit_code == 0 + + result = runner.invoke( + cli, + [ + "build", + "--plugin-dir", + "buildable-plugin", + ], + ) + + artifact_dir = Path("buildable-plugin") / "dist" + artifacts = sorted(artifact_dir.glob("*.zip")) + assert len(artifacts) == 1 + with zipfile.ZipFile(artifacts[0]) as archive: + names = set(archive.namelist()) + + assert result.exit_code == 0 + assert "构建完成：" in result.output + assert "plugin.yaml" in names + assert "main.py" in names + assert "requirements.txt" in names + assert "tests/test_plugin.py" in names + def test_main_module_invokes_cli_entrypoint(self): """Running astrbot_sdk.__main__ as a script should call cli().""" cli_mock = Mock()