zhaoxi
76a67e8237
将分散的 config.yml、workflow.yaml、sandbox.yaml 加载逻辑统一到 AppConfig 模型， 启动时一次性校验，失败则 fast-fail。sandbox.py 改为从统一配置取值，消除重复加载。 同时修复 onebot 测试并新增14个统一配置测试（总测试 285→300）。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""``utils/config_loader.py``：workflow.yaml 读/写/热重载。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import yaml
|
||
import pytest
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def isolated_yaml(tmp_path, monkeypatch):
|
||
"""每个用例用独立的临时目录作为 config 目录，避免污染真实配置文件。"""
|
||
from kilostar.utils import config_loader
|
||
|
||
monkeypatch.setattr(config_loader, "_CONFIG_DIR", tmp_path)
|
||
monkeypatch.setattr(config_loader, "_WORKFLOW_YAML", tmp_path / "workflow.yaml")
|
||
monkeypatch.setattr(config_loader, "_CONFIG_YML", tmp_path / "config.yml")
|
||
monkeypatch.setattr(config_loader, "_SANDBOX_YAML", tmp_path / "sandbox.yaml")
|
||
monkeypatch.setattr(config_loader, "_app_current", None)
|
||
return tmp_path / "workflow.yaml"
|
||
|
||
|
||
def test_get_workflow_config_returns_default_when_file_absent():
|
||
from kilostar.utils.config_loader import get_workflow_config
|
||
|
||
config = get_workflow_config()
|
||
assert config.retry.max_attempts == 5
|
||
|
||
|
||
def test_get_workflow_config_reads_from_disk(isolated_yaml):
|
||
isolated_yaml.write_text("retry:\n max_attempts: 12\n", encoding="utf-8")
|
||
|
||
from kilostar.utils.config_loader import reload_workflow_config
|
||
|
||
config = reload_workflow_config()
|
||
assert config.retry.max_attempts == 12
|
||
|
||
|
||
def test_save_workflow_config_writes_yaml_and_reloads(isolated_yaml):
|
||
from kilostar.utils.config_loader import (
|
||
save_workflow_config,
|
||
get_workflow_config,
|
||
WorkflowConfig,
|
||
RetryConfig,
|
||
)
|
||
|
||
new_config = WorkflowConfig(retry=RetryConfig(max_attempts=20))
|
||
save_workflow_config(new_config)
|
||
|
||
on_disk = yaml.safe_load(isolated_yaml.read_text(encoding="utf-8"))
|
||
assert on_disk == {"retry": {"max_attempts": 20}}
|
||
|
||
# 热重载：再次 get 应直接拿到新值
|
||
assert get_workflow_config().retry.max_attempts == 20
|
||
|
||
|
||
def test_max_attempts_validation_rejects_out_of_range():
|
||
from kilostar.utils.config_loader import RetryConfig
|
||
|
||
with pytest.raises(Exception):
|
||
RetryConfig(max_attempts=0)
|
||
|
||
with pytest.raises(Exception):
|
||
RetryConfig(max_attempts=200)
|
||
|
||
|
||
def test_reload_picks_up_external_file_changes(isolated_yaml):
|
||
"""模拟运维直接改 yaml 文件，reload 后引擎能拿到新值。"""
|
||
isolated_yaml.write_text("retry:\n max_attempts: 3\n", encoding="utf-8")
|
||
|
||
from kilostar.utils.config_loader import (
|
||
get_workflow_config,
|
||
reload_workflow_config,
|
||
)
|
||
|
||
assert reload_workflow_config().retry.max_attempts == 3
|
||
|
||
isolated_yaml.write_text("retry:\n max_attempts: 30\n", encoding="utf-8")
|
||
assert reload_workflow_config().retry.max_attempts == 30
|
||
# get_workflow_config 也读到最新
|
||
assert get_workflow_config().retry.max_attempts == 30
|