mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-17 01:49:15 +08:00
63cbab610a
Allow the dashboard to become available before plugin bootstrap completes and surface runtime readiness and failure states to API callers. Guard plugin-facing endpoints until runtime is ready and clean up provider and plugin runtime state safely across bootstrap failures, retries, stop, and restart flows.
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Tests for InitialLoader."""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from astrbot.core.initial_loader import InitialLoader
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_loader_start_awaits_initialize_core_and_schedules_runtime_bootstrap():
|
|
"""Test InitialLoader.start splits core init from background runtime bootstrap."""
|
|
loader = InitialLoader(MagicMock(), MagicMock())
|
|
call_order: list[str] = []
|
|
real_create_task = asyncio.create_task
|
|
created_tasks: list[asyncio.Task] = []
|
|
|
|
lifecycle = MagicMock()
|
|
lifecycle.dashboard_shutdown_event = asyncio.Event()
|
|
lifecycle.runtime_bootstrap_task = None
|
|
|
|
async def initialize_core() -> None:
|
|
call_order.append("initialize_core")
|
|
|
|
async def bootstrap_runtime() -> None:
|
|
call_order.append("bootstrap_runtime")
|
|
|
|
async def start_core() -> None:
|
|
call_order.append("core_start")
|
|
|
|
async def run_dashboard() -> None:
|
|
call_order.append("dashboard_run")
|
|
|
|
lifecycle.initialize = AsyncMock(
|
|
side_effect=AssertionError("initialize should not be used")
|
|
)
|
|
lifecycle.initialize_core = AsyncMock(side_effect=initialize_core)
|
|
lifecycle.bootstrap_runtime = AsyncMock(side_effect=bootstrap_runtime)
|
|
lifecycle.start = AsyncMock(side_effect=start_core)
|
|
|
|
dashboard = MagicMock()
|
|
dashboard.run = AsyncMock(side_effect=run_dashboard)
|
|
|
|
def dashboard_factory(*args, **kwargs):
|
|
del args, kwargs
|
|
call_order.append("dashboard_init")
|
|
return dashboard
|
|
|
|
def create_task(coro, *args, **kwargs):
|
|
call_order.append("create_task")
|
|
task = real_create_task(coro, *args, **kwargs)
|
|
created_tasks.append(task)
|
|
return task
|
|
|
|
with (
|
|
patch(
|
|
"astrbot.core.initial_loader.AstrBotCoreLifecycle", return_value=lifecycle
|
|
),
|
|
patch(
|
|
"astrbot.core.initial_loader.AstrBotDashboard",
|
|
side_effect=dashboard_factory,
|
|
),
|
|
patch(
|
|
"astrbot.core.initial_loader.asyncio.create_task", side_effect=create_task
|
|
),
|
|
):
|
|
await loader.start()
|
|
|
|
lifecycle.initialize.assert_not_called()
|
|
lifecycle.initialize_core.assert_awaited_once()
|
|
lifecycle.bootstrap_runtime.assert_awaited_once()
|
|
lifecycle.start.assert_awaited_once()
|
|
dashboard.run.assert_awaited_once()
|
|
assert call_order[:3] == ["initialize_core", "create_task", "dashboard_init"]
|
|
assert len(created_tasks) == 1
|
|
assert lifecycle.runtime_bootstrap_task is created_tasks[0]
|