diff --git a/.github/workflows/coverage_test.yml b/.github/workflows/coverage_test.yml index 6ae8c7b9bb..f0019ee7e6 100644 --- a/.github/workflows/coverage_test.yml +++ b/.github/workflows/coverage_test.yml @@ -37,7 +37,7 @@ jobs: mkdir -p data/temp export TESTING=true export ZHIPU_API_KEY=${{ secrets.OPENAI_API_KEY }} - pytest --cov=. -v -o log_cli=true -o log_level=DEBUG + pytest --cov=astrbot -v -o log_cli=true -o log_level=DEBUG - name: Upload results to Codecov uses: codecov/codecov-action@v5 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000000..b9807c1ded --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,381 @@ +""" +AstrBot 测试配置 + +提供共享的 pytest fixtures 和测试工具。 +""" + +import json +import os +import sys +from asyncio import Queue +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio + +# 使用 tests/fixtures/helpers.py 中的共享工具函数，避免重复定义 +from tests.fixtures.helpers import create_mock_llm_response, create_mock_message_component + +# 将项目根目录添加到 sys.path +PROJECT_ROOT = Path(__file__).parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +# 设置测试环境变量 +os.environ.setdefault("TESTING", "true") +os.environ.setdefault("ASTRBOT_TEST_MODE", "true") + + +# ============================================================ +# 测试收集和排序 +# ============================================================ + + +def pytest_collection_modifyitems(session, config, items): # noqa: ARG001 + """重新排序测试：单元测试优先，集成测试在后。""" + unit_tests = [] + integration_tests = [] + deselected = [] + profile = config.getoption("--test-profile") or os.environ.get( + "ASTRBOT_TEST_PROFILE", "all" + ) + + for item in items: + item_path = Path(str(item.path)) + is_integration = "integration" in item_path.parts + + if is_integration: + if item.get_closest_marker("integration") is None: + item.add_marker(pytest.mark.integration) + item.add_marker(pytest.mark.tier_d) + integration_tests.append(item) + else: + if item.get_closest_marker("unit") is None: + item.add_marker(pytest.mark.unit) + if any( + item.get_closest_marker(marker) is not None + for marker in ("platform", "provider", "slow") + ): + item.add_marker(pytest.mark.tier_c) + unit_tests.append(item) + + # 单元测试 -> 集成测试 + ordered_items = unit_tests + integration_tests + if profile == "blocking": + selected_items = [] + for item in ordered_items: + if item.get_closest_marker("tier_c") or item.get_closest_marker("tier_d"): + deselected.append(item) + else: + selected_items.append(item) + if deselected: + config.hook.pytest_deselected(items=deselected) + items[:] = selected_items + return + + items[:] = ordered_items + + +def pytest_addoption(parser): + """增加测试执行档位选择。""" + parser.addoption( + "--test-profile", + action="store", + default=None, + choices=["all", "blocking"], + help="Select test profile. 'blocking' excludes auto-classified tier_c/tier_d tests.", + ) + + +def pytest_configure(config): + """注册自定义标记。""" + config.addinivalue_line("markers", "unit: 单元测试") + config.addinivalue_line("markers", "integration: 集成测试") + config.addinivalue_line("markers", "slow: 慢速测试") + config.addinivalue_line("markers", "platform: 平台适配器测试") + config.addinivalue_line("markers", "provider: LLM Provider 测试") + config.addinivalue_line("markers", "db: 数据库相关测试") + config.addinivalue_line("markers", "tier_c: C-tier tests (optional / non-blocking)") + config.addinivalue_line("markers", "tier_d: D-tier tests (extended / integration)") + + +# ============================================================ +# 临时目录和文件 Fixtures +# ============================================================ + + +@pytest.fixture +def temp_dir(tmp_path: Path) -> Path: + """创建临时目录用于测试。""" + return tmp_path + + +@pytest.fixture +def event_queue() -> Queue: + """Create a shared asyncio queue fixture for tests.""" + return Queue() + + +@pytest.fixture +def platform_settings() -> dict: + """Create a shared empty platform settings fixture for adapter tests.""" + return {} + + +@pytest.fixture +def temp_data_dir(temp_dir: Path) -> Path: + """创建模拟的 data 目录结构。""" + data_dir = temp_dir / "data" + data_dir.mkdir() + + # 创建必要的子目录 + (data_dir / "config").mkdir() + (data_dir / "plugins").mkdir() + (data_dir / "temp").mkdir() + (data_dir / "attachments").mkdir() + + return data_dir + + +@pytest.fixture +def temp_config_file(temp_data_dir: Path) -> Path: + """创建临时配置文件。""" + config_path = temp_data_dir / "config" / "cmd_config.json" + default_config = { + "provider": [], + "platform": [], + "provider_settings": {}, + "default_personality": None, + "timezone": "Asia/Shanghai", + } + config_path.write_text(json.dumps(default_config, indent=2), encoding="utf-8") + return config_path + + +@pytest.fixture +def temp_db_file(temp_data_dir: Path) -> Path: + """创建临时数据库文件路径。""" + return temp_data_dir / "test.db" + + +# ============================================================ +# Mock Fixtures +# ============================================================ + + +@pytest.fixture +def mock_provider(): + """创建模拟的 Provider。""" + provider = MagicMock() + provider.provider_config = { + "id": "test-provider", + "type": "openai_chat_completion", + "model": "gpt-4o-mini", + } + provider.get_model = MagicMock(return_value="gpt-4o-mini") + provider.text_chat = AsyncMock() + provider.text_chat_stream = AsyncMock() + provider.terminate = AsyncMock() + return provider + + +@pytest.fixture +def mock_platform(): + """创建模拟的 Platform。""" + platform = MagicMock() + platform.platform_name = "test_platform" + platform.platform_meta = MagicMock() + platform.platform_meta.support_proactive_message = False + platform.send_message = AsyncMock() + platform.terminate = AsyncMock() + return platform + + +@pytest.fixture +def mock_conversation(): + """创建模拟的 Conversation。""" + from astrbot.core.db.po import ConversationV2 + + return ConversationV2( + conversation_id="test-conv-id", + platform_id="test_platform", + user_id="test_user", + content=[], + persona_id=None, + ) + + +@pytest.fixture +def mock_event(): + """创建模拟的 AstrMessageEvent。""" + event = MagicMock() + event.unified_msg_origin = "test_umo" + event.session_id = "test_session" + event.message_str = "Hello, world!" + event.message_obj = MagicMock() + event.message_obj.message = [] + event.message_obj.sender = MagicMock() + event.message_obj.sender.user_id = "test_user" + event.message_obj.sender.nickname = "Test User" + event.message_obj.group_id = None + event.message_obj.group = None + event.get_platform_name = MagicMock(return_value="test_platform") + event.get_platform_id = MagicMock(return_value="test_platform") + event.get_group_id = MagicMock(return_value=None) + event.get_extra = MagicMock(return_value=None) + event.set_extra = MagicMock() + event.trace = MagicMock() + event.platform_meta = MagicMock() + event.platform_meta.support_proactive_message = False + return event + + +# ============================================================ +# 配置 Fixtures +# ============================================================ + + +@pytest.fixture +def astrbot_config(temp_config_file: Path): + """创建 AstrBotConfig 实例。""" + from astrbot.core.config.astrbot_config import AstrBotConfig + + config = AstrBotConfig() + config._config_path = str(temp_config_file) # noqa: SLF001 + return config + + +@pytest.fixture +def main_agent_build_config(): + """创建 MainAgentBuildConfig 实例。""" + from astrbot.core.astr_main_agent import MainAgentBuildConfig + + return MainAgentBuildConfig( + tool_call_timeout=60, + tool_schema_mode="full", + provider_wake_prefix="", + streaming_response=True, + sanitize_context_by_modalities=False, + kb_agentic_mode=False, + file_extract_enabled=False, + context_limit_reached_strategy="truncate_by_turns", + llm_safety_mode=True, + computer_use_runtime="local", + add_cron_tools=True, + ) + + +# ============================================================ +# 数据库 Fixtures +# ============================================================ + + +@pytest_asyncio.fixture +async def temp_db(temp_db_file: Path): + """创建临时数据库实例。""" + from astrbot.core.db.sqlite import SQLiteDatabase + + db = SQLiteDatabase(str(temp_db_file)) + try: + yield db + finally: + await db.engine.dispose() + if temp_db_file.exists(): + temp_db_file.unlink() + + +# ============================================================ +# Context Fixtures +# ============================================================ + + +@pytest_asyncio.fixture +async def mock_context( + astrbot_config, + temp_db, + mock_provider, + mock_platform, +): + """创建模拟的插件上下文。""" + from asyncio import Queue + + from astrbot.core.star.context import Context + + event_queue = Queue() + + provider_manager = MagicMock() + provider_manager.get_using_provider = MagicMock(return_value=mock_provider) + provider_manager.get_provider_by_id = MagicMock(return_value=mock_provider) + + platform_manager = MagicMock() + conversation_manager = MagicMock() + message_history_manager = MagicMock() + persona_manager = MagicMock() + persona_manager.personas_v3 = [] + astrbot_config_mgr = MagicMock() + knowledge_base_manager = MagicMock() + cron_manager = MagicMock() + subagent_orchestrator = None + + context = Context( + event_queue, + astrbot_config, + temp_db, + provider_manager, + platform_manager, + conversation_manager, + message_history_manager, + persona_manager, + astrbot_config_mgr, + knowledge_base_manager, + cron_manager, + subagent_orchestrator, + ) + + return context + + +# ============================================================ +# Provider Request Fixtures +# ============================================================ + + +@pytest.fixture +def provider_request(): + """创建 ProviderRequest 实例。""" + from astrbot.core.provider.entities import ProviderRequest + + return ProviderRequest( + prompt="Hello", + session_id="test_session", + image_urls=[], + contexts=[], + system_prompt="You are a helpful assistant.", + ) + + +# ============================================================ +# 跳过条件 +# ============================================================ + + +def pytest_runtest_setup(item): + """在测试运行前检查跳过条件。""" + # 跳过需要 API Key 但未设置的 Provider 测试 + if item.get_closest_marker("provider"): + if not os.environ.get("TEST_PROVIDER_API_KEY"): + pytest.skip("TEST_PROVIDER_API_KEY not set") + + # 跳过需要特定平台的测试 + if item.get_closest_marker("platform"): + required_platform = None + marker = item.get_closest_marker("platform") + if marker and marker.args: + required_platform = marker.args[0] + + if required_platform and not os.environ.get( + f"TEST_{required_platform.upper()}_ENABLED" + ): + pytest.skip(f"TEST_{required_platform.upper()}_ENABLED not set") diff --git a/tests/fixtures/__init__.py b/tests/fixtures/__init__.py new file mode 100644 index 0000000000..16e927d2cf --- /dev/null +++ b/tests/fixtures/__init__.py @@ -0,0 +1,64 @@ +""" +AstrBot 测试数据 + +此目录存放测试用的静态数据和配置文件。 + +目录结构: +- fixtures/ + ├── configs/ # 测试配置文件 + ├── messages/ # 测试消息数据 + ├── plugins/ # 测试插件 + ├── knowledge_base/ # 测试知识库数据 + ├── mocks/ # Mock 模块 + └── helpers.py # 辅助函数 +""" + +import json +from pathlib import Path + +from .helpers import ( + NoopAwaitable, + create_mock_discord_attachment, + create_mock_discord_channel, + create_mock_discord_user, + create_mock_file, + create_mock_llm_response, + create_mock_message_component, + create_mock_update, + make_platform_config, +) + +FIXTURES_DIR = Path(__file__).parent + + +def load_fixture(filename: str) -> dict: + """加载 JSON 格式的测试数据。""" + filepath = FIXTURES_DIR / filename + if not filepath.exists(): + raise FileNotFoundError(f"Fixture not found: {filepath}") + return json.loads(filepath.read_text(encoding="utf-8")) + + +def get_fixture_path(filename: str) -> Path: + """获取测试数据文件路径。""" + filepath = FIXTURES_DIR / filename + if not filepath.exists(): + raise FileNotFoundError(f"Fixture not found: {filepath}") + return filepath + + +__all__ = [ + "FIXTURES_DIR", + "load_fixture", + "get_fixture_path", + # 辅助函数 + "NoopAwaitable", + "make_platform_config", + "create_mock_update", + "create_mock_file", + "create_mock_discord_attachment", + "create_mock_discord_user", + "create_mock_discord_channel", + "create_mock_message_component", + "create_mock_llm_response", +] diff --git a/tests/fixtures/configs/test_cmd_config.json b/tests/fixtures/configs/test_cmd_config.json new file mode 100644 index 0000000000..2b92302a4b --- /dev/null +++ b/tests/fixtures/configs/test_cmd_config.json @@ -0,0 +1,21 @@ +{ + "provider": [ + { + "id": "test-openai", + "type": "openai_chat_completion", + "model": "gpt-4o-mini", + "key": ["test-key"] + } + ], + "platform": [], + "provider_settings": { + "default_personality": null, + "prompt_prefix": "", + "image_caption_provider_id": "", + "datetime_system_prompt": true, + "identifier": true, + "group_name_display": true + }, + "default_personality": null, + "timezone": "Asia/Shanghai" +} diff --git a/tests/fixtures/helpers.py b/tests/fixtures/helpers.py new file mode 100644 index 0000000000..8f64ab6c97 --- /dev/null +++ b/tests/fixtures/helpers.py @@ -0,0 +1,332 @@ +"""测试辅助函数和工具类。 + +提供统一的测试辅助工具，减少测试代码重复。 +""" + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from astrbot.core.message.components import BaseMessageComponent + + +class NoopAwaitable: + """可等待的空操作对象。 + + 用于 mock 需要返回 awaitable 对象的方法。 + """ + + def __await__(self): + if False: + yield + return None + + +# ============================================================ +# 平台配置工厂 +# ============================================================ + + +def make_platform_config(platform_type: str, **kwargs) -> dict: + """平台配置工厂函数。 + + Args: + platform_type: 平台类型 (telegram, discord, aiocqhttp 等) + **kwargs: 覆盖默认配置的字段 + + Returns: + dict: 平台配置字典 + """ + configs = { + "telegram": { + "id": "test_telegram", + "telegram_token": "test_token_123", + "telegram_api_base_url": "https://api.telegram.org/bot", + "telegram_file_base_url": "https://api.telegram.org/file/bot", + "telegram_command_register": True, + "telegram_command_auto_refresh": True, + "telegram_command_register_interval": 300, + "telegram_media_group_timeout": 2.5, + "telegram_media_group_max_wait": 10.0, + "start_message": "Welcome to AstrBot!", + }, + "discord": { + "id": "test_discord", + "discord_token": "test_token_123", + "discord_proxy": None, + "discord_command_register": True, + "discord_guild_id_for_debug": None, + "discord_activity_name": "Playing AstrBot", + }, + "aiocqhttp": { + "id": "test_aiocqhttp", + "ws_reverse_host": "0.0.0.0", + "ws_reverse_port": 6199, + "ws_reverse_token": "test_token", + }, + "webchat": { + "id": "test_webchat", + }, + "wecom": { + "id": "test_wecom", + "wecom_corpid": "test_corpid", + "wecom_secret": "test_secret", + }, + } + config = configs.get(platform_type, {"id": f"test_{platform_type}"}).copy() + config.update(kwargs) + return config + + +# ============================================================ +# Telegram 辅助函数 +# ============================================================ + + +def create_mock_update( + message_text: str | None = "Hello World", + chat_type: str = "private", + chat_id: int = 123456789, + user_id: int = 987654321, + username: str = "test_user", + message_id: int = 1, + media_group_id: str | None = None, + photo: list | None = None, + video: MagicMock | None = None, + document: MagicMock | None = None, + voice: MagicMock | None = None, + sticker: MagicMock | None = None, + reply_to_message: MagicMock | None = None, + caption: str | None = None, + entities: list | None = None, + caption_entities: list | None = None, + message_thread_id: int | None = None, + is_topic_message: bool = False, +): + """创建模拟的 Telegram Update 对象。 + + Args: + message_text: 消息文本 + chat_type: 聊天类型 + chat_id: 聊天 ID + user_id: 用户 ID + username: 用户名 + message_id: 消息 ID + media_group_id: 媒体组 ID + photo: 图片列表 + video: 视频对象 + document: 文档对象 + voice: 语音对象 + sticker: 贴纸对象 + reply_to_message: 回复的消息 + caption: 说明文字 + entities: 实体列表 + caption_entities: 说明实体列表 + message_thread_id: 消息线程 ID + is_topic_message: 是否为主题消息 + + Returns: + MagicMock: 模拟的 Update 对象 + """ + update = MagicMock() + update.update_id = 1 + + # Create message mock + message = MagicMock() + message.message_id = message_id + message.chat = MagicMock() + message.chat.id = chat_id + message.chat.type = chat_type + message.message_thread_id = message_thread_id + message.is_topic_message = is_topic_message + + # Create user mock + from_user = MagicMock() + from_user.id = user_id + from_user.username = username + message.from_user = from_user + + # Set message content + message.text = message_text + message.media_group_id = media_group_id + message.photo = photo + message.video = video + message.document = document + message.voice = voice + message.sticker = sticker + message.reply_to_message = reply_to_message + message.caption = caption + message.entities = entities + message.caption_entities = caption_entities + + update.message = message + update.effective_chat = message.chat + + return update + + +def create_mock_file(file_path: str = "https://api.telegram.org/file/test.jpg"): + """创建模拟的 Telegram File 对象。 + + Args: + file_path: 文件路径 + + Returns: + MagicMock: 模拟的 File 对象 + """ + file = MagicMock() + file.file_path = file_path + file.get_file = AsyncMock(return_value=file) + return file + + +# ============================================================ +# Discord 辅助函数 +# ============================================================ + + +def create_mock_discord_attachment( + filename: str = "test.txt", + url: str = "https://cdn.discordapp.com/test.txt", + content_type: str | None = None, + size: int = 1024, +): + """创建模拟的 Discord Attachment 对象。 + + Args: + filename: 文件名 + url: 文件 URL + content_type: 内容类型 + size: 文件大小 + + Returns: + MagicMock: 模拟的 Attachment 对象 + """ + attachment = MagicMock() + attachment.filename = filename + attachment.url = url + attachment.content_type = content_type + attachment.size = size + return attachment + + +def create_mock_discord_user( + user_id: int = 123456789, + name: str = "TestUser", + display_name: str = "Test User", + bot: bool = False, +): + """创建模拟的 Discord User 对象。 + + Args: + user_id: 用户 ID + name: 用户名 + display_name: 显示名 + bot: 是否为机器人 + + Returns: + MagicMock: 模拟的 User 对象 + """ + user = MagicMock() + user.id = user_id + user.name = name + user.display_name = display_name + user.bot = bot + user.mention = f"<@{user_id}>" + return user + + +def create_mock_discord_channel( + channel_id: int = 111222333, + channel_type: str = "text", + name: str = "general", + guild_id: int | None = 444555666, +): + """创建模拟的 Discord Channel 对象。 + + Args: + channel_id: 频道 ID + channel_type: 频道类型 + name: 频道名 + guild_id: 服务器 ID + + Returns: + MagicMock: 模拟的 Channel 对象 + """ + channel = MagicMock() + channel.id = channel_id + channel.name = name + channel.type = channel_type + + if guild_id: + channel.guild = MagicMock() + channel.guild.id = guild_id + else: + channel.guild = None + + return channel + + +# ============================================================ +# 消息组件辅助函数 +# ============================================================ + + +def create_mock_message_component( + component_type: str, + **kwargs: Any, +) -> BaseMessageComponent: + """创建模拟的消息组件。 + + Args: + component_type: 组件类型 (plain, image, at, reply, file) + **kwargs: 组件参数 + + Returns: + BaseMessageComponent: 消息组件实例 + """ + from astrbot.core.message import components as Comp + + component_map = { + "plain": Comp.Plain, + "image": Comp.Image, + "at": Comp.At, + "reply": Comp.Reply, + "file": Comp.File, + } + + component_class = component_map.get(component_type.lower()) + if not component_class: + raise ValueError(f"Unknown component type: {component_type}") + + return component_class(**kwargs) + + +def create_mock_llm_response( + completion_text: str = "Hello! How can I help you?", + role: str = "assistant", + tools_call_name: list[str] | None = None, + tools_call_args: list[dict] | None = None, + tools_call_ids: list[str] | None = None, +): + """创建模拟的 LLM 响应。 + + Args: + completion_text: 完成文本 + role: 角色 + tools_call_name: 工具调用名称列表 + tools_call_args: 工具调用参数列表 + tools_call_ids: 工具调用 ID 列表 + + Returns: + LLMResponse: 模拟的 LLM 响应 + """ + from astrbot.core.provider.entities import LLMResponse, TokenUsage + + return LLMResponse( + role=role, + completion_text=completion_text, + tools_call_name=tools_call_name or [], + tools_call_args=tools_call_args or [], + tools_call_ids=tools_call_ids or [], + usage=TokenUsage(input_other=10, output=5), + ) diff --git a/tests/fixtures/messages/test_messages.json b/tests/fixtures/messages/test_messages.json new file mode 100644 index 0000000000..0a3a7073f2 --- /dev/null +++ b/tests/fixtures/messages/test_messages.json @@ -0,0 +1,33 @@ +{ + "plain_message": { + "type": "plain", + "text": "Hello, this is a test message." + }, + "image_message": { + "type": "image", + "url": "https://example.com/test.jpg", + "file": null + }, + "at_message": { + "type": "at", + "user_id": "12345", + "nickname": "TestUser" + }, + "reply_message": { + "type": "reply", + "id": "msg_123", + "sender_nickname": "OriginalSender", + "message_str": "This is the original message" + }, + "file_message": { + "type": "file", + "name": "test.pdf", + "url": "https://example.com/test.pdf" + }, + "combined_message": { + "components": [ + {"type": "at", "user_id": "bot_id"}, + {"type": "plain", "text": " Hello bot!"} + ] + } +} diff --git a/tests/fixtures/mocks/__init__.py b/tests/fixtures/mocks/__init__.py new file mode 100644 index 0000000000..c6497f1f2b --- /dev/null +++ b/tests/fixtures/mocks/__init__.py @@ -0,0 +1,43 @@ +"""测试 Mock 模块。 + +提供统一的 mock 工具和 fixture，减少测试代码重复。 + +使用方式: + # 在测试文件顶部导入需要的 fixture + from tests.fixtures.mocks import mock_telegram_modules + + # 或使用 Builder 类创建 mock 对象 + from tests.fixtures.mocks import MockTelegramBuilder + bot = MockTelegramBuilder.create_bot() +""" + +from .aiocqhttp import ( + MockAiocqhttpBuilder, + create_mock_aiocqhttp_modules, + mock_aiocqhttp_modules, +) +from .discord import ( + MockDiscordBuilder, + create_mock_discord_modules, + mock_discord_modules, +) +from .telegram import ( + MockTelegramBuilder, + create_mock_telegram_modules, + mock_telegram_modules, +) + +__all__ = [ + # Telegram + "mock_telegram_modules", + "create_mock_telegram_modules", + "MockTelegramBuilder", + # Discord + "mock_discord_modules", + "create_mock_discord_modules", + "MockDiscordBuilder", + # Aiocqhttp + "mock_aiocqhttp_modules", + "create_mock_aiocqhttp_modules", + "MockAiocqhttpBuilder", +] diff --git a/tests/fixtures/mocks/aiocqhttp.py b/tests/fixtures/mocks/aiocqhttp.py new file mode 100644 index 0000000000..d5e3c8229e --- /dev/null +++ b/tests/fixtures/mocks/aiocqhttp.py @@ -0,0 +1,58 @@ +"""Aiocqhttp 模块 Mock 工具。 + +提供统一的 aiocqhttp 相关模块 mock 设置，避免在测试文件中重复定义。 +""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +def create_mock_aiocqhttp_modules(): + """创建 aiocqhttp 相关的 mock 模块。 + + Returns: + dict: 包含 aiocqhttp 和相关模块的 mock 对象 + """ + mock_aiocqhttp = MagicMock() + mock_aiocqhttp.CQHttp = MagicMock + mock_aiocqhttp.Event = MagicMock + mock_aiocqhttp.exceptions = MagicMock() + mock_aiocqhttp.exceptions.ActionFailed = Exception + + return mock_aiocqhttp + + +@pytest.fixture(scope="module", autouse=True) +def mock_aiocqhttp_modules(): + """Mock aiocqhttp 相关模块的 fixture。 + + 自动应用于使用此 fixture 的测试模块。 + """ + mock_aiocqhttp = create_mock_aiocqhttp_modules() + monkeypatch = pytest.MonkeyPatch() + + monkeypatch.setitem(sys.modules, "aiocqhttp", mock_aiocqhttp) + monkeypatch.setitem(sys.modules, "aiocqhttp.exceptions", mock_aiocqhttp.exceptions) + yield + monkeypatch.undo() + + +class MockAiocqhttpBuilder: + """构建 aiocqhttp 测试 mock 对象的工具类。""" + + @staticmethod + def create_bot(): + """创建 mock CQHttp bot 实例。""" + from tests.fixtures.helpers import NoopAwaitable + + bot = MagicMock() + bot.send = AsyncMock() + bot.call_action = AsyncMock() + bot.on_request = MagicMock() + bot.on_notice = MagicMock() + bot.on_message = MagicMock() + bot.on_websocket_connection = MagicMock() + bot.run_task = MagicMock(return_value=NoopAwaitable()) + return bot diff --git a/tests/fixtures/mocks/discord.py b/tests/fixtures/mocks/discord.py new file mode 100644 index 0000000000..e13786af17 --- /dev/null +++ b/tests/fixtures/mocks/discord.py @@ -0,0 +1,140 @@ +"""Discord 模块 Mock 工具。 + +提供统一的 Discord 相关模块 mock 设置，避免在测试文件中重复定义。 +""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +def create_mock_discord_modules(): + """创建 Discord 相关的 mock 模块。 + + Returns: + dict: 包含 discord 和相关模块的 mock 对象 + """ + mock_discord = MagicMock() + + # Mock discord.Intents + mock_intents = MagicMock() + mock_intents.default = MagicMock(return_value=mock_intents) + mock_discord.Intents = mock_intents + + # Mock discord.Status + mock_discord.Status = MagicMock() + mock_discord.Status.online = "online" + + # Mock discord.Bot + mock_bot = MagicMock() + mock_discord.Bot = MagicMock(return_value=mock_bot) + + # Mock discord.Embed + mock_embed = MagicMock() + mock_discord.Embed = MagicMock(return_value=mock_embed) + + # Mock discord.ui + mock_ui = MagicMock() + mock_ui.View = MagicMock + mock_ui.Button = MagicMock + mock_discord.ui = mock_ui + + # Mock discord.Message + mock_discord.Message = MagicMock + + # Mock discord.Interaction + mock_discord.Interaction = MagicMock + mock_discord.InteractionType = MagicMock() + mock_discord.InteractionType.application_command = 2 + mock_discord.InteractionType.component = 3 + + # Mock discord.File + mock_discord.File = MagicMock + + # Mock discord.SlashCommand + mock_discord.SlashCommand = MagicMock + + # Mock discord.Option + mock_discord.Option = MagicMock + + # Mock discord.SlashCommandOptionType + mock_discord.SlashCommandOptionType = MagicMock() + mock_discord.SlashCommandOptionType.string = 3 + + # Mock discord.errors + mock_discord.errors = MagicMock() + mock_discord.errors.LoginFailure = Exception + mock_discord.errors.ConnectionClosed = Exception + mock_discord.errors.NotFound = Exception + mock_discord.errors.Forbidden = Exception + + # Mock discord.abc + mock_discord.abc = MagicMock() + mock_discord.abc.GuildChannel = MagicMock + mock_discord.abc.Messageable = MagicMock + mock_discord.abc.PrivateChannel = MagicMock + + # Mock discord.channel + mock_channel = MagicMock() + mock_channel.DMChannel = MagicMock + mock_discord.channel = mock_channel + + # Mock discord.types + mock_discord.types = MagicMock() + mock_discord.types.interactions = MagicMock() + + # Mock discord.ApplicationContext + mock_discord.ApplicationContext = MagicMock + + # Mock discord.CustomActivity + mock_discord.CustomActivity = MagicMock + + return mock_discord + + +@pytest.fixture(scope="module", autouse=True) +def mock_discord_modules(): + """Mock Discord 相关模块的 fixture。 + + 自动应用于使用此 fixture 的测试模块。 + """ + mock_discord = create_mock_discord_modules() + monkeypatch = pytest.MonkeyPatch() + + monkeypatch.setitem(sys.modules, "discord", mock_discord) + monkeypatch.setitem(sys.modules, "discord.abc", mock_discord.abc) + monkeypatch.setitem(sys.modules, "discord.channel", mock_discord.channel) + monkeypatch.setitem(sys.modules, "discord.errors", mock_discord.errors) + monkeypatch.setitem(sys.modules, "discord.types", mock_discord.types) + monkeypatch.setitem( + sys.modules, + "discord.types.interactions", + mock_discord.types.interactions, + ) + monkeypatch.setitem(sys.modules, "discord.ui", mock_discord.ui) + yield + monkeypatch.undo() + + +class MockDiscordBuilder: + """构建 Discord 测试 mock 对象的工具类。""" + + @staticmethod + def create_client(): + """创建 mock Discord client 实例。""" + client = MagicMock() + client.user = MagicMock() + client.user.id = 123456789 + client.user.display_name = "TestBot" + client.user.name = "TestBot" + client.get_channel = MagicMock() + client.fetch_channel = AsyncMock() + client.get_message = MagicMock() + client.start = AsyncMock() + client.close = AsyncMock() + client.is_closed = MagicMock(return_value=False) + client.add_application_command = MagicMock() + client.sync_commands = AsyncMock() + client.change_presence = AsyncMock() + return client diff --git a/tests/fixtures/mocks/telegram.py b/tests/fixtures/mocks/telegram.py new file mode 100644 index 0000000000..fbe4d04364 --- /dev/null +++ b/tests/fixtures/mocks/telegram.py @@ -0,0 +1,141 @@ +"""Telegram 模块 Mock 工具。 + +提供统一的 Telegram 相关模块 mock 设置，避免在测试文件中重复定义。 +""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +def create_mock_telegram_modules(): + """创建 Telegram 相关的 mock 模块。 + + Returns: + dict: 包含 telegram 和相关模块的 mock 对象 + """ + mock_telegram = MagicMock() + mock_telegram.BotCommand = MagicMock + mock_telegram.Update = MagicMock + mock_telegram.constants = MagicMock() + mock_telegram.constants.ChatType = MagicMock() + mock_telegram.constants.ChatType.PRIVATE = "private" + mock_telegram.constants.ChatAction = MagicMock() + mock_telegram.constants.ChatAction.TYPING = "typing" + mock_telegram.constants.ChatAction.UPLOAD_VOICE = "upload_voice" + mock_telegram.constants.ChatAction.UPLOAD_DOCUMENT = "upload_document" + mock_telegram.constants.ChatAction.UPLOAD_PHOTO = "upload_photo" + mock_telegram.error = MagicMock() + mock_telegram.error.BadRequest = Exception + mock_telegram.ReactionTypeCustomEmoji = MagicMock + mock_telegram.ReactionTypeEmoji = MagicMock + + mock_telegram_ext = MagicMock() + mock_telegram_ext.ApplicationBuilder = MagicMock + mock_telegram_ext.ContextTypes = MagicMock + mock_telegram_ext.ExtBot = MagicMock + mock_telegram_ext.filters = MagicMock() + mock_telegram_ext.filters.ALL = MagicMock() + mock_telegram_ext.MessageHandler = MagicMock + + # Mock telegramify_markdown + mock_telegramify = MagicMock() + mock_telegramify.markdownify = lambda text, **kwargs: text + + # Mock apscheduler + mock_apscheduler = MagicMock() + mock_apscheduler.schedulers = MagicMock() + mock_apscheduler.schedulers.asyncio = MagicMock() + mock_apscheduler.schedulers.asyncio.AsyncIOScheduler = MagicMock + mock_apscheduler.schedulers.background = MagicMock() + mock_apscheduler.schedulers.background.BackgroundScheduler = MagicMock + + return { + "telegram": mock_telegram, + "telegram.ext": mock_telegram_ext, + "telegramify_markdown": mock_telegramify, + "apscheduler": mock_apscheduler, + } + + +@pytest.fixture(scope="module", autouse=True) +def mock_telegram_modules(): + """Mock Telegram 相关模块的 fixture。 + + 自动应用于使用此 fixture 的测试模块。 + """ + mocks = create_mock_telegram_modules() + monkeypatch = pytest.MonkeyPatch() + + monkeypatch.setitem(sys.modules, "telegram", mocks["telegram"]) + monkeypatch.setitem(sys.modules, "telegram.constants", mocks["telegram"].constants) + monkeypatch.setitem(sys.modules, "telegram.error", mocks["telegram"].error) + monkeypatch.setitem(sys.modules, "telegram.ext", mocks["telegram.ext"]) + monkeypatch.setitem(sys.modules, "telegramify_markdown", mocks["telegramify_markdown"]) + monkeypatch.setitem(sys.modules, "apscheduler", mocks["apscheduler"]) + monkeypatch.setitem( + sys.modules, "apscheduler.schedulers", mocks["apscheduler"].schedulers + ) + monkeypatch.setitem( + sys.modules, + "apscheduler.schedulers.asyncio", + mocks["apscheduler"].schedulers.asyncio, + ) + monkeypatch.setitem( + sys.modules, + "apscheduler.schedulers.background", + mocks["apscheduler"].schedulers.background, + ) + yield + monkeypatch.undo() + + +class MockTelegramBuilder: + """构建 Telegram 测试 mock 对象的工具类。""" + + @staticmethod + def create_bot(): + """创建 mock Telegram bot 实例。""" + bot = MagicMock() + bot.username = "test_bot" + bot.id = 12345678 + bot.base_url = "https://api.telegram.org/bottest_token_123/" + bot.send_message = AsyncMock() + bot.send_photo = AsyncMock() + bot.send_document = AsyncMock() + bot.send_voice = AsyncMock() + bot.send_chat_action = AsyncMock() + bot.delete_my_commands = AsyncMock() + bot.set_my_commands = AsyncMock() + bot.set_message_reaction = AsyncMock() + bot.edit_message_text = AsyncMock() + return bot + + @staticmethod + def create_application(): + """创建 mock Telegram Application 实例。""" + from tests.fixtures.helpers import NoopAwaitable + + app = MagicMock() + app.bot = MagicMock() + app.bot.username = "test_bot" + app.bot.base_url = "https://api.telegram.org/bottest_token_123/" + app.initialize = AsyncMock() + app.start = AsyncMock() + app.stop = AsyncMock() + app.add_handler = MagicMock() + app.updater = MagicMock() + app.updater.start_polling = MagicMock(return_value=NoopAwaitable()) + app.updater.stop = AsyncMock() + return app + + @staticmethod + def create_scheduler(): + """创建 mock APScheduler 实例。""" + scheduler = MagicMock() + scheduler.add_job = MagicMock() + scheduler.start = MagicMock() + scheduler.running = True + scheduler.shutdown = MagicMock() + return scheduler diff --git a/tests/fixtures/plugins/fixture_plugin.py b/tests/fixtures/plugins/fixture_plugin.py new file mode 100644 index 0000000000..455b5b7599 --- /dev/null +++ b/tests/fixtures/plugins/fixture_plugin.py @@ -0,0 +1,40 @@ +""" +测试插件 - 用于插件系统测试 + +这是一个最小化的测试插件，用于验证插件系统的功能。 +""" + +from astrbot.api import llm_tool, star +from astrbot.api.event import AstrMessageEvent, MessageEventResult, filter + + +@star.register("test_plugin", "AstrBot Team", "测试插件 - 用于插件系统测试", "1.0.0") +class TestPlugin(star.Star): + """测试插件类""" + + def __init__(self, context: star.Context) -> None: + super().__init__(context) + self.initialized = True + + async def terminate(self) -> None: + """插件终止""" + self.initialized = False + + @filter.command("test_cmd") + async def test_command(self, event: AstrMessageEvent) -> None: + """测试命令处理器。""" + event.set_result(MessageEventResult().message("测试命令执行成功")) + + @llm_tool("test_tool") + async def test_llm_tool(self, query: str) -> str: + """测试 LLM 工具。 + + Args: + query(string): 查询内容。 + """ + return f"测试工具执行成功: {query}" + + @filter.regex(r"^test_regex_(.+)$") + async def test_regex_handler(self, event: AstrMessageEvent) -> None: + """测试正则处理器。""" + event.set_result(MessageEventResult().message("正则匹配成功")) diff --git a/tests/fixtures/plugins/metadata.yaml b/tests/fixtures/plugins/metadata.yaml new file mode 100644 index 0000000000..2554fb15d7 --- /dev/null +++ b/tests/fixtures/plugins/metadata.yaml @@ -0,0 +1,5 @@ +name: test_plugin +description: 测试插件 - 用于插件系统测试 +version: 1.0.0 +author: AstrBot Team +repo: https://github.com/test/test_plugin