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

feat: 更新遗留 API，支持标量 JSON 值和兼容性文档，增强消息组件和事件过滤器功能

Browse Source
This commit is contained in:
whatevertogo
2026-03-13 05:44:01 +08:00
parent 96ea836edc
commit 72d600679f
14 changed files with 452 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
AGENTS.md
View File

@@ -9,6 +9,7 @@
- 2026-03-13: When checking whether a peer has finished remote initialization, avoid `getattr(mock, "remote_peer")` style probes in code that may receive `MagicMock` peers. `MagicMock` fabricates truthy child attributes, so `CapabilityProxy` should read explicit state from `peer.__dict__` or another concrete storage location instead of treating arbitrary attribute access as initialization.
- 2026-03-13: The repository has no legacy `src/astrbot_sdk/protocol` package to migrate file-for-file. `src-new/astrbot_sdk/protocol` is a v4-native protocol layer; compare it against legacy JSON-RPC behavior in `src/astrbot_sdk/runtime/*` and the maintained migration tests, not against a nonexistent old package tree.
- 2026-03-13: Old Star docs under `docs/zh/dev/star/` describe end-to-end legacy behavior, not just import surfaces. Current compat layer can import `AstrMessageEvent`, `MessageChain`, and some `filter.*` helpers, but handler result consumption is still effectively plain-text only and many documented legacy features remain absent or partial, including command groups, lifecycle/LLM hooks, session waiters, rich media helper constructors, config schema loading, and persona/provider management. Do not treat "type exists" as "old plugin behavior is compatible"; verify the runtime path end to end before declaring parity.
- 2026-03-13: Old Star docs and examples frequently import message components via `astrbot.api.message_components`, not only `astrbot.api.message`. When checking message compatibility, verify the dedicated `api.message_components` import path, legacy constructor aliases like `At(qq=...)` / `Node(uin=..., name=...)`, and helper factories such as `Image.fromURL()` before declaring the message compat surface complete.
# 开发命令

1
CLAUDE.md
View File

@@ -9,6 +9,7 @@
- 2026-03-13: When checking whether a peer has finished remote initialization, avoid `getattr(mock, "remote_peer")` style probes in code that may receive `MagicMock` peers. `MagicMock` fabricates truthy child attributes, so `CapabilityProxy` should read explicit state from `peer.__dict__` or another concrete storage location instead of treating arbitrary attribute access as initialization.
- 2026-03-13: The repository has no legacy `src/astrbot_sdk/protocol` package to migrate file-for-file. `src-new/astrbot_sdk/protocol` is a v4-native protocol layer; compare it against legacy JSON-RPC behavior in `src/astrbot_sdk/runtime/*` and the maintained migration tests, not against a nonexistent old package tree.
- 2026-03-13: Old Star docs under `docs/zh/dev/star/` describe end-to-end legacy behavior, not just import surfaces. Current compat layer can import `AstrMessageEvent`, `MessageChain`, and some `filter.*` helpers, but handler result consumption is still effectively plain-text only and many documented legacy features remain absent or partial, including command groups, lifecycle/LLM hooks, session waiters, rich media helper constructors, config schema loading, and persona/provider management. Do not treat "type exists" as "old plugin behavior is compatible"; verify the runtime path end to end before declaring parity.
- 2026-03-13: Old Star docs and examples frequently import message components via `astrbot.api.message_components`, not only `astrbot.api.message`. When checking message compatibility, verify the dedicated `api.message_components` import path, legacy constructor aliases like `At(qq=...)` / `Node(uin=..., name=...)`, and helper factories such as `Image.fromURL()` before declaring the message compat surface complete.
# 开发命令

7
src-new/astrbot_sdk/_legacy_api.py
View File

@@ -520,15 +520,16 @@ class LegacyContext:
f"迁移文档：{MIGRATION_DOC_URL}"
)
async def put_kv_data(self, key: str, value: dict[str, Any]) -> None:
async def put_kv_data(self, key: str, value: Any) -> None:
_warn_once("context.put_kv_data()", "ctx.db.set(key, value)")
ctx = self.require_runtime_context()
await ctx.db.set(key, value)
async def get_kv_data(self, key: str) -> dict[str, Any] | None:
async def get_kv_data(self, key: str, default: Any = None) -> Any:
_warn_once("context.get_kv_data()", "ctx.db.get(key)")
ctx = self.require_runtime_context()
return await ctx.db.get(key)
value = await ctx.db.get(key)
return default if value is None else value
async def delete_kv_data(self, key: str) -> None:
_warn_once("context.delete_kv_data()", "ctx.db.delete(key)")

