config 配置模块: 默认值兜底+必填校验+各模块分发工厂 (53 passed) - 客户端 7 模块齐

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())

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()