File size: 4,439 Bytes
8ede856 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """Smoke tests for critical startup and import paths."""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
from astrbot.core.pipeline.bootstrap import ensure_builtin_stages_registered
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:
code = (
"import importlib;"
"mods=["
"'astrbot.core.core_lifecycle',"
"'astrbot.core.astr_main_agent',"
"'astrbot.core.pipeline.scheduler',"
"'astrbot.core.pipeline.process_stage.method.agent_sub_stages.internal',"
"'astrbot.core.pipeline.process_stage.method.agent_sub_stages.third_party'"
"];"
"[importlib.import_module(m) for m in mods]"
)
_run_code_in_fresh_interpreter(code, "Smoke import check failed.")
def test_smoke_pipeline_stage_registration_matches_order() -> None:
ensure_builtin_stages_registered()
stage_names = {cls.__name__ for cls in registered_stages}
assert set(STAGES_ORDER).issubset(stage_names)
assert len(stage_names) == len(registered_stages)
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();"
"mock_apscheduler.triggers = MagicMock();"
"mock_apscheduler.triggers.cron = MagicMock();"
"mock_apscheduler.triggers.date = 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;"
"sys.modules['apscheduler.triggers'] = mock_apscheduler.triggers;"
"sys.modules['apscheduler.triggers.cron'] = mock_apscheduler.triggers.cron;"
"sys.modules['apscheduler.triggers.date'] = mock_apscheduler.triggers.date;"
"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.",
)
|