zhaoxi/KiloStar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(config): 统一配置加载入口，启动时校验所有YAML配置

Browse Source 
将分散的 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>
This commit is contained in:
zhaoxi
2026-06-03 13:52:03 +00:00
parent 80174acaae
commit 76a67e8237
8 changed files with 358 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_api_onebot.py
+1
View File
@@ -16,6 +16,7 @@ from kilostar.core.individual.regulatory_node.template import MessageResponse
def test_verify_token_skipped_when_env_missing(monkeypatch):
monkeypatch.delenv("ONEBOT_ACCESS_TOKEN", raising=False)
monkeypatch.setenv("KILOSTAR_ENV", "dev")
onebot_mod._verify_token(None)
onebot_mod._verify_token("anything")
tests/unit/test_config_loader.py
+7 -5
View File
@@ -8,13 +8,15 @@ import pytest
@pytest.fixture(autouse=True)
def isolated_yaml(tmp_path, monkeypatch):
"""每个用例用独立的临时 yaml，避免污染真实 config/workflow.yaml。"""
"""每个用例用独立的临时目录作为 config 目录，避免污染真实配置文件。"""
from kilostar.utils import config_loader
fake_yaml = tmp_path / "workflow.yaml"
monkeypatch.setattr(config_loader, "_WORKFLOW_YAML", fake_yaml)
monkeypatch.setattr(config_loader, "_current", None)
return fake_yaml
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():
tests/unit/test_sandbox.py
+46 -35
View File
@@ -4,11 +4,16 @@ import os
import pytest
from unittest.mock import patch
from kilostar.utils.sandbox import (
from kilostar.utils.config_loader import (
SandboxConfig,
FilesystemPolicy,
ShellPolicy,
PythonExecutorPolicy,
AppConfig,
AppInfo,
WorkflowConfig,
)
from kilostar.utils.sandbox import (
validate_path,
validate_shell_command,
validate_python_code,
@@ -24,47 +29,49 @@ from kilostar.utils.sandbox import (
@pytest.fixture(autouse=True)
def reset_sandbox_config():
"""每个测试前重置沙箱配置缓存。"""
import kilostar.utils.sandbox as mod
mod._current = None
"""每个测试前重置配置缓存。"""
import kilostar.utils.config_loader as loader
loader._app_current = None
yield
mod._current = None
loader._app_current = None
@pytest.fixture
def mock_config():
"""注入测试用的沙箱配置。"""
import kilostar.utils.sandbox as mod
cfg = SandboxConfig(
enabled=True,
filesystem=FilesystemPolicy(
workspace_root="/tmp/kilostar_workspace",
allowed_read_paths=["/tmp"],
denied_paths=["/etc/shadow", "/root"],
),
shell=ShellPolicy(
import kilostar.utils.config_loader as loader
cfg = AppConfig(
sandbox=SandboxConfig(
enabled=True,
blocked_commands=["rm -rf /", "mkfs", "shutdown"],
blocked_operators=["&&", "||", ";", "`", "$("],
max_timeout=60,
),
python_executor=PythonExecutorPolicy(
enabled=True,
max_timeout=30,
blocked_imports=["os", "subprocess", "shutil"],
blocked_builtins=["exec", "eval", "__import__"],
filesystem=FilesystemPolicy(
workspace_root="/tmp/kilostar_workspace",
allowed_read_paths=["/tmp"],
denied_paths=["/etc/shadow", "/root"],
),
shell=ShellPolicy(
enabled=True,
blocked_commands=["rm -rf /", "mkfs", "shutdown"],
blocked_operators=["&&", "||", ";", "`", "$("],
max_timeout=60,
),
python_executor=PythonExecutorPolicy(
enabled=True,
max_timeout=30,
blocked_imports=["os", "subprocess", "shutil"],
blocked_builtins=["exec", "eval", "__import__"],
),
),
)
mod._current = cfg
loader._app_current = cfg
return cfg
@pytest.fixture
def disabled_config():
"""沙箱关闭时的配置。"""
import kilostar.utils.sandbox as mod
cfg = SandboxConfig(enabled=False)
mod._current = cfg
import kilostar.utils.config_loader as loader
cfg = AppConfig(sandbox=SandboxConfig(enabled=False))
loader._app_current = cfg
return cfg
@@ -140,10 +147,12 @@ class TestValidateShellCommand:
assert validate_shell_command("rm -rf /") == "rm -rf /"
def test_shell_disabled_in_policy(self):
import kilostar.utils.sandbox as mod
mod._current = SandboxConfig(
enabled=True,
shell=ShellPolicy(enabled=False),
import kilostar.utils.config_loader as loader
loader._app_current = AppConfig(
sandbox=SandboxConfig(
enabled=True,
shell=ShellPolicy(enabled=False),
),
)
with pytest.raises(CommandViolation, match="已被沙箱策略禁用"):
validate_shell_command("ls")
@@ -190,10 +199,12 @@ class TestValidatePythonCode:
assert validate_python_code("import os") == "import os"
def test_python_disabled_in_policy(self):
import kilostar.utils.sandbox as mod
mod._current = SandboxConfig(
enabled=True,
python_executor=PythonExecutorPolicy(enabled=False),
import kilostar.utils.config_loader as loader
loader._app_current = AppConfig(
sandbox=SandboxConfig(
enabled=True,
python_executor=PythonExecutorPolicy(enabled=False),
),
)
with pytest.raises(CodeViolation, match="已被沙箱策略禁用"):
validate_python_code("print(1)")
tests/unit/test_unified_config.py
+168
View File
@@ -0,0 +1,168 @@
"""统一配置加载 (AppConfig) 的测试：多文件联合加载、缺失文件默认值、schema 校验、热重载。"""
from __future__ import annotations
import yaml
import pytest
from kilostar.utils.config_loader import (
AppConfig,
AppInfo,
WorkflowConfig,
RetryConfig,
SandboxConfig,
load_app_config,
get_app_config,
reload_app_config,
get_workflow_config,
reload_workflow_config,
save_workflow_config,
)
@pytest.fixture(autouse=True)
def isolated_config(tmp_path, monkeypatch):
"""隔离配置目录，每个测试用独立临时目录。"""
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
# ─── 多文件联合加载 ───
class TestLoadAppConfig:
def test_all_files_present(self, isolated_config):
(isolated_config / "config.yml").write_text(
"version: v0.2.0\nname: TestStar\n", encoding="utf-8"
)
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 10\n", encoding="utf-8"
)
(isolated_config / "sandbox.yaml").write_text(
"sandbox:\n enabled: false\n", encoding="utf-8"
)
cfg = load_app_config(isolated_config)
assert cfg.app.version == "v0.2.0"
assert cfg.app.name == "TestStar"
assert cfg.workflow.retry.max_attempts == 10
assert cfg.sandbox.enabled is False
def test_no_files_returns_defaults(self, isolated_config):
cfg = load_app_config(isolated_config)
assert cfg.app.version == "0.1.1-alpha"
assert cfg.app.name == "Kilostar"
assert cfg.workflow.retry.max_attempts == 5
assert cfg.sandbox.enabled is True
def test_partial_files_fills_defaults(self, isolated_config):
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 8\n", encoding="utf-8"
)
cfg = load_app_config(isolated_config)
assert cfg.workflow.retry.max_attempts == 8
assert cfg.app.version == "0.1.1-alpha"
assert cfg.sandbox.enabled is True
def test_sandbox_without_wrapper_key(self, isolated_config):
"""sandbox.yaml 不带顶层 'sandbox:' key 也能正常解析。"""
(isolated_config / "sandbox.yaml").write_text(
"enabled: false\n", encoding="utf-8"
)
cfg = load_app_config(isolated_config)
assert cfg.sandbox.enabled is False
# ─── Schema 校验 ───
class TestSchemaValidation:
def test_invalid_retry_max_attempts_type(self, isolated_config):
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: not_a_number\n", encoding="utf-8"
)
with pytest.raises(Exception):
load_app_config(isolated_config)
def test_retry_max_attempts_out_of_range(self, isolated_config):
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 999\n", encoding="utf-8"
)
with pytest.raises(Exception):
load_app_config(isolated_config)
def test_config_yml_extra_fields_ignored(self, isolated_config):
(isolated_config / "config.yml").write_text(
"version: v1.0.0\nname: X\nunknown_field: hello\n", encoding="utf-8"
)
cfg = load_app_config(isolated_config)
assert cfg.app.version == "v1.0.0"
# ─── 单例与热重载 ───
class TestSingletonAndReload:
def test_get_app_config_is_singleton(self, isolated_config):
cfg1 = get_app_config()
cfg2 = get_app_config()
assert cfg1 is cfg2
def test_reload_full(self, isolated_config):
get_app_config()
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 42\n", encoding="utf-8"
)
cfg = reload_app_config()
assert cfg.workflow.retry.max_attempts == 42
def test_reload_section_sandbox(self, isolated_config):
get_app_config()
(isolated_config / "sandbox.yaml").write_text(
"sandbox:\n enabled: false\n", encoding="utf-8"
)
cfg = reload_app_config(section="sandbox")
assert cfg.sandbox.enabled is False
assert cfg.workflow.retry.max_attempts == 5
def test_reload_section_workflow(self, isolated_config):
get_app_config()
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 77\n", encoding="utf-8"
)
cfg = reload_app_config(section="workflow")
assert cfg.workflow.retry.max_attempts == 77
# ─── 向后兼容性 ───
class TestBackwardCompat:
def test_get_workflow_config_delegates(self, isolated_config):
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 15\n", encoding="utf-8"
)
wf = get_workflow_config()
assert wf.retry.max_attempts == 15
def test_reload_workflow_config_refreshes(self, isolated_config):
get_workflow_config()
(isolated_config / "workflow.yaml").write_text(
"retry:\n max_attempts: 33\n", encoding="utf-8"
)
wf = reload_workflow_config()
assert wf.retry.max_attempts == 33
def test_save_workflow_config_persists(self, isolated_config):
save_workflow_config(WorkflowConfig(retry=RetryConfig(max_attempts=25)))
on_disk = yaml.safe_load(
(isolated_config / "workflow.yaml").read_text(encoding="utf-8")
)
assert on_disk == {"retry": {"max_attempts": 25}}
assert get_workflow_config().retry.max_attempts == 25
tests/unit/test_workflow_graph.py
+3 -2
View File
@@ -295,9 +295,10 @@ def test_workflow_graph_state_defaults():
async def test_loop_retry_exceeds_max_attempts_fails(monkeypatch):
"""环跳转超过 max_attempts 后，工作流应直接 FAILED 而非无限重试。"""
from kilostar.utils import config_loader
from kilostar.utils.config_loader import WorkflowConfig, RetryConfig
monkeypatch.setattr(config_loader, "_app_current", None)
from kilostar.utils.config_loader import WorkflowConfig, RetryConfig, AppConfig
monkeypatch.setattr(config_loader, "_current", WorkflowConfig(retry=RetryConfig(max_attempts=2)))
monkeypatch.setattr(config_loader, "_app_current", AppConfig(workflow=WorkflowConfig(retry=RetryConfig(max_attempts=2))))
deps, sink = _make_deps(
skill_outputs=[