test: add comprehensive tests and improve test structure for adapters

Author: whatevertogo

AGENTS.md

@@ -28,6 +28,12 @@ Runs on `http://localhost:3000` by default.
 5. Use English for all new comments.
 6. For path handling, use `pathlib.Path` instead of string paths, and use `astrbot.core.utils.path_utils` to get the AstrBot data and temp directory.
 
+## Testing
+
+When you modify functionality, add or update a corresponding test and run it locally (e.g. `uv run pytest tests/path/to/test_xxx.py --cov=astrbot.xxx`).
+Use `--cov-report term-missing` or similar to generate coverage information.
+
+
 ## PR instructions
 
 1. Title format: use conventional commit messages

astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py

@@ -1,9 +1,10 @@
 import asyncio
+import importlib
 import itertools
 import logging
 import time
 import uuid
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Callable
 from typing import Any, cast
 
 import aiocqhttp
@@ -45,6 +46,7 @@ def __init__(
         platform_config: dict,
         platform_settings: dict,
         event_queue: asyncio.Queue,
+        bot_factory: Callable[..., Any] | None = None,
     ) -> None:
         super().__init__(platform_config, event_queue)
 
@@ -59,14 +61,7 @@ def __init__(
             support_streaming_message=False,
         )
 
-        self.bot = aiocqhttp.CQHttp(
-            use_ws_reverse=True,
-            import_name="aiocqhttp",
-            api_timeout_sec=180,
-            access_token=platform_config.get(
-                "ws_reverse_token",
-            ),  # 以防旧版本配置不存在
-        )
+        self.bot = self._create_bot(platform_config, bot_factory=bot_factory)
 
         @self.bot.on_request()
         async def request(event: aiocqhttp.Event) -> None:
@@ -113,6 +108,29 @@ async def private(event: aiocqhttp.Event) -> None:
         def on_websocket_connection(_) -> None:
             logger.info("aiocqhttp(OneBot v11) 适配器已连接。")
 
+    @staticmethod
+    def _create_bot(
+        platform_config: dict,
+        bot_factory: Callable[..., Any] | None = None,
+    ) -> aiocqhttp.CQHttp:
+        if bot_factory is None:
+            # Resolve aiocqhttp at runtime so tests that swap sys.modules later
+            # still affect bot creation even if this module was imported earlier.
+            aiocqhttp_module = importlib.import_module("aiocqhttp")
+            bot_factory = aiocqhttp_module.CQHttp
+
+        return cast(
+            aiocqhttp.CQHttp,
+            bot_factory(
+                use_ws_reverse=True,
+                import_name="aiocqhttp",
+                api_timeout_sec=180,
+                access_token=platform_config.get(
+                    "ws_reverse_token",
+                ),  # 以防旧版本配置不存在
+            ),
+        )
+
     async def send_by_session(
         self,
         session: MessageSesion,

tests/conftest.py

@@ -7,6 +7,7 @@
 import json
 import os
 import sys
+from asyncio import Queue
 from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
@@ -33,6 +34,10 @@ 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))
@@ -41,14 +46,44 @@ def pytest_collection_modifyitems(session, config, items):  # noqa: ARG001
         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)
 
     # 单元测试 -> 集成测试
-    items[:] = unit_tests + integration_tests
+    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):
@@ -59,6 +94,8 @@ def pytest_configure(config):
     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)")
 
 
 # ============================================================
