"""``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