fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-18 18:10:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add UMO config overrides

Browse Source
This commit is contained in:
Soulter
2026-07-05 14:47:58 +08:00
parent 25cbd41e08
commit 31e7af82d7
56 changed files with 2813 additions and 1447 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
tests/test_preprocess_stage.py
View File

@@ -28,6 +28,12 @@ class FakeEvent:
self.temporary_local_files.append(path)
def _make_preprocess_stage(config: dict) -> PreProcessStage:
stage = PreProcessStage()
stage.ctx = SimpleNamespace(astrbot_config=config)
return stage
@pytest.mark.asyncio
async def test_preprocess_preserves_image_formats_and_tracks_temp_files(
tmp_path, monkeypatch
@@ -77,10 +83,7 @@ async def test_preprocess_preserves_image_formats_and_tracks_temp_files(
),
]
)
stage = PreProcessStage()
stage.config = {}
stage.platform_settings = {}
stage.stt_settings = {"enable": False}
stage = _make_preprocess_stage({"provider_stt_settings": {"enable": False}})
await stage.process(event)
@@ -114,10 +117,12 @@ async def test_preprocess_path_mapping_accepts_file_uri(tmp_path):
target_image = target_root / "photo.jpg"
PILImage.new("RGB", (2, 2), (255, 0, 0)).save(target_image)
event = FakeEvent([Image(file="", url=source_image.as_uri())])
stage = PreProcessStage()
stage.config = {}
stage.platform_settings = {"path_mapping": [f"{source_root}:{target_root}"]}
stage.stt_settings = {"enable": False}
stage = _make_preprocess_stage(
{
"platform_settings": {"path_mapping": [f"{source_root}:{target_root}"]},
"provider_stt_settings": {"enable": False},
}
)
await stage.process(event)

43
tests/test_qqofficial_group_message_create.py
View File

@@ -1,5 +1,4 @@
import asyncio
import re
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import AsyncMock
@@ -287,8 +286,6 @@ async def test_webhook_group_send_by_session_without_cached_msg_id_omits_msg_id(
def test_qqofficial_ws_is_not_excluded_from_segmented_reply():
stage = RespondStage()
stage.enable_seg = True
stage.only_llm_result = False
result = MessageEventResult(chain=[Plain("hello")])
event = SimpleNamespace(
@@ -296,13 +293,11 @@ def test_qqofficial_ws_is_not_excluded_from_segmented_reply():
get_platform_name=lambda: "qq_official",
)
assert stage.is_seg_reply_required(cast(Any, event)) is True
assert stage.is_seg_reply_required(cast(Any, event), True, False) is True
def test_qqofficial_webhook_remains_excluded_from_segmented_reply():
stage = RespondStage()
stage.enable_seg = True
stage.only_llm_result = False
result = MessageEventResult(chain=[Plain("hello")])
event = SimpleNamespace(
@@ -310,26 +305,12 @@ def test_qqofficial_webhook_remains_excluded_from_segmented_reply():
get_platform_name=lambda: "qq_official_webhook",
)
assert stage.is_seg_reply_required(cast(Any, event)) is False
assert stage.is_seg_reply_required(cast(Any, event), True, False) is False
@pytest.mark.asyncio
async def test_result_decorate_segments_qqofficial_ws_plain_result():
stage = ResultDecorateStage()
stage.reply_prefix = ""
stage.content_safe_check_reply = False
stage.enable_segmented_reply = True
stage.only_llm_result = False
stage.words_count_threshold = 100
stage.split_mode = "words"
stage.split_words = ["。"]
stage.split_words_pattern = re.compile(r"(.*?(。)|.+$)", re.DOTALL)
stage.content_cleanup_rule = ""
stage.show_reasoning = False
stage.tts_trigger_probability = 0
stage.reply_with_mention = False
stage.reply_with_quote = False
stage.forward_threshold = 1000
setattr(
stage,
"ctx",
@@ -338,13 +319,33 @@ async def test_result_decorate_segments_qqofficial_ws_plain_result():
context=SimpleNamespace(get_using_tts_provider=lambda _umo: None)
),
astrbot_config={
"platform_settings": {
"reply_prefix": "",
"reply_with_mention": False,
"reply_with_quote": False,
"forward_threshold": 1000,
"segmented_reply": {
"enable": True,
"only_llm_result": False,
"words_count_threshold": 100,
"split_mode": "words",
"split_words": ["。"],
"content_cleanup_rule": "",
},
},
"provider_tts_settings": {
"enable": False,
"use_file_service": False,
"dual_output": False,
"trigger_probability": 0,
},
"provider_settings": {"display_reasoning_text": False},
"content_safety": {"also_use_in_response": False},
"callback_api_base": "",
"t2i": False,
"t2i_word_threshold": 150,
"t2i_strategy": "local",
"t2i_active_template": "",
},
),
)

1
tests/test_smoke.py
View File

@@ -77,7 +77,6 @@ def test_builtin_stage_bootstrap_is_idempotent() -> None:
expected_stage_names = {
"WakingCheckStage",
"WhitelistCheckStage",
"SessionStatusCheckStage",
"RateLimitStage",
"ContentSafetyCheckStage",
"PreProcessStage",

31
tests/unit/test_kb_manager_resilience.py
View File

@@ -264,6 +264,37 @@ async def test_ensure_vec_db_clears_stale_init_error(
assert helper.vec_db is mock_vec_db
@pytest.mark.asyncio
async def test_retrieve_accepts_kb_id_in_kb_names(
stub_provider_manager_module,
mock_provider_manager,
mock_knowledge_base,
):
"""Test that retrieve accepts KB IDs through the legacy kb_names field."""
from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager
kb_helper = MagicMock()
kb_helper.kb = mock_knowledge_base
kb_helper.init_error = None
kb_mgr = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
kb_mgr.provider_manager = mock_provider_manager
kb_mgr.kb_insts = {mock_knowledge_base.kb_id: kb_helper}
kb_mgr.retrieval_manager = MagicMock()
kb_mgr.retrieval_manager.retrieve = AsyncMock(return_value=[])
result = await kb_mgr.retrieve(
query="hello",
kb_names=[mock_knowledge_base.kb_id],
)
assert result is None
kb_mgr.retrieval_manager.retrieve.assert_awaited_once()
assert kb_mgr.retrieval_manager.retrieve.await_args.kwargs["kb_ids"] == [
mock_knowledge_base.kb_id
]
@pytest.mark.asyncio
async def test_ensure_vec_db_sets_init_error_on_failure(
stub_provider_manager_module,

12
tests/unit/test_message_tools.py
View File

@@ -96,9 +96,15 @@ class _DummyRespondEvent:
def _make_respond_stage() -> RespondStage:
"""Build a minimally initialized RespondStage for unit tests."""
stage = RespondStage()
stage.config = {"provider_settings": {}}
stage.platform_settings = {"path_mapping": []}
stage.enable_seg = False
stage.ctx = SimpleNamespace(
astrbot_config={
"provider_settings": {},
"platform_settings": {
"path_mapping": [],
"segmented_reply": {"enable": False},
},
}
)
return stage