@@ -72,6 +109,18 @@ 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 目录结构。"""

tests/test_api_key_open_api.py

@@ -12,7 +12,7 @@
 from astrbot.dashboard.server import AstrBotDashboard
 
 
-@pytest_asyncio.fixture(scope="module")
+@pytest_asyncio.fixture(scope="module", loop_scope="module")
 async def core_lifecycle_td(tmp_path_factory):
     tmp_db_path = tmp_path_factory.mktemp("data") / "test_data_api_key.db"
     db = SQLiteDatabase(str(tmp_db_path))
@@ -37,7 +37,7 @@ def app(core_lifecycle_td: AstrBotCoreLifecycle):
     return server.app
 
 
-@pytest_asyncio.fixture(scope="module")
+@pytest_asyncio.fixture(scope="module", loop_scope="module")
 async def authenticated_header(app: Quart, core_lifecycle_td: AstrBotCoreLifecycle):
     test_client = app.test_client()
     response = await test_client.post(
@@ -258,7 +258,7 @@ async def test_open_chat_sessions_pagination(
     assert create_data["status"] == "ok"
     raw_key = create_data["data"]["api_key"]
 
-    creator = "alice"
+    creator = f"alice_{uuid.uuid4().hex[:8]}"
     for idx in range(3):
         await core_lifecycle_td.db.create_platform_session(
             creator=creator,
@@ -276,7 +276,8 @@ async def test_open_chat_sessions_pagination(
     )
 
     page_1_res = await test_client.get(
-        "/api/v1/chat/sessions?page=1&page_size=2&username=alice",
+        "/api/v1/chat/sessions?page=1&page_size=2&username="
+        f"{creator}",
         headers={"X-API-Key": raw_key},
     )
     assert page_1_res.status_code == 200
@@ -286,10 +287,11 @@ async def test_open_chat_sessions_pagination(
     assert page_1_data["data"]["page_size"] == 2
     assert page_1_data["data"]["total"] == 3
     assert len(page_1_data["data"]["sessions"]) == 2
-    assert all(item["creator"] == "alice" for item in page_1_data["data"]["sessions"])
+    assert all(item["creator"] == creator for item in page_1_data["data"]["sessions"])
 
     page_2_res = await test_client.get(
-        "/api/v1/chat/sessions?page=2&page_size=2&username=alice",
+        "/api/v1/chat/sessions?page=2&page_size=2&username="
+        f"{creator}",
         headers={"X-API-Key": raw_key},
     )
     assert page_2_res.status_code == 200

tests/test_smoke.py

@@ -7,18 +7,32 @@
 from pathlib import Path
 
 from astrbot.core.pipeline.bootstrap import ensure_builtin_stages_registered
-from astrbot.core.pipeline.stage import Stage, registered_stages
-from astrbot.core.pipeline.stage_order import STAGES_ORDER
 from astrbot.core.pipeline.process_stage.method.agent_sub_stages.internal import (
     InternalAgentSubStage,
 )
 from astrbot.core.pipeline.process_stage.method.agent_sub_stages.third_party import (
     ThirdPartyAgentSubStage,
 )
+from astrbot.core.pipeline.stage import Stage, registered_stages
+from astrbot.core.pipeline.stage_order import STAGES_ORDER
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _run_code_in_fresh_interpreter(code: str, failure_message: str) -> None:
+    proc = subprocess.run(
+        [sys.executable, "-c", code],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert proc.returncode == 0, (
+        f"{failure_message}\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}\n"
+    )
 
 
 def test_smoke_critical_imports_in_fresh_interpreter() -> None:
-    repo_root = Path(__file__).resolve().parents[1]
     code = (
         "import importlib;"
         "mods=["
@@ -30,18 +44,7 @@ def test_smoke_critical_imports_in_fresh_interpreter() -> None:
         "];"
         "[importlib.import_module(m) for m in mods]"
     )
-    proc = subprocess.run(
-        [sys.executable, "-c", code],
-        cwd=repo_root,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert proc.returncode == 0, (
-        "Smoke import check failed.\n"
-        f"stdout:\n{proc.stdout}\n"
-        f"stderr:\n{proc.stderr}\n"
-    )
+    _run_code_in_fresh_interpreter(code, "Smoke import check failed.")
 
 
 def test_smoke_pipeline_stage_registration_matches_order() -> None:
@@ -55,3 +58,58 @@ def test_smoke_pipeline_stage_registration_matches_order() -> None:
 def test_smoke_agent_sub_stages_are_stage_subclasses() -> None:
     assert issubclass(InternalAgentSubStage, Stage)
     assert issubclass(ThirdPartyAgentSubStage, Stage)
+
+
+def test_pipeline_package_exports_remain_compatible() -> None:
+    import astrbot.core.pipeline as pipeline
+
+    assert pipeline.ProcessStage is not None
+    assert pipeline.RespondStage is not None
+    assert isinstance(pipeline.STAGES_ORDER, list)
+    assert "ProcessStage" in pipeline.STAGES_ORDER
+
+
+def test_builtin_stage_bootstrap_is_idempotent() -> None:
+    ensure_builtin_stages_registered()
+    before_count = len(registered_stages)
+    stage_names = {cls.__name__ for cls in registered_stages}
+
+    expected_stage_names = {
+        "WakingCheckStage",
+        "WhitelistCheckStage",
+        "SessionStatusCheckStage",
+        "RateLimitStage",
+        "ContentSafetyCheckStage",
+        "PreProcessStage",
+        "ProcessStage",
+        "ResultDecorateStage",
+        "RespondStage",
+    }
+
+    assert expected_stage_names.issubset(stage_names)
+
+    ensure_builtin_stages_registered()
+    assert len(registered_stages) == before_count
+
+
+def test_pipeline_import_is_stable_with_mocked_apscheduler() -> None:
+    """Regression: importing pipeline should not require cron/apscheduler modules."""
+    code = (
+        "import sys;"
+        "from unittest.mock import MagicMock;"
+        "mock_apscheduler = MagicMock();"
+        "mock_apscheduler.schedulers = MagicMock();"
+        "mock_apscheduler.schedulers.asyncio = MagicMock();"
+        "mock_apscheduler.schedulers.background = MagicMock();"
+        "sys.modules['apscheduler'] = mock_apscheduler;"
+        "sys.modules['apscheduler.schedulers'] = mock_apscheduler.schedulers;"
+        "sys.modules['apscheduler.schedulers.asyncio'] = mock_apscheduler.schedulers.asyncio;"
+        "sys.modules['apscheduler.schedulers.background'] = mock_apscheduler.schedulers.background;"
+        "import astrbot.core.pipeline as pipeline;"
+        "assert pipeline.ProcessStage is not None;"
+        "assert pipeline.RespondStage is not None"
+    )
+    _run_code_in_fresh_interpreter(
+        code,
+        "Pipeline import should not depend on real apscheduler package.",
+    )