fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-18 02:00:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 添加测试框架和相关测试用例，涵盖 API 装饰器、事件、上下文和传输通信

Browse Source
This commit is contained in:
whatevertogo
2026-03-13 00:29:12 +08:00
parent c3ccc2b5da
commit 9bf831810d
12 changed files with 1217 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
AGENTS.md
View File

@@ -4,3 +4,25 @@
- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
- 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
# 开发命令
## 格式化与检查
在提交代码前，请依次运行以下命令：
```bash
pyclean . # 清理 Python 缓存文件
ruff format . # 使用 ruff 格式化代码
ruff check . --fix # 使用 ruff 检查并自动修复问题
```
## 测试
```bash
python run_tests.py # 运行所有测试
python run_tests.py -v # 详细输出
python run_tests.py -k "test_peer" # 运行匹配模式的测试
python run_tests.py --cov # 运行测试并生成覆盖率报告
```

22
CLAUDE.md
View File

@@ -4,3 +4,25 @@
- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
- 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
# 开发命令
## 格式化与检查
在提交代码前，请依次运行以下命令：
```bash
pyclean . # 清理 Python 缓存文件
ruff format . # 使用 ruff 格式化代码
ruff check . --fix # 使用 ruff 检查并自动修复问题
```
## 测试
```bash
python run_tests.py # 运行所有测试
python run_tests.py -v # 详细输出
python run_tests.py -k "test_peer" # 运行匹配模式的测试
python run_tests.py --cov # 运行测试并生成覆盖率报告
```

55
pyproject.toml
View File