179
src-new/astrbot_sdk/api/event/filter.py
View File

@@ -2,8 +2,10 @@
当前兼容层保证以下能力可运行：
- ``command(name)`` -> ``on_command(name)``
- ``regex(pattern)`` -> ``on_message(regex=pattern)``
- ``command(name, alias=..., priority=...)`` -> ``CommandTrigger``
- ``regex(pattern, priority=...)`` -> ``MessageTrigger``
- ``event_message_type(...)`` -> 记录消息类型约束
- ``platform_adapter_type(...)`` -> 记录平台约束
- ``permission(ADMIN)`` / ``permission_type(PermissionType.ADMIN)``
-> ``require_admin``
@@ -17,7 +19,8 @@ import enum
from abc import ABCMeta, abstractmethod
from typing import Any
from ...decorators import on_command, on_message, require_admin
from ...decorators import _get_or_create_meta, require_admin
from ...protocol.descriptors import CommandTrigger, MessageTrigger
from ..basic.astrbot_config import AstrBotConfig
from .astr_message_event import AstrMessageEvent
from .message_type import MessageType
@@ -71,6 +74,7 @@ class EventMessageTypeFilter:
class PlatformAdapterType(enum.Flag):
AIOCQHTTP = enum.auto()
QQOFFICIAL = enum.auto()
GEWECHAT = enum.auto()
TELEGRAM = enum.auto()
WECOM = enum.auto()
LARK = enum.auto()
@@ -86,6 +90,7 @@ class PlatformAdapterType(enum.Flag):
ALL = (
AIOCQHTTP
| QQOFFICIAL
| GEWECHAT
| TELEGRAM
| WECOM
| LARK
@@ -104,6 +109,7 @@ class PlatformAdapterType(enum.Flag):
ADAPTER_NAME_2_TYPE = {
"aiocqhttp": PlatformAdapterType.AIOCQHTTP,
"qq_official": PlatformAdapterType.QQOFFICIAL,
"gewechat": PlatformAdapterType.GEWECHAT,
"telegram": PlatformAdapterType.TELEGRAM,
"wecom": PlatformAdapterType.WECOM,
"lark": PlatformAdapterType.LARK,
@@ -180,12 +186,113 @@ class CustomFilterAnd(CustomFilter):
return self.filter1.filter(event, cfg) and self.filter2.filter(event, cfg)
def command(name: str):
return on_command(name)
EVENT_MESSAGE_TYPE_NAMES = {
EventMessageType.GROUP_MESSAGE: "group",
EventMessageType.PRIVATE_MESSAGE: "private",
EventMessageType.OTHER_MESSAGE: "other",
}
def regex(pattern: str):
return on_message(regex=pattern)
def _merge_unique(existing: list[str], additions: list[str]) -> list[str]:
merged: list[str] = []
for item in [*existing, *additions]:
if item not in merged:
merged.append(item)
return merged
def _normalize_aliases(*alias_groups: Any) -> list[str]:
aliases: list[str] = []
for alias_group in alias_groups:
if alias_group is None:
continue
if isinstance(alias_group, str):
values = [alias_group]
elif isinstance(alias_group, set):
values = sorted(str(item) for item in alias_group)
else:
values = [str(item) for item in alias_group]
aliases = _merge_unique(aliases, values)
return aliases
def _existing_trigger_constraints(
trigger: CommandTrigger | MessageTrigger | None,
) -> tuple[list[str], list[str], list[str]]:
if isinstance(trigger, CommandTrigger):
return list(trigger.platforms), list(trigger.message_types), []
if isinstance(trigger, MessageTrigger):
return (
list(trigger.platforms),
list(trigger.message_types),
list(trigger.keywords),
)
return [], [], []
def _apply_priority(meta, priority: int | None) -> None:
if priority is not None:
meta.priority = priority
def _selected_message_types(event_type: EventMessageType) -> list[str]:
selected: list[str] = []
for flag, name in EVENT_MESSAGE_TYPE_NAMES.items():
if event_type & flag:
selected.append(name)
return selected
def _selected_platforms(
platform_type: PlatformAdapterType | str,
) -> list[str]:
if isinstance(platform_type, str):
return [platform_type]
selected: list[str] = []
for name, flag in ADAPTER_NAME_2_TYPE.items():
if platform_type & flag:
selected.append(name)
return selected
def command(
name: str,
alias: set[str] | list[str] | tuple[str, ...] | str | None = None,
*,
aliases: set[str] | list[str] | tuple[str, ...] | str | None = None,
priority: int | None = None,
desc: str | None = None,
):
def decorator(func):
meta = _get_or_create_meta(func)
platforms, message_types, _ = _existing_trigger_constraints(meta.trigger)
meta.trigger = CommandTrigger(
command=name,
aliases=_normalize_aliases(alias, aliases),
description=desc,
platforms=platforms,
message_types=message_types,
)
_apply_priority(meta, priority)
return func
return decorator
def regex(pattern: str, *, priority: int | None = None):
def decorator(func):
meta = _get_or_create_meta(func)
platforms, message_types, keywords = _existing_trigger_constraints(meta.trigger)
meta.trigger = MessageTrigger(
regex=pattern,
keywords=keywords,
platforms=platforms,
message_types=message_types,
)
_apply_priority(meta, priority)
return func
return decorator
def permission(level: str | PermissionType):
@@ -216,8 +323,6 @@ def _unsupported_factory(name: str, replacement: str | None = None):
custom_filter = _unsupported_factory("custom_filter")
event_message_type = _unsupported_factory("event_message_type")
platform_adapter_type = _unsupported_factory("platform_adapter_type")
after_message_sent = _unsupported_factory("after_message_sent")
on_astrbot_loaded = _unsupported_factory("on_astrbot_loaded")
on_platform_loaded = _unsupported_factory("on_platform_loaded")
@@ -227,6 +332,62 @@ on_llm_response = _unsupported_factory("on_llm_response")
command_group = _unsupported_factory("command_group")
def event_message_type(
level: EventMessageType,
*,
priority: int | None = None,
):
message_types = _selected_message_types(level)
def decorator(func):
meta = _get_or_create_meta(func)
if meta.trigger is None:
meta.trigger = MessageTrigger(message_types=message_types)
elif isinstance(meta.trigger, MessageTrigger):
meta.trigger.message_types = _merge_unique(
meta.trigger.message_types,
message_types,
)
elif isinstance(meta.trigger, CommandTrigger):
meta.trigger.message_types = _merge_unique(
meta.trigger.message_types,
message_types,
)
else:
raise NotImplementedError(
"event_message_type() 目前只支持消息/命令处理器。"
)
_apply_priority(meta, priority)
return func
return decorator
def platform_adapter_type(
level: PlatformAdapterType | str,
*,
priority: int | None = None,
):
platforms = _selected_platforms(level)
def decorator(func):
meta = _get_or_create_meta(func)
if meta.trigger is None:
meta.trigger = MessageTrigger(platforms=platforms)
elif isinstance(meta.trigger, MessageTrigger):
meta.trigger.platforms = _merge_unique(meta.trigger.platforms, platforms)
elif isinstance(meta.trigger, CommandTrigger):
meta.trigger.platforms = _merge_unique(meta.trigger.platforms, platforms)
else:
raise NotImplementedError(
"platform_adapter_type() 目前只支持消息/命令处理器。"
)
_apply_priority(meta, priority)
return func
return decorator
class _FilterNamespace:
command = staticmethod(command)
regex = staticmethod(regex)

53
src-new/astrbot_sdk/api/message/components.py
View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from enum import Enum
from typing import Literal
from pydantic import BaseModel, Field
from pydantic import AliasChoices, BaseModel, Field
class ComponentType(str, Enum):
@@ -50,30 +50,60 @@ class Plain(BaseMessageComponent):
class Image(BaseMessageComponent):
type: Literal[CompT.Image] = CompT.Image
file: str
file: str = Field(validation_alias=AliasChoices("file", "url", "path"))
@classmethod
def fromURL(cls, url: str) -> "Image":
return cls(file=url)
@classmethod
def fromFileSystem(cls, path: str) -> "Image":
return cls(file=path)
class Record(BaseMessageComponent):
type: Literal[CompT.Record] = CompT.Record
file: str
file: str = Field(validation_alias=AliasChoices("file", "url", "path"))
@classmethod
def fromURL(cls, url: str) -> "Record":
return cls(file=url)
@classmethod
def fromFileSystem(cls, path: str) -> "Record":
return cls(file=path)
class Video(BaseMessageComponent):
type: Literal[CompT.Video] = CompT.Video
file: str
file: str = Field(validation_alias=AliasChoices("file", "url", "path"))
@classmethod
def fromURL(cls, url: str) -> "Video":
return cls(file=url)
@classmethod
def fromFileSystem(cls, path: str) -> "Video":
return cls(file=path)
class File(BaseMessageComponent):
type: Literal[CompT.File] = CompT.File
file_name: str
file_name: str = Field(validation_alias=AliasChoices("file_name", "name"))
mime_type: str | None = None
file: str
file: str = Field(validation_alias=AliasChoices("file", "url", "path"))
class At(BaseMessageComponent):
type: Literal[CompT.At] = CompT.At
user_id: str | None = None
user_name: str | None = None
user_id: str | None = Field(
default=None,
validation_alias=AliasChoices("user_id", "qq"),
)
user_name: str | None = Field(
default=None,
validation_alias=AliasChoices("user_name", "name"),
)
class AtAll(At):
@@ -92,8 +122,11 @@ class Reply(BaseMessageComponent):
class Node(BaseMessageComponent):
type: Literal[CompT.Node] = CompT.Node
sender_id: str
nickname: str | None = None
sender_id: str = Field(validation_alias=AliasChoices("sender_id", "uin"))
nickname: str | None = Field(
default=None,
validation_alias=AliasChoices("nickname", "name"),
)
content: list[BaseMessageComponent] = Field(default_factory=list)

19
src-new/astrbot_sdk/clients/db.py
View File

@@ -16,7 +16,7 @@
功能说明：
- 数据永久存储，除非用户显式删除
- 值类型为 dict，支持结构化数据
- 值类型支持任意 JSON 数据
- 支持前缀查询键列表
TODO:
@@ -48,14 +48,14 @@ class DBClient:
"""
self._proxy = proxy
async def get(self, key: str) -> dict[str, Any] | None:
async def get(self, key: str) -> Any | None:
"""获取指定键的值。
Args:
key: 数据键名
Returns:
存储的字典值，若键不存在或值非 dict 则返回 None
存储的值，若键不存在则返回 None
示例:
data = await ctx.db.get("user_settings")
@@ -63,24 +63,19 @@ class DBClient:
print(data["theme"])
"""
output = await self._proxy.call("db.get", {"key": key})
value = output.get("value")
return value if isinstance(value, dict) else None
return output.get("value")
async def set(self, key: str, value: dict[str, Any]) -> None:
async def set(self, key: str, value: Any) -> None:
"""设置键值对。
Args:
key: 数据键名
value: 要存储的字典值
Raises:
TypeError: 如果 value 不是 dict
value: 要存储的 JSON 值
示例:
await ctx.db.set("user_settings", {"theme": "dark", "lang": "zh"})
await ctx.db.set("greeted", True)
"""
if not isinstance(value, dict):
raise TypeError("db.set 的 value 必须是 dict")
await self._proxy.call("db.set", {"key": key, "value": value})
async def delete(self, key: str) -> None:

12
src-new/astrbot_sdk/protocol/descriptors.py
View File

@@ -1,6 +1,6 @@
"""v4 协议描述符模型。
`protocol` 是 v4 新引入的协议层抽象，不对应旧树中的一个同名目录。这里
`protocol` 是 v4 新引入的协议层抽象，不对应旧树(圣诞树)中的一个同名目录。这里
定义的是跨进程握手和调度时使用的声明式元数据，而不是运行时的具体处理器/
能力实现。
"""
@@ -107,12 +107,12 @@ DB_GET_INPUT_SCHEMA = _object_schema(
)
DB_GET_OUTPUT_SCHEMA = _object_schema(
required=("value",),
value=_nullable({"type": "object"}),
value=_nullable({}),
)
DB_SET_INPUT_SCHEMA = _object_schema(
required=("key", "value"),
key={"type": "string"},
value={"type": "object"},
value={},
)
DB_SET_OUTPUT_SCHEMA = _object_schema()
DB_DELETE_INPUT_SCHEMA = _object_schema(
@@ -260,12 +260,16 @@ class CommandTrigger(_DescriptorBase):
command: 命令名称（不含前缀，如 "help"）
aliases: 命令别名列表
description: 命令描述，用于帮助文档
platforms: 允许的平台列表，为空表示所有平台
message_types: 限定的消息类型列表，为空表示不限
"""
type: Literal["command"] = "command"
command: str
aliases: list[str] = Field(default_factory=list)
description: str | None = None
platforms: list[str] = Field(default_factory=list)
message_types: list[str] = Field(default_factory=list)
class MessageTrigger(_DescriptorBase):
@@ -280,6 +284,7 @@ class MessageTrigger(_DescriptorBase):
regex: 正则表达式模式，匹配消息文本
keywords: 关键词列表，消息包含任一关键词即触发
platforms: 目标平台列表，为空表示所有平台
message_types: 限定的消息类型列表，为空表示不限
Note:
`regex` 和 `keywords` 可以同时为空，此时表示“任意消息均可触发”，
@@ -290,6 +295,7 @@ class MessageTrigger(_DescriptorBase):
regex: str | None = None
keywords: list[str] = Field(default_factory=list)
platforms: list[str] = Field(default_factory=list)
message_types: list[str] = Field(default_factory=list)
class EventTrigger(_DescriptorBase):

7
src-new/astrbot_sdk/runtime/capability_router.py
View File

@@ -122,7 +122,7 @@ class _CapabilityRegistration:
class CapabilityRouter:
def __init__(self) -> None:
self._registrations: dict[str, _CapabilityRegistration] = {}
self.db_store: dict[str, dict[str, Any]] = {}
self.db_store: dict[str, Any] = {}
self.memory_store: dict[str, dict[str, Any]] = {}
self.sent_messages: list[dict[str, Any]] = []
self._register_builtin_capabilities()
@@ -277,10 +277,7 @@ class CapabilityRouter:
_request_id: str, payload: dict[str, Any], _token
) -> dict[str, Any]:
key = str(payload.get("key", ""))
value = payload.get("value")
if not isinstance(value, dict):
raise AstrBotError.invalid_input("db.set 的 value 必须是 object")
self.db_store[key] = value
self.db_store[key] = payload.get("value")
return {}
async def db_delete(

86
tests_v4/test_api_event_filter.py
View File

@@ -9,12 +9,16 @@ import pytest
from astrbot_sdk.api.event.filter import (
ADMIN,
EventMessageType,
PermissionType,
PlatformAdapterType,
command,
command_group,
event_message_type,
filter,
permission,
permission_type,
platform_adapter_type,
regex,
)
from astrbot_sdk.decorators import get_handler_meta
@@ -45,6 +49,20 @@ class TestCommandFilter:
assert hasattr(test_handler, "__astrbot_handler_meta__")
def test_command_supports_alias_and_priority(self):
"""command() should map legacy alias/priority arguments."""
@command("hello", alias={"hi", "hey"}, priority=3, desc="greeting")
async def hello_handler():
pass
meta = get_handler_meta(hello_handler)
assert meta is not None
assert meta.priority == 3
assert isinstance(meta.trigger, CommandTrigger)
assert sorted(meta.trigger.aliases) == ["hey", "hi"]
assert meta.trigger.description == "greeting"
class TestRegexFilter:
"""Tests for regex() filter function."""
@@ -181,6 +199,74 @@ class TestFilterNamespace:
assert meta.permissions.require_admin is True
class TestCompatFilterComposition:
"""Tests for legacy filter composition helpers."""
def test_event_message_type_creates_message_trigger(self):
"""event_message_type() should create a message trigger when missing."""
@event_message_type(EventMessageType.PRIVATE_MESSAGE)
async def private_handler():
pass
meta = get_handler_meta(private_handler)
assert isinstance(meta.trigger, MessageTrigger)
assert meta.trigger.message_types == ["private"]
def test_event_message_type_merges_into_command_trigger(self):
"""event_message_type() should preserve command triggers."""
@command("hello")
@event_message_type(EventMessageType.GROUP_MESSAGE)
async def group_command():
pass
meta = get_handler_meta(group_command)
assert isinstance(meta.trigger, CommandTrigger)
assert meta.trigger.command == "hello"
assert meta.trigger.message_types == ["group"]
def test_platform_adapter_type_creates_message_trigger(self):
"""platform_adapter_type() should create a message trigger when missing."""
@platform_adapter_type(PlatformAdapterType.AIOCQHTTP | PlatformAdapterType.KOOK)
async def platform_handler():
pass
meta = get_handler_meta(platform_handler)
assert isinstance(meta.trigger, MessageTrigger)
assert meta.trigger.platforms == ["aiocqhttp", "kook"]
def test_platform_adapter_type_merges_into_command_trigger(self):
"""platform_adapter_type() should preserve command triggers."""
@command("hello")
@platform_adapter_type(PlatformAdapterType.AIOCQHTTP)
async def platform_command():
pass
meta = get_handler_meta(platform_command)
assert isinstance(meta.trigger, CommandTrigger)
assert meta.trigger.command == "hello"
assert meta.trigger.platforms == ["aiocqhttp"]
def test_command_preserves_existing_platform_and_message_constraints(self):
"""command() should not discard previously-registered compat filters."""
@command("hello", alias={"hi"})
@platform_adapter_type(PlatformAdapterType.QQOFFICIAL)
@event_message_type(EventMessageType.PRIVATE_MESSAGE)
async def compat_command():
pass
meta = get_handler_meta(compat_command)
assert isinstance(meta.trigger, CommandTrigger)
assert meta.trigger.command == "hello"
assert meta.trigger.aliases == ["hi"]
assert meta.trigger.platforms == ["qq_official"]
assert meta.trigger.message_types == ["private"]
class TestAdminConstant:
"""Tests for ADMIN constant."""

32
tests_v4/test_api_legacy_context.py
View File

@@ -232,6 +232,21 @@ class TestLegacyContextMethods:
mock_db.set.assert_called_once_with("test_key", {"value": 123})
@pytest.mark.asyncio
async def test_put_kv_data_accepts_scalar_value(self):
"""put_kv_data() should support scalar JSON values."""
mock_db = AsyncMock()
mock_ctx = MagicMock()
mock_ctx.db = mock_db
legacy_ctx = LegacyContext("test_plugin")
legacy_ctx._runtime_context = mock_ctx
await legacy_ctx.put_kv_data("greeted", True)
mock_db.set.assert_called_once_with("greeted", True)
@pytest.mark.asyncio
async def test_get_kv_data(self):
"""get_kv_data() should delegate to db.get()."""
@@ -249,6 +264,23 @@ class TestLegacyContextMethods:
mock_db.get.assert_called_once_with("my_key")
assert result == {"data": "hello"}
@pytest.mark.asyncio
async def test_get_kv_data_returns_default_when_missing(self):
"""get_kv_data() should honor the legacy default parameter."""
mock_db = AsyncMock()
mock_db.get = AsyncMock(return_value=None)
mock_ctx = MagicMock()
mock_ctx.db = mock_db
legacy_ctx = LegacyContext("test_plugin")
legacy_ctx._runtime_context = mock_ctx
result = await legacy_ctx.get_kv_data("missing", False)
mock_db.get.assert_called_once_with("missing")
assert result is False
@pytest.mark.asyncio
async def test_delete_kv_data(self):
"""delete_kv_data() should delegate to db.delete()."""

60
tests_v4/test_api_message_components.py Normal file
View File

@@ -0,0 +1,60 @@
"""
Tests for legacy message component compatibility helpers.
"""
from __future__ import annotations
from astrbot_sdk.api import message_components as Comp
class TestLegacyMessageComponentAliases:
"""Tests for legacy constructor aliases."""
def test_at_accepts_qq_alias(self):
"""At should accept the legacy qq field name."""
component = Comp.At(qq="123456", name="Tester")
assert component.user_id == "123456"
assert component.user_name == "Tester"
def test_file_accepts_name_alias(self):
"""File should accept the legacy name field name."""
component = Comp.File(file="/tmp/demo.txt", name="demo.txt")
assert component.file == "/tmp/demo.txt"
assert component.file_name == "demo.txt"
def test_node_accepts_uin_and_name_aliases(self):
"""Node should accept the legacy uin/name constructor fields."""
component = Comp.Node(uin="10001", name="AstrBot")
assert component.sender_id == "10001"
assert component.nickname == "AstrBot"
class TestLegacyMessageComponentFactories:
"""Tests for legacy media helper factories."""
def test_image_from_url(self):
"""Image.fromURL() should create a component with file payload."""
component = Comp.Image.fromURL("https://example.com/image.png")
assert component.file == "https://example.com/image.png"
def test_image_from_file_system(self):
"""Image.fromFileSystem() should create a component with file payload."""
component = Comp.Image.fromFileSystem("C:/tmp/image.png")
assert component.file == "C:/tmp/image.png"
def test_video_from_url(self):
"""Video.fromURL() should create a component with file payload."""
component = Comp.Video.fromURL("https://example.com/video.mp4")
assert component.file == "https://example.com/video.mp4"
def test_record_from_file_system(self):
"""Record.fromFileSystem() should create a component with file payload."""
component = Comp.Record.fromFileSystem("C:/tmp/audio.wav")
assert component.file == "C:/tmp/audio.wav"

29
tests_v4/test_capability_router.py
View File

@@ -759,19 +759,28 @@ class TestBuiltinDbCapabilities:
assert "other" not in result["keys"]
@pytest.mark.asyncio
async def test_db_set_invalid_value(self):
"""db.set should reject non-object value."""
async def test_db_set_scalar_value(self):
"""db.set should accept scalar JSON values."""
router = CapabilityRouter()
token = CancelToken()
with pytest.raises(AstrBotError, match="value 必须是 object"):
await router.execute(
"db.set",
{"key": "test", "value": "not_an_object"},
stream=False,
cancel_token=token,
request_id="req-1",
)
await router.execute(
"db.set",
{"key": "test", "value": True},
stream=False,
cancel_token=token,
request_id="req-1",
)
result = await router.execute(
"db.get",
{"key": "test"},
stream=False,
cancel_token=token,
request_id="req-2",
)
assert result["value"] is True
class TestBuiltinPlatformCapabilities:

21
tests_v4/test_db_client.py
View File

@@ -49,15 +49,15 @@ class TestDBClientGet:
assert result is None
@pytest.mark.asyncio
async def test_get_returns_none_for_non_dict_value(self):
"""get() should return None when value is not a dict."""
async def test_get_returns_scalar_value(self):
"""get() should preserve non-dict scalar values."""
proxy = AsyncMock(spec=CapabilityProxy)
proxy.call = AsyncMock(return_value={"value": "not a dict"})
proxy.call = AsyncMock(return_value={"value": True})
client = DBClient(proxy)
result = await client.get("my_key")
assert result is None
assert result is True
@pytest.mark.asyncio
async def test_get_returns_none_when_value_key_missing(self):
@@ -126,13 +126,18 @@ class TestDBClientSet:
)
@pytest.mark.asyncio
async def test_set_raises_type_error_for_non_dict_value(self):
"""set() should reject non-dict values before calling proxy."""
async def test_set_accepts_scalar_value(self):
"""set() should accept scalar JSON values."""
proxy = AsyncMock(spec=CapabilityProxy)
proxy.call = AsyncMock(return_value={})
client = DBClient(proxy)
with pytest.raises(TypeError, match="db.set 的 value 必须是 dict"):
await client.set("bad", "not a dict")
await client.set("flag", True)
proxy.call.assert_called_once_with(
"db.set",
{"key": "flag", "value": True},
)
class TestDBClientDelete:

5
tests_v4/test_protocol_descriptors.py
View File

@@ -59,6 +59,8 @@ class TestCommandTrigger:
assert trigger.command == "hello"
assert trigger.aliases == []
assert trigger.description is None
assert trigger.platforms == []
assert trigger.message_types == []
def test_with_aliases_and_description(self):
"""CommandTrigger should accept aliases and description."""
@@ -92,6 +94,7 @@ class TestMessageTrigger:
assert trigger.regex is None
assert trigger.keywords == []
assert trigger.platforms == []
assert trigger.message_types == []
def test_with_regex(self):
"""MessageTrigger should accept regex pattern."""
@@ -114,10 +117,12 @@ class TestMessageTrigger:
regex=r"test",
keywords=["keyword"],
platforms=["platform"],
message_types=["private"],
)
assert trigger.regex == "test"
assert trigger.keywords == ["keyword"]
assert trigger.platforms == ["platform"]
assert trigger.message_types == ["private"]
class TestEventTrigger: