test: enhance test framework with comprehensive fixtures and mocks

.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

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")