mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
feat: 添加插件初始化、验证和构建命令，增强 CLI 功能
This commit is contained in:
whatevertogo
@@ -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`
|
||||
|
||||
150
docs/v4/quickstart.md
Normal file
150
docs/v4/quickstart.md
Normal file
@@ -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 <id>` 切换 session
|
||||
- `/user <id>` 切换 user
|
||||
- `/platform <name>` 切换 platform
|
||||
- `/group <id>` 切换为群消息
|
||||
- `/private` 切回私聊
|
||||
- `/event <type>` 切换事件类型
|
||||
- `/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)。
|
||||
@@ -2,6 +2,8 @@
|
||||
旧版 ``astrbot.core.platform.register`` 导入路径兼容入口。
|
||||
TODO: 目前仅保留符号以兼容导入，后续如果需要，可以在这里实现一个基于当前平台能力的注册系统适配器。
|
||||
"""
|
||||
|
||||
|
||||
def register_platform_adapter(*args, **kwargs):
|
||||
raise NotImplementedError(
|
||||
"astrbot.core.platform.register_platform_adapter() 尚未在 v4 兼容层实现。"
|
||||
|
||||
@@ -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 <plugin-dir>/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",
|
||||
|
||||
@@ -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 '<module>:<Class>'")
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user