From adbfd7b4bb9ff9f422b20bae1d68f3fd92099f65 Mon Sep 17 00:00:00 2001 From: lou Date: Sun, 9 Aug 2026 23:04:19 +0800 Subject: [PATCH] =?UTF-8?q?config=20=E9=85=8D=E7=BD=AE=E6=A8=A1=E5=9D=97:?= =?UTF-8?q?=20=E9=BB=98=E8=AE=A4=E5=80=BC=E5=85=9C=E5=BA=95+=E5=BF=85?= =?UTF-8?q?=E5=A1=AB=E6=A0=A1=E9=AA=8C+=E5=90=84=E6=A8=A1=E5=9D=97?= =?UTF-8?q?=E5=88=86=E5=8F=91=E5=B7=A5=E5=8E=82=20(53=20passed)=20-=20?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=207=20=E6=A8=A1=E5=9D=97=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.py | 187 +++++++++++++++++++++++++++++++++++++++++++ tests/test_config.py | 123 ++++++++++++++++++++++++++++ 2 files changed, 310 insertions(+) create mode 100644 config.py create mode 100644 tests/test_config.py diff --git a/config.py b/config.py new file mode 100644 index 0000000..653f442 --- /dev/null +++ b/config.py @@ -0,0 +1,187 @@ +""" +配置模块 (config) +模块: 客户端 / 配置 +输入: config.json +输出: 各模块配置注入 (crypto 拿密钥, transfer 拿地址) + +配置外置, 源码不硬编码; 必填项缺失直接报错, 不带着坏配置跑。 +默认配置文件: config/config.json (可 --init 生成模板) +""" + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +DEFAULT_CONFIG = Path(__file__).resolve().parent / "config" / "config.json" + +# 默认值 (用户 config.json 逐层覆盖) +DEFAULTS: dict[str, Any] = { + "server": {"url": "", "timeout": 30, "max_retries": 3, "retry_delay": 1.0}, + "crypto": {"key_id": "default_key", "keyring_path": "config/keyring.json"}, + "splitter": {"chunk_size_mb": 10}, + "state": {"db_path": "config/tasks.db", "cleanup_days": 7}, +} + +# 必填路径 (缺失即报错) +REQUIRED: list[tuple[str, str]] = [("server", "url")] + + +class ConfigError(ValueError): + """配置缺失或非法""" + + +class AppConfig: + """加载/校验/分发配置 (默认值兜底 + 用户覆盖)""" + + def __init__(self, config_path: str | Path | None = None) -> None: + self.config_path = Path(config_path) if config_path else DEFAULT_CONFIG + self.data: dict[str, Any] = {} + self.load() + + def load(self) -> None: + """读取 config.json, 与默认值合并, 校验必填项""" + self.data = json.loads(json.dumps(DEFAULTS)) # 深拷贝默认值 + if self.config_path.exists(): + user = json.loads(self.config_path.read_text(encoding="utf-8")) + self._merge(self.data, user) + self.validate() + + @staticmethod + def _merge(base: dict[str, Any], override: dict[str, Any]) -> None: + for key, val in override.items(): + if key in base and isinstance(base[key], dict) and isinstance(val, dict): + AppConfig._merge(base[key], val) + else: + base[key] = val + + def validate(self) -> None: + """必填校验 (服务器/密钥/卷大小), 任一非法抛 ConfigError""" + for section, key in REQUIRED: + v = self.data.get(section, {}).get(key) + if not isinstance(v, str) or not v: + raise ConfigError( + f"必填配置缺失: {section}.{key} (用 --init 生成 config.json 模板)" + ) + url: str = self.data["server"]["url"] + if not url.startswith(("http://", "https://")): + raise ConfigError(f"server.url 必须以 http:// 或 https:// 开头: {url}") + if self.data["splitter"]["chunk_size_mb"] <= 0: + raise ConfigError("splitter.chunk_size_mb 必须大于 0") + if self.data["server"]["max_retries"] < 1: + raise ConfigError("server.max_retries 必须 >= 1") + if self.data["server"]["timeout"] <= 0: + raise ConfigError("server.timeout 必须大于 0") + + # ---------- 便捷访问 ---------- + + @property + def server_url(self) -> str: + return self.data["server"]["url"] + + @property + def timeout(self) -> int: + return self.data["server"]["timeout"] + + @property + def max_retries(self) -> int: + return self.data["server"]["max_retries"] + + @property + def retry_delay(self) -> float: + return self.data["server"]["retry_delay"] + + @property + def key_id(self) -> str: + return self.data["crypto"]["key_id"] + + @property + def keyring_path(self) -> Path: + return Path(self.data["crypto"]["keyring_path"]) + + @property + def chunk_size(self) -> int: + """卷大小 (字节)""" + return self.data["splitter"]["chunk_size_mb"] * 1024 * 1024 + + @property + def db_path(self) -> Path: + return Path(self.data["state"]["db_path"]) + + @property + def cleanup_days(self) -> int: + return self.data["state"]["cleanup_days"] + + # ---------- 分发: 构造各模块实例 ---------- + + def build_crypto(self): + """crypto 拿密钥 (keyring 路径)""" + from crypto import CryptoEngine + return CryptoEngine(self.keyring_path) + + def build_transfer(self): + """transfer 拿服务器地址/重试参数""" + from transfer import TransferClient + return TransferClient( + self.server_url, + timeout=self.timeout, + max_retries=self.max_retries, + retry_delay=self.retry_delay, + ) + + def build_store(self): + """state 拿数据库路径""" + from state import TaskStore + return TaskStore(self.db_path) + + +def main() -> int: + ap = argparse.ArgumentParser(description="配置: 加载校验 / 生成模板") + ap.add_argument("--init", action="store_true", help="生成默认 config.json 模板") + ap.add_argument("--path", default=str(DEFAULT_CONFIG), help="配置文件路径") + args = ap.parse_args() + + if args.init: + path = Path(args.path) + if path.exists(): + print(f"[错误] 已存在: {path}, 不覆盖") + return 1 + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "server": { + "url": "http://127.0.0.1:8000", + "timeout": 30, + "max_retries": 3, + "retry_delay": 1.0, + }, + "crypto": {"key_id": "default_key", "keyring_path": "config/keyring.json"}, + "splitter": {"chunk_size_mb": 10}, + "state": {"db_path": "config/tasks.db", "cleanup_days": 7}, + }, + indent=2, + ensure_ascii=False, + ) + + "\n", + encoding="utf-8", + ) + print(f"[config] 模板已生成: {path} (改 server.url 后即可用)") + return 0 + + try: + cfg = AppConfig(args.path) + except (ConfigError, json.JSONDecodeError) as e: + print(f"[错误] 配置无效: {e}") + return 1 + print(f"[config] 已加载: {cfg.config_path}") + print(f" server: {cfg.server_url} (timeout={cfg.timeout}, retries={cfg.max_retries})") + print(f" crypto: key_id={cfg.key_id}, keyring={cfg.keyring_path}") + print(f" splitter: 卷大小 {cfg.data['splitter']['chunk_size_mb']} MB") + print(f" state: db={cfg.db_path}, 清理保留 {cfg.cleanup_days} 天") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..d5ec143 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,123 @@ +"""config 模块行为测试 (真实 JSON 文件读写, 非 mock) + +运行: cd /home/lou/文档/7z-encrypt && .venv/bin/python -m unittest discover -s tests -v +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from config import AppConfig, ConfigError + +VALID = { + "server": {"url": "http://127.0.0.1:8000", "timeout": 30, "max_retries": 3}, + "crypto": {"key_id": "default_key", "keyring_path": "config/keyring.json"}, + "splitter": {"chunk_size_mb": 10}, + "state": {"db_path": "config/tasks.db", "cleanup_days": 7}, +} + + +def write_cfg(td: str, data: dict) -> str: + path = os.path.join(td, "config.json") + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f) + return path + + +class TestLoad(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory(prefix='hermes-test-') + self.addCleanup(self._td.cleanup) + + def test_valid_config(self): + cfg = AppConfig(write_cfg(self._td.name, VALID)) + self.assertEqual(cfg.server_url, "http://127.0.0.1:8000") + self.assertEqual(cfg.key_id, "default_key") + self.assertEqual(cfg.chunk_size, 10 * 1024 * 1024) + self.assertEqual(cfg.cleanup_days, 7) + + def test_defaults_filled(self): + cfg = AppConfig(write_cfg(self._td.name, {"server": {"url": "http://x:1"}})) + self.assertEqual(cfg.timeout, 30) # 默认值兜底 + self.assertEqual(cfg.max_retries, 3) + self.assertEqual(cfg.retry_delay, 1.0) + self.assertEqual(cfg.data["splitter"]["chunk_size_mb"], 10) + self.assertEqual(cfg.key_id, "default_key") + + def test_user_overrides_default(self): + data = dict(VALID) + data["server"] = {"url": "http://x:1", "timeout": 5} + cfg = AppConfig(write_cfg(self._td.name, data)) + self.assertEqual(cfg.timeout, 5) # 用户覆盖 + self.assertEqual(cfg.max_retries, 3) # 未覆盖的保持默认 + + def test_missing_file(self): + with self.assertRaises(ConfigError): + AppConfig(os.path.join(self._td.name, "nope.json")) + + def test_invalid_json(self): + path = os.path.join(self._td.name, "config.json") + with open(path, "w", encoding="utf-8") as f: + f.write("{not json") + with self.assertRaises(json.JSONDecodeError): + AppConfig(path) + + +class TestValidate(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory(prefix='hermes-test-') + self.addCleanup(self._td.cleanup) + + def _cfg(self, **mut): + data = json.loads(json.dumps(VALID)) + for k, v in mut.items(): + data[k] = v + return AppConfig(write_cfg(self._td.name, data)) + + def test_missing_server_url(self): + with self.assertRaises(ConfigError): + self._cfg(server={}) + + def test_bad_url_scheme(self): + with self.assertRaises(ConfigError): + self._cfg(server={"url": "ftp://x"}) + + def test_bad_chunk_size(self): + with self.assertRaises(ConfigError): + self._cfg(splitter={"chunk_size_mb": 0}) + + def test_bad_max_retries(self): + with self.assertRaises(ConfigError): + self._cfg(server={"url": "http://x:1", "max_retries": 0}) + + def test_bad_timeout(self): + with self.assertRaises(ConfigError): + self._cfg(server={"url": "http://x:1", "timeout": -1}) + + +class TestFactory(unittest.TestCase): + """分发: 构造各模块实例 (真实构造, 非 mock)""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory(prefix='hermes-test-') + self.addCleanup(self._td.cleanup) + + def test_build_instances(self): + cfg = AppConfig(write_cfg(self._td.name, VALID)) + from crypto import CryptoEngine + from transfer import TransferClient + from state import TaskStore + + self.assertIsInstance(cfg.build_crypto(), CryptoEngine) + t = cfg.build_transfer() + self.assertIsInstance(t, TransferClient) + self.assertEqual(t.base_url, VALID["server"]["url"]) + self.assertEqual(t.max_retries, 3) + self.assertIsInstance(cfg.build_store(), TaskStore) + + +if __name__ == '__main__': + unittest.main()