mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
feat: 添加测试用例，覆盖 API 装饰器、事件过滤器和遗留上下文的功能
This commit is contained in:
whatevertogo
171
tests_v4/test_api_event_filter.py
Normal file
171
tests_v4/test_api_event_filter.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
Tests for api/event/filter.py - Event filter decorators and utilities.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from astrbot_sdk.api.event.filter import ADMIN, command, filter, permission, regex
|
||||
from astrbot_sdk.decorators import get_handler_meta
|
||||
from astrbot_sdk.protocol.descriptors import CommandTrigger, MessageTrigger
|
||||
|
||||
|
||||
class TestCommandFilter:
|
||||
"""Tests for command() filter function."""
|
||||
|
||||
def test_command_creates_command_trigger(self):
|
||||
"""command() should create a CommandTrigger."""
|
||||
|
||||
@command("hello")
|
||||
async def hello_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(hello_handler)
|
||||
assert meta is not None
|
||||
assert isinstance(meta.trigger, CommandTrigger)
|
||||
assert meta.trigger.command == "hello"
|
||||
|
||||
def test_command_is_decorator(self):
|
||||
"""command() should be usable as a decorator."""
|
||||
|
||||
@command("test")
|
||||
async def test_handler():
|
||||
pass
|
||||
|
||||
assert hasattr(test_handler, "__astrbot_handler_meta__")
|
||||
|
||||
|
||||
class TestRegexFilter:
|
||||
"""Tests for regex() filter function."""
|
||||
|
||||
def test_regex_creates_message_trigger_with_regex(self):
|
||||
"""regex() should create a MessageTrigger with regex pattern."""
|
||||
|
||||
@regex(r"\d+")
|
||||
async def number_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(number_handler)
|
||||
assert meta is not None
|
||||
assert isinstance(meta.trigger, MessageTrigger)
|
||||
assert meta.trigger.regex == r"\d+"
|
||||
|
||||
def test_regex_pattern_is_stored(self):
|
||||
"""regex() should store the pattern correctly."""
|
||||
|
||||
@regex(r"hello\s+world")
|
||||
async def greeting_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(greeting_handler)
|
||||
assert meta.trigger.regex == r"hello\s+world"
|
||||
|
||||
|
||||
class TestPermissionFilter:
|
||||
"""Tests for permission() filter function."""
|
||||
|
||||
def test_permission_admin_sets_require_admin(self):
|
||||
"""permission(ADMIN) should set require_admin permission."""
|
||||
|
||||
@permission(ADMIN)
|
||||
async def admin_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(admin_handler)
|
||||
assert meta is not None
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
def test_permission_non_admin_passes_through(self):
|
||||
"""permission() with non-ADMIN level should pass through."""
|
||||
|
||||
@permission("user")
|
||||
async def user_handler():
|
||||
pass
|
||||
|
||||
# Should not set admin permission
|
||||
meta = get_handler_meta(user_handler)
|
||||
assert meta is None or not meta.permissions.require_admin
|
||||
|
||||
def test_permission_can_be_combined_with_command(self):
|
||||
"""permission() can be combined with other decorators."""
|
||||
|
||||
@command("admin")
|
||||
@permission(ADMIN)
|
||||
async def admin_command():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(admin_command)
|
||||
assert isinstance(meta.trigger, CommandTrigger)
|
||||
assert meta.trigger.command == "admin"
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
|
||||
class TestFilterNamespace:
|
||||
"""Tests for filter namespace object."""
|
||||
|
||||
def test_filter_namespace_has_command(self):
|
||||
"""filter namespace should have command method."""
|
||||
assert hasattr(filter, "command")
|
||||
assert callable(filter.command)
|
||||
|
||||
def test_filter_namespace_has_regex(self):
|
||||
"""filter namespace should have regex method."""
|
||||
assert hasattr(filter, "regex")
|
||||
assert callable(filter.regex)
|
||||
|
||||
def test_filter_namespace_has_permission(self):
|
||||
"""filter namespace should have permission method."""
|
||||
assert hasattr(filter, "permission")
|
||||
assert callable(filter.permission)
|
||||
|
||||
def test_filter_command_works_as_decorator(self):
|
||||
"""filter.command() should work as a decorator."""
|
||||
|
||||
@filter.command("ping")
|
||||
async def ping_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(ping_handler)
|
||||
assert meta.trigger.command == "ping"
|
||||
|
||||
def test_filter_regex_works_as_decorator(self):
|
||||
"""filter.regex() should work as a decorator."""
|
||||
|
||||
@filter.regex(r"test")
|
||||
async def test_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(test_handler)
|
||||
assert meta.trigger.regex == r"test"
|
||||
|
||||
def test_filter_permission_admin_works(self):
|
||||
"""filter.permission(ADMIN) should set admin permission."""
|
||||
|
||||
@filter.permission(ADMIN)
|
||||
async def admin_handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(admin_handler)
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
|
||||
class TestAdminConstant:
|
||||
"""Tests for ADMIN constant."""
|
||||
|
||||
def test_admin_constant_value(self):
|
||||
"""ADMIN constant should be 'admin'."""
|
||||
assert ADMIN == "admin"
|
||||
|
||||
|
||||
class TestModuleExports:
|
||||
"""Tests for module exports."""
|
||||
|
||||
def test_all_exports(self):
|
||||
"""Module should export expected names."""
|
||||
from astrbot_sdk.api.event.filter import __all__
|
||||
|
||||
assert "ADMIN" in __all__
|
||||
assert "command" in __all__
|
||||
assert "regex" in __all__
|
||||
assert "permission" in __all__
|
||||
assert "filter" in __all__
|
||||
364
tests_v4/test_api_legacy_context.py
Normal file
364
tests_v4/test_api_legacy_context.py
Normal file
@@ -0,0 +1,364 @@
|
||||
"""
|
||||
Tests for _legacy_api.py - Legacy compatibility layer.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from astrbot_sdk._legacy_api import (
|
||||
MIGRATION_DOC_URL,
|
||||
CommandComponent,
|
||||
Context,
|
||||
LegacyContext,
|
||||
LegacyConversationManager,
|
||||
)
|
||||
from astrbot_sdk.star import Star
|
||||
|
||||
|
||||
class TestLegacyContext:
|
||||
"""Tests for LegacyContext."""
|
||||
|
||||
def test_init_with_plugin_id(self):
|
||||
"""LegacyContext should store plugin_id."""
|
||||
ctx = LegacyContext("test_plugin")
|
||||
assert ctx.plugin_id == "test_plugin"
|
||||
|
||||
def test_has_conversation_manager(self):
|
||||
"""LegacyContext should have conversation_manager."""
|
||||
ctx = LegacyContext("test_plugin")
|
||||
assert hasattr(ctx, "conversation_manager")
|
||||
assert isinstance(ctx.conversation_manager, LegacyConversationManager)
|
||||
|
||||
def test_require_runtime_context_raises_when_not_bound(self):
|
||||
"""require_runtime_context() should raise when not bound."""
|
||||
ctx = LegacyContext("test_plugin")
|
||||
with pytest.raises(RuntimeError, match="尚未绑定运行时 Context"):
|
||||
ctx.require_runtime_context()
|
||||
|
||||
def test_bind_runtime_context(self):
|
||||
"""bind_runtime_context() should bind a NewContext."""
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.plugin_id = "test"
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx.bind_runtime_context(mock_ctx)
|
||||
|
||||
assert legacy_ctx.require_runtime_context() is mock_ctx
|
||||
|
||||
def test_context_alias_is_legacy_context(self):
|
||||
"""Context should be an alias for LegacyContext."""
|
||||
assert Context is LegacyContext
|
||||
|
||||
|
||||
class TestLegacyConversationManager:
|
||||
"""Tests for LegacyConversationManager."""
|
||||
|
||||
def test_init_with_parent(self):
|
||||
"""LegacyConversationManager should store parent reference."""
|
||||
parent = LegacyContext("test_plugin")
|
||||
manager = LegacyConversationManager(parent)
|
||||
assert manager._parent is parent
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_new_conversation_creates_id(self):
|
||||
"""new_conversation() should create a conversation ID."""
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.plugin_id = "my_plugin"
|
||||
mock_ctx.db = AsyncMock()
|
||||
mock_ctx.db.get = AsyncMock(return_value=None)
|
||||
mock_ctx.db.set = AsyncMock()
|
||||
|
||||
legacy_ctx = LegacyContext("my_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
conv_id = await legacy_ctx.conversation_manager.new_conversation(
|
||||
unified_msg_origin="session-1"
|
||||
)
|
||||
|
||||
assert conv_id.startswith("my_plugin-conv-")
|
||||
assert conv_id.endswith("-1")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_new_conversation_stores_metadata(self):
|
||||
"""new_conversation() should store conversation metadata."""
|
||||
stored_data = {}
|
||||
|
||||
async def mock_get(key):
|
||||
return stored_data.get(key)
|
||||
|
||||
async def mock_set(key, value):
|
||||
stored_data[key] = value
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.plugin_id = "my_plugin"
|
||||
mock_ctx.db = MagicMock()
|
||||
mock_ctx.db.get = mock_get
|
||||
mock_ctx.db.set = mock_set
|
||||
|
||||
legacy_ctx = LegacyContext("my_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
conv_id = await legacy_ctx.conversation_manager.new_conversation(
|
||||
unified_msg_origin="session-1",
|
||||
platform_id="telegram",
|
||||
title="Test Chat",
|
||||
persona_id="assistant",
|
||||
)
|
||||
|
||||
# Verify stored data
|
||||
assert "__compat_conversations__" in stored_data
|
||||
assert conv_id in stored_data["__compat_conversations__"]
|
||||
data = stored_data["__compat_conversations__"][conv_id]
|
||||
assert data["platform_id"] == "telegram"
|
||||
assert data["title"] == "Test Chat"
|
||||
assert data["persona_id"] == "assistant"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_new_conversation_increments_counter(self):
|
||||
"""new_conversation() should increment counter per origin."""
|
||||
stored_data = {}
|
||||
|
||||
async def mock_get(key):
|
||||
return stored_data.get(key)
|
||||
|
||||
async def mock_set(key, value):
|
||||
stored_data[key] = value
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.plugin_id = "my_plugin"
|
||||
mock_ctx.db = MagicMock()
|
||||
mock_ctx.db.get = mock_get
|
||||
mock_ctx.db.set = mock_set
|
||||
|
||||
legacy_ctx = LegacyContext("my_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
id1 = await legacy_ctx.conversation_manager.new_conversation(
|
||||
unified_msg_origin="session-1"
|
||||
)
|
||||
id2 = await legacy_ctx.conversation_manager.new_conversation(
|
||||
unified_msg_origin="session-1"
|
||||
)
|
||||
id3 = await legacy_ctx.conversation_manager.new_conversation(
|
||||
unified_msg_origin="session-2"
|
||||
)
|
||||
|
||||
assert id1.endswith("-1")
|
||||
assert id2.endswith("-2")
|
||||
assert id3.endswith("-1") # Different session, starts at 1
|
||||
|
||||
|
||||
class TestLegacyContextMethods:
|
||||
"""Tests for LegacyContext methods that delegate to NewContext."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_put_kv_data(self):
|
||||
"""put_kv_data() should delegate to db.set()."""
|
||||
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("test_key", {"value": 123})
|
||||
|
||||
mock_db.set.assert_called_once_with("test_key", {"value": 123})
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_kv_data(self):
|
||||
"""get_kv_data() should delegate to db.get()."""
|
||||
mock_db = AsyncMock()
|
||||
mock_db.get = AsyncMock(return_value={"data": "hello"})
|
||||
|
||||
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("my_key")
|
||||
|
||||
mock_db.get.assert_called_once_with("my_key")
|
||||
assert result == {"data": "hello"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_kv_data(self):
|
||||
"""delete_kv_data() should delegate to db.delete()."""
|
||||
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.delete_kv_data("to_delete")
|
||||
|
||||
mock_db.delete.assert_called_once_with("to_delete")
|
||||
|
||||
|
||||
class TestLegacyContextSendMessage:
|
||||
"""Tests for LegacyContext.send_message()."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_message_with_plain_string(self):
|
||||
"""send_message() should handle plain string."""
|
||||
mock_platform = AsyncMock()
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.platform = mock_platform
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
await legacy_ctx.send_message("session-1", "hello world")
|
||||
|
||||
mock_platform.send.assert_called_once_with("session-1", "hello world")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_message_with_get_plain_text_method(self):
|
||||
"""send_message() should use get_plain_text() if available."""
|
||||
|
||||
class MockMessageChain:
|
||||
def get_plain_text(self):
|
||||
return "extracted text"
|
||||
|
||||
mock_platform = AsyncMock()
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.platform = mock_platform
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
await legacy_ctx.send_message("session-1", MockMessageChain())
|
||||
|
||||
mock_platform.send.assert_called_once_with("session-1", "extracted text")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_message_with_to_text_method(self):
|
||||
"""send_message() should use to_text() if get_plain_text() not available."""
|
||||
|
||||
class MockMessageChain:
|
||||
def to_text(self):
|
||||
return "to_text result"
|
||||
|
||||
mock_platform = AsyncMock()
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.platform = mock_platform
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
await legacy_ctx.send_message("session-1", MockMessageChain())
|
||||
|
||||
mock_platform.send.assert_called_once_with("session-1", "to_text result")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_message_falls_back_to_str(self):
|
||||
"""send_message() should fall back to str() if no text method."""
|
||||
|
||||
class MockObject:
|
||||
def __str__(self):
|
||||
return "stringified"
|
||||
|
||||
mock_platform = AsyncMock()
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.platform = mock_platform
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
await legacy_ctx.send_message("session-1", MockObject())
|
||||
|
||||
mock_platform.send.assert_called_once_with("session-1", "stringified")
|
||||
|
||||
|
||||
class TestLegacyContextLLMMethods:
|
||||
"""Tests for LegacyContext LLM methods."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_llm_generate_delegates_to_chat_raw(self):
|
||||
"""llm_generate() should delegate to ctx.llm.chat_raw()."""
|
||||
mock_llm = AsyncMock()
|
||||
mock_llm.chat_raw = AsyncMock(return_value=MagicMock(text="response"))
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.llm = mock_llm
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
result = await legacy_ctx.llm_generate(
|
||||
chat_provider_id="provider-1",
|
||||
prompt="hello",
|
||||
system_prompt="be helpful",
|
||||
contexts=[{"role": "user", "content": "hi"}],
|
||||
)
|
||||
|
||||
mock_llm.chat_raw.assert_called_once()
|
||||
assert result is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_loop_agent_delegates_to_chat_raw(self):
|
||||
"""tool_loop_agent() should delegate to ctx.llm.chat_raw()."""
|
||||
mock_llm = AsyncMock()
|
||||
mock_llm.chat_raw = AsyncMock(return_value=MagicMock(text="response"))
|
||||
|
||||
mock_ctx = MagicMock()
|
||||
mock_ctx.llm = mock_llm
|
||||
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
legacy_ctx._runtime_context = mock_ctx
|
||||
|
||||
result = await legacy_ctx.tool_loop_agent(
|
||||
chat_provider_id="provider-1",
|
||||
prompt="hello",
|
||||
max_steps=10,
|
||||
)
|
||||
|
||||
mock_llm.chat_raw.assert_called_once()
|
||||
call_kwargs = mock_llm.chat_raw.call_args[1]
|
||||
assert call_kwargs["max_steps"] == 10
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_llm_tools_returns_none(self):
|
||||
"""add_llm_tools() should return None (deprecated)."""
|
||||
legacy_ctx = LegacyContext("test_plugin")
|
||||
|
||||
result = await legacy_ctx.add_llm_tools("tool1", "tool2")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestCommandComponent:
|
||||
"""Tests for CommandComponent."""
|
||||
|
||||
def test_is_star_subclass(self):
|
||||
"""CommandComponent should be a Star subclass."""
|
||||
assert issubclass(CommandComponent, Star)
|
||||
|
||||
def test_is_not_new_star(self):
|
||||
"""CommandComponent should NOT be recognized as new-style star."""
|
||||
assert CommandComponent.__astrbot_is_new_star__() is False
|
||||
|
||||
def test_create_legacy_context(self):
|
||||
"""_astrbot_create_legacy_context() should create LegacyContext."""
|
||||
ctx = CommandComponent._astrbot_create_legacy_context("my_plugin")
|
||||
assert isinstance(ctx, LegacyContext)
|
||||
assert ctx.plugin_id == "my_plugin"
|
||||
|
||||
|
||||
class TestMigrationDocUrl:
|
||||
"""Tests for migration documentation URL."""
|
||||
|
||||
def test_migration_doc_url_exists(self):
|
||||
"""MIGRATION_DOC_URL should be defined."""
|
||||
assert MIGRATION_DOC_URL is not None
|
||||
assert "docs.astrbot.app" in MIGRATION_DOC_URL
|
||||
77
tests_v4/test_api_modules.py
Normal file
77
tests_v4/test_api_modules.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Tests for API module exports and re-exports.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestApiStarModule:
|
||||
"""Tests for api/star module exports."""
|
||||
|
||||
def test_star_module_exports_context(self):
|
||||
"""api.star should export Context."""
|
||||
from astrbot_sdk.api.star import Context
|
||||
|
||||
assert Context is not None
|
||||
|
||||
def test_star_context_is_legacy_context(self):
|
||||
"""api.star.Context should be LegacyContext."""
|
||||
from astrbot_sdk._legacy_api import LegacyContext
|
||||
from astrbot_sdk.api.star import Context
|
||||
|
||||
assert Context is LegacyContext
|
||||
|
||||
|
||||
class TestApiComponentsModule:
|
||||
"""Tests for api/components module exports."""
|
||||
|
||||
def test_components_module_exports_command_component(self):
|
||||
"""api.components should export CommandComponent."""
|
||||
from astrbot_sdk.api.components import CommandComponent
|
||||
|
||||
assert CommandComponent is not None
|
||||
|
||||
def test_command_component_is_from_legacy_api(self):
|
||||
"""api.components.CommandComponent should be from _legacy_api."""
|
||||
from astrbot_sdk._legacy_api import CommandComponent as LegacyCommandComponent
|
||||
from astrbot_sdk.api.components import CommandComponent
|
||||
|
||||
assert CommandComponent is LegacyCommandComponent
|
||||
|
||||
|
||||
class TestApiEventModule:
|
||||
"""Tests for api/event module exports."""
|
||||
|
||||
def test_event_module_exports(self):
|
||||
"""api.event should export expected names."""
|
||||
from astrbot_sdk.api.event import ADMIN, AstrMessageEvent, filter
|
||||
|
||||
assert ADMIN == "admin"
|
||||
assert filter is not None
|
||||
assert AstrMessageEvent is not None
|
||||
|
||||
def test_astr_message_event_is_message_event(self):
|
||||
"""AstrMessageEvent should be MessageEvent."""
|
||||
from astrbot_sdk.api.event import AstrMessageEvent
|
||||
from astrbot_sdk.events import MessageEvent
|
||||
|
||||
assert AstrMessageEvent is MessageEvent
|
||||
|
||||
def test_all_exports(self):
|
||||
"""api.event should export all expected names."""
|
||||
from astrbot_sdk.api.event import __all__
|
||||
|
||||
assert "ADMIN" in __all__
|
||||
assert "AstrMessageEvent" in __all__
|
||||
assert "filter" in __all__
|
||||
|
||||
|
||||
class TestApiModule:
|
||||
"""Tests for top-level api module."""
|
||||
|
||||
def test_api_module_exists(self):
|
||||
"""api module should be importable."""
|
||||
import astrbot_sdk.api
|
||||
|
||||
assert astrbot_sdk.api is not None
|
||||
331
tests_v4/test_decorators.py
Normal file
331
tests_v4/test_decorators.py
Normal file
@@ -0,0 +1,331 @@
|
||||
"""
|
||||
Tests for decorators.py - Handler decorator infrastructure.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from astrbot_sdk.decorators import (
|
||||
HANDLER_META_ATTR,
|
||||
HandlerMeta,
|
||||
get_handler_meta,
|
||||
on_command,
|
||||
on_event,
|
||||
on_message,
|
||||
on_schedule,
|
||||
require_admin,
|
||||
)
|
||||
from astrbot_sdk.protocol.descriptors import (
|
||||
CommandTrigger,
|
||||
EventTrigger,
|
||||
MessageTrigger,
|
||||
Permissions,
|
||||
ScheduleTrigger,
|
||||
)
|
||||
|
||||
|
||||
class TestHandlerMeta:
|
||||
"""Tests for HandlerMeta dataclass."""
|
||||
|
||||
def test_default_values(self):
|
||||
"""HandlerMeta should have default values."""
|
||||
meta = HandlerMeta()
|
||||
assert meta.trigger is None
|
||||
assert meta.priority == 0
|
||||
assert isinstance(meta.permissions, Permissions)
|
||||
|
||||
def test_trigger_assignment(self):
|
||||
"""HandlerMeta should accept trigger assignment."""
|
||||
trigger = CommandTrigger(command="test")
|
||||
meta = HandlerMeta(trigger=trigger)
|
||||
assert meta.trigger is trigger
|
||||
|
||||
def test_priority_assignment(self):
|
||||
"""HandlerMeta should accept priority assignment."""
|
||||
meta = HandlerMeta(priority=10)
|
||||
assert meta.priority == 10
|
||||
|
||||
def test_permissions_assignment(self):
|
||||
"""HandlerMeta should accept permissions assignment."""
|
||||
permissions = Permissions(require_admin=True)
|
||||
meta = HandlerMeta(permissions=permissions)
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
|
||||
class TestGetHandlerMeta:
|
||||
"""Tests for get_handler_meta function."""
|
||||
|
||||
def test_returns_none_for_undecorated_function(self):
|
||||
"""get_handler_meta should return None for undecorated functions."""
|
||||
async def plain_function():
|
||||
pass
|
||||
|
||||
assert get_handler_meta(plain_function) is None
|
||||
|
||||
def test_returns_meta_for_decorated_function(self):
|
||||
"""get_handler_meta should return meta for decorated functions."""
|
||||
|
||||
@on_command("test")
|
||||
async def decorated():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(decorated)
|
||||
assert meta is not None
|
||||
assert isinstance(meta, HandlerMeta)
|
||||
|
||||
|
||||
class TestOnCommandDecorator:
|
||||
"""Tests for @on_command decorator."""
|
||||
|
||||
def test_sets_handler_meta_attribute(self):
|
||||
"""@on_command should set __astrbot_handler_meta__ attribute."""
|
||||
|
||||
@on_command("hello")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
assert hasattr(handler, HANDLER_META_ATTR)
|
||||
|
||||
def test_creates_command_trigger(self):
|
||||
"""@on_command should create CommandTrigger."""
|
||||
|
||||
@on_command("hello")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, CommandTrigger)
|
||||
assert meta.trigger.command == "hello"
|
||||
|
||||
def test_supports_aliases(self):
|
||||
"""@on_command should store aliases."""
|
||||
|
||||
@on_command("hello", aliases=["hi", "hey"])
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.aliases == ["hi", "hey"]
|
||||
|
||||
def test_supports_description(self):
|
||||
"""@on_command should store description."""
|
||||
|
||||
@on_command("hello", description="Say hello")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.description == "Say hello"
|
||||
|
||||
def test_default_empty_aliases(self):
|
||||
"""@on_command should default to empty aliases."""
|
||||
|
||||
@on_command("test")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.aliases == []
|
||||
|
||||
def test_preserves_function(self):
|
||||
"""@on_command should preserve the original function."""
|
||||
|
||||
@on_command("test")
|
||||
async def handler():
|
||||
return "result"
|
||||
|
||||
assert handler.__name__ == "handler"
|
||||
|
||||
|
||||
class TestOnMessageDecorator:
|
||||
"""Tests for @on_message decorator."""
|
||||
|
||||
def test_creates_message_trigger(self):
|
||||
"""@on_message should create MessageTrigger."""
|
||||
|
||||
@on_message()
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, MessageTrigger)
|
||||
|
||||
def test_supports_regex(self):
|
||||
"""@on_message should store regex pattern."""
|
||||
|
||||
@on_message(regex=r"\d+")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.regex == r"\d+"
|
||||
|
||||
def test_supports_keywords(self):
|
||||
"""@on_message should store keywords."""
|
||||
|
||||
@on_message(keywords=["hello", "hi"])
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.keywords == ["hello", "hi"]
|
||||
|
||||
def test_supports_platforms(self):
|
||||
"""@on_message should store platforms."""
|
||||
|
||||
@on_message(platforms=["telegram", "discord"])
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.platforms == ["telegram", "discord"]
|
||||
|
||||
def test_defaults_empty_lists(self):
|
||||
"""@on_message should default to empty lists."""
|
||||
|
||||
@on_message()
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.keywords == []
|
||||
assert meta.trigger.platforms == []
|
||||
|
||||
|
||||
class TestOnEventDecorator:
|
||||
"""Tests for @on_event decorator."""
|
||||
|
||||
def test_creates_event_trigger(self):
|
||||
"""@on_event should create EventTrigger."""
|
||||
|
||||
@on_event("message_received")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, EventTrigger)
|
||||
assert meta.trigger.event_type == "message_received"
|
||||
|
||||
def test_various_event_types(self):
|
||||
"""@on_event should handle various event types."""
|
||||
|
||||
event_types = ["message_received", "user_joined", "custom_event"]
|
||||
|
||||
for event_type in event_types:
|
||||
@on_event(event_type)
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.event_type == event_type
|
||||
|
||||
|
||||
class TestOnScheduleDecorator:
|
||||
"""Tests for @on_schedule decorator."""
|
||||
|
||||
def test_creates_cron_trigger(self):
|
||||
"""@on_schedule with cron should create ScheduleTrigger."""
|
||||
|
||||
@on_schedule(cron="* * * * *")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, ScheduleTrigger)
|
||||
assert meta.trigger.cron == "* * * * *"
|
||||
|
||||
def test_creates_interval_trigger(self):
|
||||
"""@on_schedule with interval should create ScheduleTrigger."""
|
||||
|
||||
@on_schedule(interval_seconds=60)
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.trigger.interval_seconds == 60
|
||||
|
||||
|
||||
class TestRequireAdminDecorator:
|
||||
"""Tests for @require_admin decorator."""
|
||||
|
||||
def test_sets_require_admin_permission(self):
|
||||
"""@require_admin should set require_admin in permissions."""
|
||||
|
||||
@require_admin
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
def test_can_combine_with_other_decorators(self):
|
||||
"""@require_admin can be combined with other decorators."""
|
||||
|
||||
@on_command("admin")
|
||||
@require_admin
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, CommandTrigger)
|
||||
assert meta.trigger.command == "admin"
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
def test_order_does_not_matter_for_permissions(self):
|
||||
"""Decorator order should not affect permissions."""
|
||||
|
||||
@require_admin
|
||||
@on_command("admin")
|
||||
async def handler1():
|
||||
pass
|
||||
|
||||
@on_command("admin")
|
||||
@require_admin
|
||||
async def handler2():
|
||||
pass
|
||||
|
||||
meta1 = get_handler_meta(handler1)
|
||||
meta2 = get_handler_meta(handler2)
|
||||
|
||||
assert meta1.permissions.require_admin is True
|
||||
assert meta2.permissions.require_admin is True
|
||||
|
||||
|
||||
class TestDecoratorChaining:
|
||||
"""Tests for chaining multiple decorators."""
|
||||
|
||||
def test_multiple_decorators_share_meta(self):
|
||||
"""Multiple decorators should share the same meta object."""
|
||||
|
||||
@on_command("test")
|
||||
@require_admin
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert isinstance(meta.trigger, CommandTrigger)
|
||||
assert meta.permissions.require_admin is True
|
||||
|
||||
def test_last_trigger_wins(self):
|
||||
"""Last trigger decorator should override previous ones."""
|
||||
|
||||
# Decorators are applied bottom-up, so on_message wins
|
||||
@on_message()
|
||||
@on_command("override")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
# on_message is applied last (outermost), so it wins
|
||||
assert isinstance(meta.trigger, MessageTrigger)
|
||||
|
||||
def test_permissions_accumulate(self):
|
||||
"""Permissions should accumulate across decorators."""
|
||||
|
||||
@require_admin
|
||||
@on_command("test")
|
||||
async def handler():
|
||||
pass
|
||||
|
||||
meta = get_handler_meta(handler)
|
||||
assert meta.permissions.require_admin is True
|
||||
Reference in New Issue
Block a user