wehub-resource-sync
104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
import threading
|
|
import tomllib
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from unittest.mock import patch
|
|
|
|
from app.config import config
|
|
from app.models.llm_provider import LLM_PROVIDER_REGISTRY
|
|
|
|
|
|
class TestConfigPersistence:
|
|
@staticmethod
|
|
def _load_example_config():
|
|
config_path = Path(__file__).resolve().parents[2] / "config.example.toml"
|
|
return tomllib.loads(config_path.read_text(encoding="utf-8"))
|
|
|
|
def test_example_config_documents_runtime_settings(self):
|
|
"""示例配置应展示用户需要手工维护的服务、素材和高级运行参数。"""
|
|
example_config = self._load_example_config()
|
|
app_config = example_config["app"]
|
|
|
|
assert example_config["listen_host"] == "0.0.0.0"
|
|
assert example_config["listen_port"] == 8080
|
|
assert example_config["log_level"] == "DEBUG"
|
|
assert app_config["video_source"] in {"pexels", "pixabay", "coverr", "local"}
|
|
assert "match_materials_to_script" in app_config
|
|
assert example_config["whisper"]["device"] == "cpu"
|
|
|
|
def test_example_config_covers_llm_provider_registry(self):
|
|
"""Registry 中可配置的 Provider 字段必须能在示例文件中被发现。"""
|
|
app_config = self._load_example_config()["app"]
|
|
|
|
for provider in LLM_PROVIDER_REGISTRY:
|
|
if provider.show_api_key:
|
|
assert provider.config_key("api_key") in app_config
|
|
if provider.show_base_url:
|
|
assert provider.config_key("base_url") in app_config
|
|
if provider.requires_model_name:
|
|
assert provider.config_key("model_name") in app_config
|
|
for field in provider.extra_fields:
|
|
assert provider.config_key(field.config_suffix) in app_config
|
|
|
|
def test_upload_post_settings_belong_to_app_section(self):
|
|
"""发布配置必须位于 app 节点，确保示例文件与运行时读取路径一致。"""
|
|
example_config = self._load_example_config()
|
|
upload_post_keys = {
|
|
"upload_post_enabled",
|
|
"upload_post_api_key",
|
|
"upload_post_username",
|
|
"upload_post_platforms",
|
|
"upload_post_auto_upload",
|
|
"upload_post_youtube_privacy_status",
|
|
}
|
|
|
|
assert upload_post_keys <= example_config["app"].keys()
|
|
assert upload_post_keys.isdisjoint(example_config.get("ui", {}).keys())
|
|
|
|
def test_save_config_uses_parseable_atomic_output(self):
|
|
"""
|
|
配置保存先写临时文件再原子替换。测试同时确认输出仍是合法 TOML，
|
|
且成功保存后不会在配置目录遗留临时文件。
|
|
"""
|
|
original_cfg = dict(config._cfg)
|
|
original_app = dict(config.app)
|
|
try:
|
|
with TemporaryDirectory() as temp_dir:
|
|
config_path = Path(temp_dir) / "config.toml"
|
|
config.app["atomic_save_test"] = "ok"
|
|
with (
|
|
patch.object(config, "root_dir", temp_dir),
|
|
patch.object(config, "config_file", str(config_path)),
|
|
):
|
|
config.save_config()
|
|
|
|
saved_config = tomllib.loads(config_path.read_text(encoding="utf-8"))
|
|
assert saved_config["app"]["atomic_save_test"] == "ok"
|
|
assert list(Path(temp_dir).glob(".config-*.toml.tmp")) == []
|
|
finally:
|
|
config.app.clear()
|
|
config.app.update(original_app)
|
|
config._cfg.clear()
|
|
config._cfg.update(original_cfg)
|
|
|
|
def test_runtime_config_lock_blocks_concurrent_config_writes(self):
|
|
"""长任务持有运行锁时，其它会话不能在任务中途改写全局配置。"""
|
|
write_started = threading.Event()
|
|
write_finished = threading.Event()
|
|
|
|
def update_config():
|
|
write_started.set()
|
|
config.app["runtime_lock_test"] = "updated"
|
|
write_finished.set()
|
|
|
|
config.app.pop("runtime_lock_test", None)
|
|
with config.runtime_config_lock():
|
|
worker = threading.Thread(target=update_config)
|
|
worker.start()
|
|
assert write_started.wait(timeout=1)
|
|
assert not write_finished.wait(timeout=0.05)
|
|
|
|
worker.join(timeout=1)
|
|
assert write_finished.is_set()
|
|
config.app.pop("runtime_lock_test", None)
|