@@ -29,3 +29,58 @@ package-dir = {"" = "src-new"}
[tool.setuptools.packages.find]
where = ["src-new"]
# ============================================================
# Pytest Configuration
# ============================================================
[project.optional-dependencies]
test = [
"pytest>=8.0.0",
"pytest-asyncio>=0.24.0",
"pytest-cov>=5.0.0",
]
[tool.pytest.ini_options]
testpaths = ["tests_v4"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
addopts = [
"-v",
"--tb=short",
"--strict-markers",
]
markers = [
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
"integration: marks tests as integration tests",
"unit: marks tests as unit tests",
]
filterwarnings = [
"ignore::DeprecationWarning",
"ignore::PendingDeprecationWarning",
]
# ============================================================
# Coverage Configuration
# ============================================================
[tool.coverage.run]
source = ["src-new/astrbot_sdk"]
branch = true
omit = [
"*/tests/*",
"*/__pycache__/*",
"*/_legacy_api.py",
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise AssertionError",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"if TYPE_CHECKING:",
]
show_missing = true

53
run_tests.py Normal file
View File

@@ -0,0 +1,53 @@
#!/usr/bin/env python
"""
Test runner script for astrbot-sdk.
Usage:
python run_tests.py # Run all tests
python run_tests.py -v # Verbose output
python run_tests.py -k "test_peer" # Run tests matching pattern
python run_tests.py --cov # Run with coverage
python run_tests.py -m "not slow" # Skip slow tests
"""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
def main() -> int:
"""Run tests with pytest."""
project_root = Path(__file__).parent
tests_dir = project_root / "tests_v4"
# Build pytest command
cmd = [sys.executable, "-m", "pytest", str(tests_dir)]
# Parse arguments
args = sys.argv[1:]
# Handle --cov flag
if "--cov" in args:
args.remove("--cov")
cmd.extend([
"--cov=src-new/astrbot_sdk",
"--cov-report=term-missing",
"--cov-report=html:.htmlcov",
])
# Default flags if no specific args
if not args:
cmd.extend(["-v", "--tb=short"])
cmd.extend(args)
print(f"Running: {' '.join(cmd)}")
print("-" * 60)
result = subprocess.run(cmd, cwd=project_root)
return result.returncode
if __name__ == "__main__":
sys.exit(main())

171
tests_v4/README.md Normal file
View File

@@ -0,0 +1,171 @@
# AstrBot SDK Test Framework
## Overview
This test suite uses **pytest** with `pytest-asyncio` for testing the AstrBot SDK v4 implementation.
## Test Structure
```
tests_v4/
├── conftest.py # Shared fixtures and configuration
├── pytest.ini # Pytest configuration
├── test_api_contract.py # API contract tests
├── test_api_decorators.py # Decorator and Star class tests
├── test_context.py # Context and CancelToken tests
├── test_entrypoints.py # CLI entrypoint tests (requires installation)
├── test_events.py # MessageEvent and PlainTextResult tests
├── test_legacy_adapter.py # Legacy API compatibility tests
├── test_peer.py # Peer communication tests
├── test_protocol.py # Protocol message tests
├── test_runtime.py # Supervisor/Worker runtime tests
├── test_script_migrations.py # Migration script tests
└── test_supervisor_migration.py # Supervisor migration tests
```
## Running Tests
### All Tests
```bash
# Using the runner script
python run_tests.py
# Or directly with pytest
python -m pytest tests_v4/ -v
```
### Specific Tests
```bash
# Run specific file
python -m pytest tests_v4/test_peer.py -v
# Run specific test class
python -m pytest tests_v4/test_peer.py::PeerRuntimeTest -v
# Run specific test
python -m pytest tests_v4/test_peer.py::PeerRuntimeTest::test_initialize_and_call_builtin_capabilities -v
# Run tests matching pattern
python -m pytest tests_v4/ -k "peer" -v
```
### With Coverage
```bash
python run_tests.py --cov
# Or directly
python -m pytest tests_v4/ --cov=src-new/astrbot_sdk --cov-report=term-missing
```
### Skip Slow Tests
```bash
python -m pytest tests_v4/ -m "not slow"
```
### Integration Tests Only
```bash
python -m pytest tests_v4/ -m integration
```
## Test Markers
| Marker | Description |
|--------|-------------|
| `@pytest.mark.unit` | Unit tests (fast, no external dependencies) |
| `@pytest.mark.integration` | Integration tests (may require setup) |
| `@pytest.mark.slow` | Slow tests (can be skipped with `-m "not slow"`) |
## Available Fixtures
The `conftest.py` provides these fixtures:
### Transport Fixtures
- `transport_pair`: Creates a connected pair of in-memory transports for testing peer communication
- `core_peer`: Creates a core peer with default handlers
- `plugin_peer`: Creates a plugin peer connected to core_peer
### Helper Fixtures
- `fake_env_manager`: Provides a fake environment manager for testing
- `temp_plugin_dir`: Creates a temporary directory for plugin testing
- `test_plugin`: Creates a minimal test plugin
### Usage Example
```python
async def test_my_feature(core_peer, plugin_peer):
"""Test using pytest fixtures."""
await plugin_peer.initialize([])
result = await plugin_peer.invoke("llm.chat", {"prompt": "hello"})
assert result["text"] == "Echo: hello"
```
## Writing New Tests
### Test File Naming
- Test files should start with `test_`
- Test classes should start with `Test`
- Test functions should start with `test_`
### Async Tests
Use `@pytest.mark.asyncio` or rely on auto mode:
```python
# Both work due to asyncio_mode = auto
async def test_async_auto():
await asyncio.sleep(0)
@pytest.mark.asyncio
async def test_async_explicit():
await asyncio.sleep(0)
```
### Using Fixtures
```python
def test_with_fixture(transport_pair):
left, right = transport_pair
# Use transports...
async def test_async_fixture(core_peer):
# core_peer is already started
await core_peer.invoke(...)
```
## Dependencies
Install test dependencies:
```bash
pip install pytest pytest-asyncio pytest-cov
```
Or use the optional dependency group:
```bash
pip install -e ".[test]"
```
## Test Categories
### Unit Tests (Fast)
- `test_context.py` - CancelToken and Context tests
- `test_events.py` - MessageEvent and PlainTextResult tests
- `test_api_decorators.py` - Decorator tests
- `test_protocol.py` - Protocol message tests
### Integration Tests (Slower)
- `test_peer.py` - Peer communication with real transports
- `test_runtime.py` - Supervisor/Worker process tests
- `test_legacy_adapter.py` - Legacy API compatibility
- `test_script_migrations.py` - Migration script tests

210
tests_v4/conftest.py Normal file
View File

@@ -0,0 +1,210 @@
# Test configuration
import asyncio
from collections.abc import Generator
from pathlib import Path
from typing import Any
import pytest
# 将 src-new 加入路径 - 这使得测试可以运行，但不算"已安装"
import sys
SRC_NEW_PATH = str(Path(__file__).parent.parent / "src-new")
sys.path.insert(0, SRC_NEW_PATH)
# ============================================================
# Async Configuration
# ============================================================
@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
"""Create an event loop for async tests."""
policy = asyncio.get_event_loop_policy()
loop = policy.new_event_loop()
yield loop
loop.close()
# ============================================================
# Transport Fixtures
# ============================================================
class MemoryTransport:
"""In-memory transport for testing peer communication."""
def __init__(self) -> None:
self._closed = asyncio.Event()
self._message_handler = None
self.partner: MemoryTransport | None = None
def set_message_handler(self, handler) -> None:
self._message_handler = handler
async def start(self) -> None:
self._closed.clear()
async def stop(self) -> None:
self._closed.set()
async def wait_closed(self) -> None:
await self._closed.wait()
async def send(self, payload: str) -> None:
if self.partner is None:
raise RuntimeError("MemoryTransport 未连接 partner")
if self.partner._message_handler is not None:
await self.partner._message_handler(payload)
async def _dispatch(self, payload: str) -> None:
if self._message_handler is not None:
await self._message_handler(payload)
@pytest.fixture
def transport_pair() -> tuple[MemoryTransport, MemoryTransport]:
"""Create a connected pair of in-memory transports."""
left = MemoryTransport()
right = MemoryTransport()
left.partner = right
right.partner = left
return left, right
# ============================================================
# Mock/Fake Fixtures
# ============================================================
class FakeEnvManager:
"""Fake environment manager for testing."""
def prepare_environment(self, _plugin: Any) -> Path:
return Path(sys.executable)
@pytest.fixture
def fake_env_manager() -> FakeEnvManager:
"""Provide a fake environment manager."""
return FakeEnvManager()
# ============================================================
# Peer Fixtures
# ============================================================
from astrbot_sdk.protocol.messages import InitializeOutput, PeerInfo
from astrbot_sdk.runtime.capability_router import CapabilityRouter
from astrbot_sdk.runtime.peer import Peer
@pytest.fixture
async def core_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer:
"""Create a core peer with default handlers."""
left, _ = transport_pair
router = CapabilityRouter()
peer = Peer(
transport=left,
peer_info=PeerInfo(name="core", role="core", version="v4"),
)
async def init_handler(_message) -> InitializeOutput:
return InitializeOutput(
peer=PeerInfo(name="core", role="core", version="v4"),
capabilities=router.descriptors(),
metadata={},
)
async def invoke_handler(message, token):
return await router.execute(
message.capability,
message.input,
stream=message.stream,
cancel_token=token,
request_id=message.id,
)
peer.set_initialize_handler(init_handler)
peer.set_invoke_handler(invoke_handler)
await peer.start()
yield peer
await peer.stop()
@pytest.fixture
async def plugin_peer(transport_pair: tuple[MemoryTransport, MemoryTransport]) -> Peer:
"""Create a plugin peer connected to core."""
_, right = transport_pair
peer = Peer(
transport=right,
peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
)
await peer.start()
yield peer
await peer.stop()
# ============================================================
# Temporary Plugin Fixtures
# ============================================================
import tempfile
import textwrap
@pytest.fixture
def temp_plugin_dir() -> Generator[Path, None, None]:
"""Create a temporary directory for plugin testing."""
with tempfile.TemporaryDirectory() as temp_dir:
yield Path(temp_dir)
def create_test_plugin(plugin_root: Path, name: str = "test_plugin") -> None:
"""Helper to create a minimal test plugin."""
(plugin_root / "commands").mkdir(parents=True, exist_ok=True)
(plugin_root / "commands" / "__init__.py").write_text("", encoding="utf-8")
(plugin_root / "requirements.txt").write_text("", encoding="utf-8")
(plugin_root / "plugin.yaml").write_text(
textwrap.dedent(
f"""\
_schema_version: 2
name: {name}
display_name: Test Plugin
desc: test
author: tester
version: 0.1.0
runtime:
python: "{sys.version_info.major}.{sys.version_info.minor}"
components:
- class: commands.sample:TestPlugin
type: command
name: test
description: test command
"""
),
encoding="utf-8",
)
(plugin_root / "commands" / "sample.py").write_text(
textwrap.dedent(
"""\
from astrbot_sdk import Context, MessageEvent, Star, on_command
class TestPlugin(Star):
@on_command("test")
async def test_cmd(self, event: MessageEvent, ctx: Context):
await event.reply("test ok")
"""
),
encoding="utf-8",
)
@pytest.fixture
def test_plugin(temp_plugin_dir: Path) -> Path:
"""Create a test plugin and return its root directory."""
plugin_root = temp_plugin_dir / "plugins" / "test_plugin"
create_test_plugin(plugin_root)
return temp_plugin_dir / "plugins"

14
tests_v4/pytest.ini Normal file
View File

@@ -0,0 +1,14 @@
[pytest]
testpaths = .
python_files = test_*.py
python_classes = Test*
python_functions = test_*
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
filterwarnings =
ignore::DeprecationWarning
ignore::PendingDeprecationWarning
markers =
slow: marks tests as slow (deselect with '-m "not slow"')
integration: marks tests as integration tests
unit: marks tests as unit tests

246
tests_v4/test_api_decorators.py Normal file
View File

@@ -0,0 +1,246 @@
"""
Unit tests for API decorators and Star class.
"""
from __future__ import annotations
import pytest
from astrbot_sdk import Context, MessageEvent, Star
from astrbot_sdk.decorators import (
get_handler_meta,
on_command,
on_event,
on_message,
on_schedule,
require_admin,
)
from astrbot_sdk.protocol.descriptors import (
CommandTrigger,
EventTrigger,
MessageTrigger,
ScheduleTrigger,
)
class TestOnCommandDecorator:
"""Tests for @on_command decorator."""
def test_decorator_sets_handler_meta(self):
"""@on_command should set __astrbot_handler_meta__."""
@on_command("hello")
async def hello_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(hello_handler)
assert meta is not None
assert isinstance(meta.trigger, CommandTrigger)
assert meta.trigger.command == "hello"
def test_decorator_supports_aliases(self):
"""@on_command should support command aliases."""
@on_command("hello", aliases=["hi", "hey"])
async def hello_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(hello_handler)
assert meta.trigger.aliases == ["hi", "hey"]
def test_decorator_supports_description(self):
"""@on_command should support description."""
@on_command("hello", description="Say hello")
async def hello_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(hello_handler)
assert meta.trigger.description == "Say hello"
class TestOnMessageDecorator:
"""Tests for @on_message decorator."""
def test_decorator_sets_handler_meta(self):
"""@on_message should set __astrbot_handler_meta__."""
@on_message()
async def message_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(message_handler)
assert meta is not None
assert isinstance(meta.trigger, MessageTrigger)
def test_decorator_supports_keywords(self):
"""@on_message should support keyword filtering."""
@on_message(keywords=["hello", "hi"])
async def keyword_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(keyword_handler)
assert meta.trigger.keywords == ["hello", "hi"]
def test_decorator_supports_regex(self):
"""@on_message should support regex filtering."""
@on_message(regex=r"\d+")
async def regex_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(regex_handler)
assert meta.trigger.regex == r"\d+"
class TestOnEventDecorator:
"""Tests for @on_event decorator."""
def test_decorator_sets_handler_meta(self):
"""@on_event should set __astrbot_handler_meta__."""
@on_event("message_received")
async def event_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(event_handler)
assert meta is not None
assert isinstance(meta.trigger, EventTrigger)
assert meta.trigger.event_type == "message_received"
class TestOnScheduleDecorator:
"""Tests for @on_schedule decorator."""
def test_decorator_sets_cron_trigger(self):
"""@on_schedule should create ScheduleTrigger with cron."""
@on_schedule(cron="* * * * *")
async def scheduled_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(scheduled_handler)
assert meta is not None
assert isinstance(meta.trigger, ScheduleTrigger)
assert meta.trigger.cron == "* * * * *"
def test_decorator_sets_interval_trigger(self):
"""@on_schedule should create ScheduleTrigger with interval."""
@on_schedule(interval_seconds=60)
async def interval_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(interval_handler)
assert meta.trigger.interval_seconds == 60
class TestRequireAdminDecorator:
"""Tests for @require_admin decorator."""
def test_decorator_sets_admin_permission(self):
"""@require_admin should set require_admin permission."""
@require_admin
async def admin_handler(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(admin_handler)
assert meta.permissions.require_admin is True
def test_can_combine_with_on_command(self):
"""@require_admin can be combined with @on_command."""
@on_command("admin")
@require_admin
async def admin_cmd(event: MessageEvent, ctx: Context):
pass
meta = get_handler_meta(admin_cmd)
assert isinstance(meta.trigger, CommandTrigger)
assert meta.trigger.command == "admin"
assert meta.permissions.require_admin is True
class TestStarClass:
"""Tests for Star base class."""
def test_star_is_new_star_by_default(self):
"""Star subclasses should be recognized as new-style."""
class MyPlugin(Star):
pass
assert MyPlugin.__astrbot_is_new_star__() is True
def test_star_collects_handler_names_from_decorators(self):
"""Star should collect decorated method names in __handlers__."""
class MyPlugin(Star):
@on_command("hello")
async def hello(self, event: MessageEvent, ctx: Context):
pass
@on_message()
async def on_msg(self, event: MessageEvent, ctx: Context):
pass
assert "hello" in MyPlugin.__handlers__
assert "on_msg" in MyPlugin.__handlers__
@pytest.mark.asyncio
async def test_star_on_error_calls_reply(self):
"""Star.on_error should call event.reply with error message."""
replies = []
class MyPlugin(Star):
pass
plugin = MyPlugin()
# Create event with mock reply handler
event = MessageEvent(text="test", session_id="s1")
event.bind_reply_handler(lambda text: replies.append(text) or asyncio.sleep(0))
# Create context (not used in default impl)
ctx = None
# on_error should call reply
await plugin.on_error(RuntimeError("test error"), event, ctx)
assert len(replies) == 1
assert "问题" in replies[0]
class TestTriggerModels:
"""Tests for trigger model validation."""
def test_command_trigger_validation(self):
"""CommandTrigger should validate command name."""
trigger = CommandTrigger(command="hello")
assert trigger.command == "hello"
def test_message_trigger_optional_keywords(self):
"""MessageTrigger should have optional keywords."""
trigger = MessageTrigger()
assert trigger.keywords == []
trigger_with_keywords = MessageTrigger(keywords=["a", "b"])
assert trigger_with_keywords.keywords == ["a", "b"]
def test_event_trigger_validation(self):
"""EventTrigger should store event type."""
trigger = EventTrigger(event_type="custom_event")
assert trigger.event_type == "custom_event"
def test_schedule_trigger_requires_one_strategy(self):
"""ScheduleTrigger should require exactly one strategy."""
with pytest.raises(ValueError):
ScheduleTrigger()
with pytest.raises(ValueError):
ScheduleTrigger(cron="* * * * *", interval_seconds=10)
trigger = ScheduleTrigger(interval_seconds=30)
assert trigger.interval_seconds == 30
import asyncio # For the async test

166
tests_v4/test_conftest_fixtures.py Normal file
View File

@@ -0,0 +1,166 @@
"""
Pytest-based tests for transport and peer communication.
These tests demonstrate the pytest fixtures defined in conftest.py.
"""
from __future__ import annotations
import pytest
from astrbot_sdk.context import CancelToken
from astrbot_sdk.errors import AstrBotError
from astrbot_sdk.protocol.descriptors import CapabilityDescriptor
from astrbot_sdk.protocol.messages import EventMessage, InitializeOutput, PeerInfo, ResultMessage
from astrbot_sdk.runtime.capability_router import CapabilityRouter, StreamExecution
from astrbot_sdk.runtime.peer import Peer
class TestTransportPair:
"""Tests for MemoryTransport fixture."""
async def test_transport_pair_is_connected(self, transport_pair):
"""Transport pair should be bidirectionally connected."""
left, right = transport_pair
assert left.partner is right
assert right.partner is left
async def test_transport_can_send_message(self, transport_pair):
"""Messages sent through transport should be received by partner."""
left, right = transport_pair
received = []
async def handler(payload):
received.append(payload)
right.set_message_handler(handler)
await left.start()
await right.start()
await left.send("test message")
assert len(received) == 1
assert received[0] == "test message"
class TestPeerConnection:
"""Tests for peer-to-peer communication using fixtures."""
async def test_plugin_can_initialize(self, core_peer, plugin_peer):
"""Plugin should be able to initialize with core."""
await plugin_peer.initialize([])
assert plugin_peer.remote_peer is not None
assert plugin_peer.remote_peer.name == "core"
async def test_plugin_can_invoke_capability(self, core_peer, plugin_peer):
"""Plugin should be able to invoke llm.chat capability."""
await plugin_peer.initialize([])
result = await plugin_peer.invoke("llm.chat", {"prompt": "hello"})
assert result["text"] == "Echo: hello"
async def test_plugin_can_stream_capability(self, core_peer, plugin_peer):
"""Plugin should be able to stream llm.stream_chat capability."""
await plugin_peer.initialize([])
stream = await plugin_peer.invoke_stream("llm.stream_chat", {"prompt": "hi"})
chunks = [event.data["text"] async for event in stream]
assert "".join(chunks) == "Echo: hi"
class TestProtocolErrors:
"""Tests for protocol error handling."""
async def test_stream_false_receiving_event_is_error(self, transport_pair):
"""stream=false receiving event should raise protocol error."""
left, right = transport_pair
plugin = Peer(
transport=right,
peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
)
await left.start()
await plugin.start()
task = asyncio.create_task(
plugin.invoke("llm.chat", {"prompt": "bad"}, request_id="req-1")
)
await asyncio.sleep(0)
await left.send(EventMessage(id="req-1", phase="started").model_dump_json())
with pytest.raises(AstrBotError) as exc_info:
await task
assert exc_info.value.code == "protocol_error"
await plugin.stop()
await left.stop()
async def test_stream_true_receiving_result_is_error(self, transport_pair):
"""stream=true receiving result should raise protocol error."""
import asyncio
left, right = transport_pair
plugin = Peer(
transport=right,
peer_info=PeerInfo(name="plugin", role="plugin", version="v4"),
)
await left.start()
await plugin.start()
stream = await plugin.invoke_stream(
"llm.stream_chat", {"prompt": "bad"}, request_id="stream-1"
)
await left.send(ResultMessage(id="stream-1", success=True, output={}).model_dump_json())
with pytest.raises(AstrBotError) as exc_info:
async for _ in stream:
pass
assert exc_info.value.code == "protocol_error"
await plugin.stop()
await left.stop()
class TestCapabilityRouter:
"""Tests for CapabilityRouter."""
def test_capability_name_validation(self):
"""Capability names must follow namespace.method format."""
router = CapabilityRouter()
invalid_names = ["llm", "llm.chat.extra", "LLM.chat", "llm.Chat"]
for name in invalid_names:
with pytest.raises(ValueError) as exc_info:
router.register(
CapabilityDescriptor(name=name, description="invalid")
)
assert name in str(exc_info.value)
def test_reserved_namespaces_rejected_for_exposed(self):
"""Reserved namespaces should be rejected for exposed registrations."""
router = CapabilityRouter()
reserved_names = ["handler.demo", "system.health", "internal.trace"]
for name in reserved_names:
with pytest.raises(ValueError) as exc_info:
router.register(
CapabilityDescriptor(name=name, description="reserved")
)
assert name in str(exc_info.value)
def test_reserved_namespaces_allowed_for_hidden(self):
"""Reserved namespaces should be allowed for hidden registrations."""
router = CapabilityRouter()
router.register(
CapabilityDescriptor(name="system.health", description="internal only"),
exposed=False,
)
descriptors = router.descriptors()
assert "system.health" not in [d.name for d in descriptors]
import asyncio # Import at end to avoid circular import issues in test file

124
tests_v4/test_context.py Normal file
View File

@@ -0,0 +1,124 @@
"""
Unit tests for Context module.
"""
from __future__ import annotations
import asyncio
import pytest
from astrbot_sdk.context import CancelToken, Context
from astrbot_sdk.protocol.messages import PeerInfo
from astrbot_sdk.runtime.peer import Peer
class TestCancelToken:
"""Tests for CancelToken."""
def test_initial_state(self):
"""CancelToken should start uncancelled."""
token = CancelToken()
assert not token.cancelled
def test_cancel_sets_state(self):
"""cancel() should set cancelled state."""
token = CancelToken()
token.cancel()
assert token.cancelled
def test_raise_if_cancelled_raises_when_cancelled(self):
"""raise_if_cancelled() should raise CancelledError when cancelled."""
token = CancelToken()
token.cancel()
with pytest.raises(asyncio.CancelledError):
token.raise_if_cancelled()
def test_raise_if_cancelled_no_raise_when_not_cancelled(self):
"""raise_if_cancelled() should not raise when not cancelled."""
token = CancelToken()
token.raise_if_cancelled() # Should not raise
@pytest.mark.asyncio
async def test_wait_blocks_until_cancelled(self):
"""wait() should block until cancel() is called."""
token = CancelToken()
async def cancel_after_delay():
await asyncio.sleep(0.01)
token.cancel()
task = asyncio.create_task(cancel_after_delay())
await token.wait()
assert token.cancelled
await task
class TestContext:
"""Tests for Context."""
def test_context_has_platform_facade(self, transport_pair):
"""Context should have platform facade."""
left, _ = transport_pair
peer = Peer(
transport=left,
peer_info=PeerInfo(name="test", role="plugin", version="v4"),
)
ctx = Context(peer=peer, plugin_id="test_plugin")
assert hasattr(ctx, "platform")
assert hasattr(ctx.platform, "send")
def test_context_has_llm_facade(self, transport_pair):
"""Context should have LLM facade."""
left, _ = transport_pair
peer = Peer(
transport=left,
peer_info=PeerInfo(name="test", role="plugin", version="v4"),
)
ctx = Context(peer=peer, plugin_id="test_plugin")
assert hasattr(ctx, "llm")
assert hasattr(ctx.llm, "chat")
assert hasattr(ctx.llm, "stream_chat")
def test_context_has_db_facade(self, transport_pair):
"""Context should have database facade."""
left, _ = transport_pair
peer = Peer(
transport=left,
peer_info=PeerInfo(name="test", role="plugin", version="v4"),
)
ctx = Context(peer=peer, plugin_id="test_plugin")
assert hasattr(ctx, "db")
assert hasattr(ctx.db, "get")
assert hasattr(ctx.db, "set")
assert hasattr(ctx.db, "delete")
def test_context_has_plugin_id(self, transport_pair):
"""Context should store plugin_id."""
left, _ = transport_pair
peer = Peer(
transport=left,
peer_info=PeerInfo(name="test", role="plugin", version="v4"),
)
ctx = Context(peer=peer, plugin_id="my_plugin")
assert ctx.plugin_id == "my_plugin"
def test_context_has_logger(self, transport_pair):
"""Context should have a logger bound with plugin_id."""
left, _ = transport_pair
peer = Peer(
transport=left,
peer_info=PeerInfo(name="test", role="plugin", version="v4"),
)
ctx = Context(peer=peer, plugin_id="test_plugin")
assert hasattr(ctx, "logger")

25
tests_v4/test_entrypoints.py
View File

@@ -4,7 +4,32 @@ import subprocess
import sys
import unittest
import pytest
def _is_astrbot_sdk_installed_in_site_packages() -> bool:
"""Check if astrbot_sdk is installed via pip (not just in PYTHONPATH)."""
try:
result = subprocess.run(
[sys.executable, "-c", "import astrbot_sdk; print(astrbot_sdk.__file__)"],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
return False
# Check if installed in site-packages, not just in local path
location = result.stdout.strip()
return "site-packages" in location or "dist-packages" in location
except Exception:
return False
@pytest.mark.integration
@pytest.mark.skipif(
not _is_astrbot_sdk_installed_in_site_packages(),
reason="astrbot_sdk not installed in site-packages (run: pip install -e .)"
)
class EntryPointTest(unittest.TestCase):
def test_import_package(self) -> None:
process = subprocess.run(

109
tests_v4/test_events.py Normal file
View File

@@ -0,0 +1,109 @@
"""
Unit tests for Events module.
"""
from __future__ import annotations
import asyncio
import pytest
from astrbot_sdk.events import MessageEvent, PlainTextResult
class TestMessageEvent:
"""Tests for MessageEvent."""
def test_from_payload_creates_event(self):
"""from_payload() should create a MessageEvent from dict."""
payload = {
"text": "hello world",
"session_id": "session-1",
"user_id": "user-1",
"platform": "test",
}
event = MessageEvent.from_payload(payload)
assert event.text == "hello world"
assert event.session_id == "session-1"
assert event.user_id == "user-1"
assert event.platform == "test"
def test_from_payload_handles_missing_fields(self):
"""from_payload() should handle missing optional fields."""
payload = {"text": "test"}
event = MessageEvent.from_payload(payload)
assert event.text == "test"
assert event.session_id == "" # Falls back to empty string
assert event.user_id is None # Optional field
def test_from_payload_preserves_raw_payload(self):
"""from_payload() should preserve raw payload."""
payload = {
"text": "test",
"extra_field": "extra_value",
}
event = MessageEvent.from_payload(payload)
assert event.raw == payload
assert event.raw["extra_field"] == "extra_value"
@pytest.mark.asyncio
async def test_bind_reply_handler(self):
"""bind_reply_handler() should enable reply functionality."""
event = MessageEvent(text="hello", session_id="s1")
replies = []
async def capture_reply(text: str) -> None:
replies.append(text)
event.bind_reply_handler(capture_reply)
await event.reply("response")
assert replies == ["response"]
@pytest.mark.asyncio
async def test_reply_without_handler_raises(self):
"""reply() without bound handler should raise."""
event = MessageEvent(text="hello", session_id="s1")
with pytest.raises(RuntimeError, match="未绑定 reply handler"):
await event.reply("response")
def test_to_payload(self):
"""to_payload() should serialize event to dict."""
event = MessageEvent(
text="hello",
session_id="session-1",
user_id="user-1",
platform="test",
)
payload = event.to_payload()
assert payload["text"] == "hello"
assert payload["session_id"] == "session-1"
assert payload["user_id"] == "user-1"
assert payload["platform"] == "test"
def test_plain_result_createsPlainTextResult(self):
"""plain_result() should create PlainTextResult."""
event = MessageEvent(text="hello", session_id="s1")
result = event.plain_result("test output")
assert isinstance(result, PlainTextResult)
assert result.text == "test output"
class TestPlainTextResult:
"""Tests for PlainTextResult."""
def test_create_plain_text_result(self):
"""Should create PlainTextResult."""
result = PlainTextResult(text="hello")
assert result.text == "hello"
def test_plain_text_result_is_dataclass(self):
"""PlainTextResult should be a dataclass with slots."""
result = PlainTextResult(text="test")
# It's a dataclass, check attribute access works
assert result.text == "test"