refactor(mytoolkit): 移除历史兼容，统一分层配置键，补全 Config 单测
- config.Config 仅支持 ~/.mytoolkit/config.json 分层键 - 删除旧 env.json/module-config.json 迁移逻辑与 flat-key 别名 - ssh/bib/image 命令改用 connections.* / paths.* / secrets.api_keys.* - 单测覆盖分层读写、嵌套删除、export、resolve_key、权限
This commit is contained in:
Zhengshou Lai
@@ -41,7 +41,7 @@ def to_markdown(files, dry_run):
|
||||
@handle_errors
|
||||
def cv_update():
|
||||
"""Update CV bibliography using biber."""
|
||||
cv_path_str = config.get("path_cv")
|
||||
cv_path_str = config.get("paths.cv")
|
||||
if cv_path_str:
|
||||
cv_path = Path(cv_path_str).expanduser()
|
||||
else:
|
||||
|
||||
@@ -201,9 +201,9 @@ def generate_image_cmd(prompt, file, size, ratio, output, no_watermark, b64):
|
||||
elif not prompt:
|
||||
raise click.UsageError("必须提供 prompt 或使用 -f/--file 从文件读取")
|
||||
|
||||
api_key = config.get("apikey_ark")
|
||||
api_key = config.get("secrets.api_keys.ark")
|
||||
if not api_key:
|
||||
click.echo("Error: apikey_ark not set. Run: mytoolkit env set apikey_ark <value>", err=True)
|
||||
click.echo("Error: secrets.api_keys.ark not set. Run: mytoolkit env set secrets.api_keys.ark <value>", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
if size and ratio:
|
||||
|
||||
+13
-12
@@ -11,18 +11,19 @@ from mytoolkit.utils import handle_errors
|
||||
|
||||
def _get_hosts() -> dict:
|
||||
"""Get SSH hosts from config."""
|
||||
hosts = {}
|
||||
for key in config.get_all().keys():
|
||||
if key.startswith("ssh_") and not key.endswith("_port") and not key.endswith("_key"):
|
||||
name = key[4:] # Remove 'ssh_' prefix
|
||||
host = config.get(f"ssh_{name}")
|
||||
port = config.get(f"ssh_{name}_port") or "22"
|
||||
key_file = config.get(f"ssh_{name}_key")
|
||||
if host:
|
||||
hosts[name] = {"host": host, "port": int(port)}
|
||||
if key_file:
|
||||
hosts[name]["key"] = key_file
|
||||
return hosts
|
||||
hosts = config.get("connections.ssh") or {}
|
||||
result = {}
|
||||
for name, cfg in hosts.items():
|
||||
if not isinstance(cfg, dict):
|
||||
continue
|
||||
host = cfg.get("host")
|
||||
if not host:
|
||||
continue
|
||||
result[name] = {"host": host, "port": int(cfg.get("port", 22))}
|
||||
key_file = cfg.get("key")
|
||||
if key_file:
|
||||
result[name]["key"] = key_file
|
||||
return result
|
||||
|
||||
|
||||
def _build_ssh_base_cmd(host_config: dict) -> list[str]:
|
||||
|
||||
+63
-321
@@ -1,8 +1,7 @@
|
||||
"""Configuration management for mytoolkit.
|
||||
|
||||
Unified config file: ~/.mytoolkit/config.json
|
||||
Schema version: 2 (hierarchical)
|
||||
Legacy: ~/.mytoolkit/env.json and flat keys.* format (auto-migrated)
|
||||
Storage: ~/.mytoolkit/config.json
|
||||
Schema: hierarchical dict, e.g. ``secrets.api_keys.ark``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -12,85 +11,20 @@ import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_MYTOOLKIT_HOME = Path(os.environ.get("MYTOOLKIT_HOME", Path.home() / ".mytoolkit"))
|
||||
CONFIG_PATH = _MYTOOLKIT_HOME / "config.json"
|
||||
_LEGACY_ENV_PATH = _MYTOOLKIT_HOME / "env.json"
|
||||
_LEGACY_MODULE_PATH = Path(__file__).parent / "config.json"
|
||||
_XIAOHE_CONFIG_PATH = Path.home() / ".xiaohe" / "agent" / "config.json"
|
||||
_XIAOHE_SETTINGS_PATH = Path.home() / ".xiaohe" / "agent" / "settings.json"
|
||||
_MAIL_CONFIG_PATH = _MYTOOLKIT_HOME / "mail.json"
|
||||
_METABOT_BOTS_PATH = Path.home() / ".metabot" / "bots.json"
|
||||
|
||||
# Mapping from legacy flat keys to hierarchical paths.
|
||||
_LEGACY_KEY_MAP: dict[str, str] = {
|
||||
# API keys
|
||||
"apikey_ark": "secrets.api_keys.ark",
|
||||
"apikey_deepseek": "secrets.api_keys.deepseek",
|
||||
"apikey_elsevier": "secrets.api_keys.elsevier",
|
||||
"apikey_kimi": "secrets.api_keys.kimi",
|
||||
"apikey_qwen": "secrets.api_keys.qwen",
|
||||
"apikey_wiley": "secrets.api_keys.wiley",
|
||||
"deepseek": "secrets.api_keys.deepseek",
|
||||
"overleaf_token": "secrets.api_keys.overleaf",
|
||||
# Feishu bots
|
||||
"feishu_myagent_appid": "secrets.feishu.myagent.app_id",
|
||||
"feishu_myagent_secret": "secrets.feishu.myagent.secret",
|
||||
"feishu_myclaude_appid": "secrets.feishu.myclaude.app_id",
|
||||
"feishu_myclaude_secret": "secrets.feishu.myclaude.secret",
|
||||
"feishu_xiaohe_appid": "secrets.feishu.xiaohe.app_id",
|
||||
"feishu_xiaohe_secret": "secrets.feishu.xiaohe.secret",
|
||||
# Service accounts
|
||||
"elsevier_email": "secrets.accounts.elsevier.email",
|
||||
"elsevier_password": "secrets.accounts.elsevier.password",
|
||||
"em_username": "secrets.accounts.em.user",
|
||||
"em_password": "secrets.accounts.em.pass",
|
||||
"orcid_email": "secrets.accounts.orcid.email",
|
||||
"orcid_password": "secrets.accounts.orcid.password",
|
||||
"sysu_ta_cms_user": "secrets.accounts.sysu_ta_cms.user",
|
||||
"sysu_ta_cms_pass": "secrets.accounts.sysu_ta_cms.pass",
|
||||
"zs_sysu_username": "secrets.accounts.zs_sysu.user",
|
||||
"zs_sysu_password": "secrets.accounts.zs_sysu.pass",
|
||||
# Volcengine
|
||||
"volc_appid": "secrets.volcengine.app_id",
|
||||
"volc_access_token": "secrets.volcengine.access_token",
|
||||
# Mail
|
||||
"mail_imap_host": "secrets.mail.imap_host",
|
||||
"mail_imap_port": "secrets.mail.imap_port",
|
||||
"mail_smtp_host": "secrets.mail.smtp_host",
|
||||
"mail_smtp_port": "secrets.mail.smtp_port",
|
||||
"mail_email": "secrets.mail.email",
|
||||
"mail_password": "secrets.mail.password",
|
||||
# Paths
|
||||
"path_academia": "paths.academia",
|
||||
"path_apaam": "paths.apaam",
|
||||
"path_research": "paths.research",
|
||||
"path_study": "paths.study",
|
||||
"path_tianhe": "paths.tianhe",
|
||||
"path_webpage": "paths.webpage",
|
||||
"openfoam_build": "paths.openfoam_build",
|
||||
"openfoam_dmg": "paths.openfoam_dmg",
|
||||
"openfoam_mount": "paths.openfoam_mount",
|
||||
# Connections
|
||||
"proxy_fastgithub": "connections.proxy.fastgithub",
|
||||
"proxy_pandafan": "connections.proxy.pandafan",
|
||||
"ssh_starlight": "connections.ssh.starlight.host",
|
||||
"ssh_starlight_port": "connections.ssh.starlight.port",
|
||||
"ssh_tianhe": "connections.ssh.tianhe.host",
|
||||
"ssh_tianhe_key": "connections.ssh.tianhe.key",
|
||||
"ssh_tianhe_port": "connections.ssh.tianhe.port",
|
||||
# Settings
|
||||
"default_tts_voice": "settings.default_tts_voice",
|
||||
# Downloads
|
||||
"downloads_apaam_user": "secrets.downloads.apaam.user",
|
||||
"downloads_apaam_pass": "secrets.downloads.apaam.pass",
|
||||
"downloads_kongyong_user": "secrets.downloads.kongyong.user",
|
||||
"downloads_kongyong_pass": "secrets.downloads.kongyong.pass",
|
||||
"downloads_sysu_group_user": "secrets.downloads.sysu_group.user",
|
||||
"downloads_sysu_group_pass": "secrets.downloads.sysu_group.pass",
|
||||
}
|
||||
_HOME = Path(os.environ.get("MYTOOLKIT_HOME", Path.home() / ".mytoolkit"))
|
||||
CONFIG_PATH = _HOME / "config.json"
|
||||
|
||||
# Reverse map for export / flat listing.
|
||||
_HIERARCHY_TO_LEGACY = {v: k for k, v in _LEGACY_KEY_MAP.items()}
|
||||
|
||||
def _get_path(data: dict, path: str) -> Any:
|
||||
"""Get a value from nested dict using dot-separated path."""
|
||||
parts = path.split(".")
|
||||
d: Any = data
|
||||
for part in parts:
|
||||
if not isinstance(d, dict) or part not in d:
|
||||
return None
|
||||
d = d[part]
|
||||
return d
|
||||
|
||||
|
||||
def _set_path(data: dict, path: str, value: Any) -> None:
|
||||
@@ -101,283 +35,91 @@ def _set_path(data: dict, path: str, value: Any) -> None:
|
||||
data[parts[-1]] = value
|
||||
|
||||
|
||||
def _get_path(data: dict, path: str) -> Any:
|
||||
"""Get a value from nested dict using dot-separated path."""
|
||||
def _remove_path(data: dict, path: str) -> bool:
|
||||
"""Remove a leaf value from nested dict using dot-separated path.
|
||||
|
||||
Returns ``True`` if the leaf existed and was removed.
|
||||
"""
|
||||
parts = path.split(".")
|
||||
d = data
|
||||
for part in parts:
|
||||
d: Any = data
|
||||
for part in parts[:-1]:
|
||||
if not isinstance(d, dict) or part not in d:
|
||||
return None
|
||||
return False
|
||||
d = d[part]
|
||||
return d
|
||||
if isinstance(d, dict) and parts[-1] in d:
|
||||
del d[parts[-1]]
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _walk(data: Any, prefix: str = "") -> dict[str, Any]:
|
||||
"""Return a flat dict of dot-path -> value for nested data."""
|
||||
flat: dict[str, Any] = {}
|
||||
if isinstance(data, dict):
|
||||
for k, v in data.items():
|
||||
key = f"{prefix}.{k}" if prefix else k
|
||||
if isinstance(v, dict):
|
||||
flat.update(_walk(v, key))
|
||||
else:
|
||||
flat[key] = v
|
||||
return flat
|
||||
|
||||
|
||||
class Config:
|
||||
"""Hierarchical config manager for mytoolkit.
|
||||
"""Hierarchical config manager for ``~/.mytoolkit/config.json``."""
|
||||
|
||||
Storage schema (version 2)::
|
||||
|
||||
{
|
||||
"version": 2,
|
||||
"secrets": {"api_keys": {...}, "accounts": {...}, ...},
|
||||
"paths": {...},
|
||||
"connections": {...},
|
||||
"settings": {...}
|
||||
}
|
||||
|
||||
Backward compatible access via legacy flat key names is supported.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
self._config_path = CONFIG_PATH
|
||||
self._migrate_legacy()
|
||||
self._data = self._load()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Migration
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _migrate_legacy(self) -> None:
|
||||
"""One-shot: merge legacy configs into hierarchical config.json."""
|
||||
self._config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
existing: dict = {}
|
||||
if self._config_path.exists():
|
||||
try:
|
||||
existing = json.loads(self._config_path.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
existing = {}
|
||||
|
||||
# Only migrate if current file is old format or missing version.
|
||||
version = existing.get("version")
|
||||
if version == 2:
|
||||
return
|
||||
|
||||
migrated: dict = {}
|
||||
|
||||
def _migrate_flat(data: dict) -> None:
|
||||
"""Migrate flat key/value pairs into migrated hierarchy."""
|
||||
for k, v in data.items():
|
||||
if v in (None, ""):
|
||||
continue
|
||||
path = _LEGACY_KEY_MAP.get(k) or k
|
||||
_set_path(migrated, path, v)
|
||||
|
||||
# Priority (lowest first): external / legacy → mytoolkit old config wins.
|
||||
# Skip external configs in isolated test environments.
|
||||
no_external = os.environ.get("MYTOOLKIT_NO_EXTERNAL_MIGRATION", "0") == "1"
|
||||
# 1) Legacy ~/.mytoolkit/env.json (vars.* → flat keys)
|
||||
if _LEGACY_ENV_PATH.exists():
|
||||
try:
|
||||
legacy = json.loads(_LEGACY_ENV_PATH.read_text())
|
||||
_migrate_flat(legacy.get("vars", {}))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
# 2) ~/.xiaohe/agent/config.json (keys.*)
|
||||
if not no_external and _XIAOHE_CONFIG_PATH.exists():
|
||||
try:
|
||||
xiaohe = json.loads(_XIAOHE_CONFIG_PATH.read_text())
|
||||
_migrate_flat(xiaohe.get("keys", {}))
|
||||
_migrate_flat(xiaohe.get("settings", {}))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
# 3) ~/.xiaohe/agent/settings.json (flat settings)
|
||||
if not no_external and _XIAOHE_SETTINGS_PATH.exists():
|
||||
try:
|
||||
xiaohe_settings = json.loads(_XIAOHE_SETTINGS_PATH.read_text())
|
||||
_migrate_flat(xiaohe_settings)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
# 4) ~/.mytoolkit/mail.json
|
||||
if not no_external and _MAIL_CONFIG_PATH.exists():
|
||||
try:
|
||||
mail = json.loads(_MAIL_CONFIG_PATH.read_text())
|
||||
for k, v in mail.items():
|
||||
if v in (None, ""):
|
||||
continue
|
||||
path = _LEGACY_KEY_MAP.get(f"mail_{k}")
|
||||
if path:
|
||||
_set_path(migrated, path, v)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
# 5) ~/.metabot/bots.json -> secrets.feishu.*
|
||||
if not no_external and _METABOT_BOTS_PATH.exists():
|
||||
try:
|
||||
bots = json.loads(_METABOT_BOTS_PATH.read_text())
|
||||
for bot in bots.get("feishuBots", []):
|
||||
name = bot.get("name")
|
||||
app_id = bot.get("feishuAppId")
|
||||
secret = bot.get("feishuAppSecret")
|
||||
if name:
|
||||
if app_id:
|
||||
_set_path(migrated, f"secrets.feishu.{name}.app_id", app_id)
|
||||
if secret:
|
||||
_set_path(migrated, f"secrets.feishu.{name}.secret", secret)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
# 6) Old flat ~/.mytoolkit/config.json (keys.* / settings.* / top-level)
|
||||
# Highest priority so mytoolkit remains the source of truth.
|
||||
if existing:
|
||||
_migrate_flat(existing.get("keys", {}))
|
||||
_migrate_flat(existing.get("settings", {}))
|
||||
for k, v in existing.items():
|
||||
if k in ("version", "keys", "settings"):
|
||||
continue
|
||||
if v in (None, ""):
|
||||
continue
|
||||
path = _LEGACY_KEY_MAP.get(k)
|
||||
if path:
|
||||
_set_path(migrated, path, v)
|
||||
|
||||
# Drop empty values and normalize.
|
||||
migrated = self._cleanup(migrated)
|
||||
|
||||
# Merge with any existing version 2 data if present.
|
||||
if existing.get("version") == 2:
|
||||
self._deep_merge(existing, migrated)
|
||||
final = existing
|
||||
else:
|
||||
final = migrated
|
||||
|
||||
final["version"] = 2
|
||||
final = self._sort_dicts(final)
|
||||
self._config_path.write_text(json.dumps(final, indent=2, ensure_ascii=False) + "\n")
|
||||
self._config_path.chmod(0o600)
|
||||
|
||||
# Rename legacy env.json out of the way.
|
||||
if _LEGACY_ENV_PATH.exists():
|
||||
backup = _LEGACY_ENV_PATH.with_name("env.json.migrated")
|
||||
if not backup.exists():
|
||||
_LEGACY_ENV_PATH.rename(backup)
|
||||
|
||||
# Very old bundled config.json next to this module (early mytoolkit)
|
||||
if not self._config_path.exists() and _LEGACY_MODULE_PATH.exists():
|
||||
try:
|
||||
_LEGACY_MODULE_PATH.rename(self._config_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _cleanup(data: Any) -> Any:
|
||||
"""Remove None and empty string values recursively."""
|
||||
if isinstance(data, dict):
|
||||
return {k: Config._cleanup(v) for k, v in data.items() if v not in (None, "")}
|
||||
if isinstance(data, list):
|
||||
return [Config._cleanup(v) for v in data if v not in (None, "")]
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _deep_merge(base: dict, override: dict) -> None:
|
||||
"""Merge override into base recursively."""
|
||||
for k, v in override.items():
|
||||
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
|
||||
Config._deep_merge(base[k], v)
|
||||
else:
|
||||
base[k] = v
|
||||
|
||||
@staticmethod
|
||||
def _sort_dicts(data: Any) -> Any:
|
||||
"""Recursively return a new dict with sorted keys."""
|
||||
if isinstance(data, dict):
|
||||
return {k: Config._sort_dicts(data[k]) for k in sorted(data.keys())}
|
||||
if isinstance(data, list):
|
||||
return [Config._sort_dicts(v) for v in data]
|
||||
return data
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Load / Save
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _load(self) -> dict:
|
||||
if self._config_path.exists():
|
||||
try:
|
||||
data = json.loads(self._config_path.read_text())
|
||||
if data.get("version") == 2:
|
||||
data = json.loads(self._config_path.read_text(encoding="utf-8"))
|
||||
if isinstance(data, dict):
|
||||
return data
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
return {"version": 2}
|
||||
return {}
|
||||
|
||||
def save(self) -> None:
|
||||
"""Persist current data to disk with restrictive permissions."""
|
||||
self._config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
sorted_data = self._sort_dicts(self._data)
|
||||
self._config_path.write_text(json.dumps(sorted_data, indent=2, ensure_ascii=False) + "\n")
|
||||
self._config_path.write_text(
|
||||
json.dumps(self._data, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
self._config_path.chmod(0o600)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Accessors
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get(self, name: str) -> Any:
|
||||
"""Get a config value by hierarchical or legacy flat name."""
|
||||
# Hierarchical path.
|
||||
if "." in name:
|
||||
return _get_path(self._data, name)
|
||||
# Legacy flat key inside keys.*
|
||||
legacy_keys = self._data.get("keys", {})
|
||||
if name in legacy_keys:
|
||||
return legacy_keys[name]
|
||||
# Legacy mapping to hierarchical path.
|
||||
new_name = _LEGACY_KEY_MAP.get(name)
|
||||
if new_name:
|
||||
return _get_path(self._data, new_name)
|
||||
# Direct top-level key (e.g., keys set by tests or ad-hoc callers).
|
||||
"""Get a config value by dot-separated hierarchical name."""
|
||||
return _get_path(self._data, name)
|
||||
|
||||
def set(self, name: str, value: Any) -> None:
|
||||
"""Set a config value by hierarchical or legacy flat name."""
|
||||
if "." not in name and name in _LEGACY_KEY_MAP:
|
||||
name = _LEGACY_KEY_MAP[name]
|
||||
"""Set a config value by dot-separated hierarchical name."""
|
||||
_set_path(self._data, name, value)
|
||||
self.save()
|
||||
|
||||
def remove(self, name: str) -> bool:
|
||||
"""Remove a config value by hierarchical or legacy flat name."""
|
||||
if "." not in name and name in _LEGACY_KEY_MAP:
|
||||
name = _LEGACY_KEY_MAP[name]
|
||||
parts = name.split(".")
|
||||
d = self._data
|
||||
for part in parts[:-1]:
|
||||
if not isinstance(d, dict) or part not in d:
|
||||
return False
|
||||
d = d[part]
|
||||
if isinstance(d, dict) and parts[-1] in d:
|
||||
del d[parts[-1]]
|
||||
self.save()
|
||||
return True
|
||||
return False
|
||||
"""Remove a config value by dot-separated hierarchical name."""
|
||||
return _remove_path(self._data, name)
|
||||
|
||||
def get_all(self) -> dict[str, Any]:
|
||||
"""Return all values as a flat dict using legacy key names."""
|
||||
flat: dict[str, Any] = {}
|
||||
|
||||
def walk(data: Any, prefix: str) -> None:
|
||||
if isinstance(data, dict):
|
||||
for k, v in data.items():
|
||||
walk(v, f"{prefix}.{k}" if prefix else k)
|
||||
else:
|
||||
key = _HIERARCHY_TO_LEGACY.get(prefix, prefix)
|
||||
flat[key] = data
|
||||
|
||||
walk(self._data, "")
|
||||
# Remove structural/version keys.
|
||||
flat.pop("version", None)
|
||||
return flat
|
||||
"""Return all values as a flat dict using dot-separated paths."""
|
||||
return _walk(self._data)
|
||||
|
||||
def export(self) -> dict[str, str]:
|
||||
"""Return all values with MYCLI_ prefix for shell eval."""
|
||||
return {f"MYCLI_{k.upper()}": str(v) for k, v in self.get_all().items()}
|
||||
"""Return all values with ``MYCLI_`` prefix for shell eval."""
|
||||
return {
|
||||
f"MYCLI_{k.upper().replace('.', '_')}": str(v)
|
||||
for k, v in self.get_all().items()
|
||||
}
|
||||
|
||||
def resolve_key(self, name: str, env_var: str | None = None) -> str | None:
|
||||
def resolve_key(self, name: str, env_var: str | None = None) -> Any:
|
||||
"""Resolve a key with priority: env > config.
|
||||
|
||||
Args:
|
||||
name: Hierarchical or legacy flat config key.
|
||||
name: Dot-separated hierarchical config key.
|
||||
env_var: Optional environment variable name that overrides config.
|
||||
"""
|
||||
if env_var:
|
||||
@@ -387,5 +129,5 @@ class Config:
|
||||
return self.get(name)
|
||||
|
||||
|
||||
# Global singleton
|
||||
# Module-level singleton.
|
||||
config = Config()
|
||||
|
||||
@@ -9,4 +9,3 @@ import os
|
||||
import tempfile
|
||||
|
||||
os.environ["MYTOOLKIT_HOME"] = tempfile.mkdtemp(prefix="mytoolkit-test-home-")
|
||||
os.environ["MYTOOLKIT_NO_EXTERNAL_MIGRATION"] = "1"
|
||||
|
||||
+52
-43
@@ -1,10 +1,4 @@
|
||||
"""Tests for mytoolkit.config.Config (get/set/remove/export + legacy migration).
|
||||
|
||||
MYTOOLKIT_HOME is pointed at a temp dir by conftest.py before imports, so
|
||||
every Config() instance here operates on throwaway files.
|
||||
"""
|
||||
|
||||
import json
|
||||
"""Tests for mytoolkit.config.Config (hierarchical get/set/remove/export)."""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -17,21 +11,44 @@ def fresh_home(tmp_path, monkeypatch):
|
||||
"""Give each test its own empty MYTOOLKIT_HOME."""
|
||||
home = tmp_path / "home"
|
||||
monkeypatch.setattr(config_mod, "CONFIG_PATH", home / "config.json")
|
||||
monkeypatch.setattr(config_mod, "_LEGACY_ENV_PATH", home / "env.json")
|
||||
monkeypatch.setattr(config_mod, "_LEGACY_MODULE_PATH", home / "module-config.json")
|
||||
return home
|
||||
|
||||
|
||||
def test_set_get_remove_roundtrip(fresh_home):
|
||||
c = Config()
|
||||
assert c.get("apikey_ark") is None
|
||||
c.set("apikey_ark", "sk-test")
|
||||
assert c.get("apikey_ark") == "sk-test"
|
||||
assert c.get("secrets.api_keys.ark") is None
|
||||
c.set("secrets.api_keys.ark", "sk-test")
|
||||
assert c.get("secrets.api_keys.ark") == "sk-test"
|
||||
# Persisted: a fresh instance sees the same value.
|
||||
assert Config().get("apikey_ark") == "sk-test"
|
||||
assert c.remove("apikey_ark") is True
|
||||
assert c.remove("apikey_ark") is False
|
||||
assert c.get("apikey_ark") is None
|
||||
assert Config().get("secrets.api_keys.ark") == "sk-test"
|
||||
assert c.remove("secrets.api_keys.ark") is True
|
||||
assert c.remove("secrets.api_keys.ark") is False
|
||||
assert c.get("secrets.api_keys.ark") is None
|
||||
|
||||
|
||||
def test_nested_set_creates_intermediate_dicts(fresh_home):
|
||||
c = Config()
|
||||
c.set("connections.ssh.workstation.host", "10.0.0.5")
|
||||
assert c.get("connections.ssh.workstation.host") == "10.0.0.5"
|
||||
assert c.get("connections.ssh") == {"workstation": {"host": "10.0.0.5"}}
|
||||
|
||||
|
||||
def test_remove_nested_cleans_leaf_but_keeps_parents(fresh_home):
|
||||
c = Config()
|
||||
c.set("a.b.c", 1)
|
||||
assert c.remove("a.b.c") is True
|
||||
assert c.get("a.b.c") is None
|
||||
assert c.get("a.b") == {}
|
||||
|
||||
|
||||
def test_get_all_returns_flat_dot_paths(fresh_home):
|
||||
c = Config()
|
||||
c.set("secrets.api_keys.ark", "sk-ark")
|
||||
c.set("paths.cv", "/tmp/cv")
|
||||
assert c.get_all() == {
|
||||
"secrets.api_keys.ark": "sk-ark",
|
||||
"paths.cv": "/tmp/cv",
|
||||
}
|
||||
|
||||
|
||||
def test_get_all_returns_copy(fresh_home):
|
||||
@@ -42,10 +59,23 @@ def test_get_all_returns_copy(fresh_home):
|
||||
assert c.get("k") == "v"
|
||||
|
||||
|
||||
def test_export_prefix(fresh_home):
|
||||
def test_export_prefix_and_underscore_replacement(fresh_home):
|
||||
c = Config()
|
||||
c.set("volc_appid", "123")
|
||||
assert c.export() == {"MYCLI_VOLC_APPID": "123"}
|
||||
c.set("secrets.api_keys.ark", "sk-test")
|
||||
assert c.export() == {"MYCLI_SECRETS_API_KEYS_ARK": "sk-test"}
|
||||
|
||||
|
||||
def test_resolve_key_env_overrides_config(fresh_home, monkeypatch):
|
||||
c = Config()
|
||||
c.set("secrets.api_keys.ark", "sk-config")
|
||||
monkeypatch.setenv("ARK_API_KEY", "sk-env")
|
||||
assert c.resolve_key("secrets.api_keys.ark", env_var="ARK_API_KEY") == "sk-env"
|
||||
|
||||
|
||||
def test_resolve_key_falls_back_to_config(fresh_home):
|
||||
c = Config()
|
||||
c.set("secrets.api_keys.ark", "sk-config")
|
||||
assert c.resolve_key("secrets.api_keys.ark", env_var="MISSING_VAR") == "sk-config"
|
||||
|
||||
|
||||
def test_corrupt_config_falls_back_to_empty(fresh_home):
|
||||
@@ -55,28 +85,7 @@ def test_corrupt_config_falls_back_to_empty(fresh_home):
|
||||
assert c.get_all() == {}
|
||||
|
||||
|
||||
def test_legacy_env_json_migrated(fresh_home):
|
||||
fresh_home.mkdir(parents=True)
|
||||
(fresh_home / "env.json").write_text(
|
||||
json.dumps({"vars": {"apikey_ark": "sk-old", "path_study": "/tmp/x"}})
|
||||
)
|
||||
def test_file_permissions_are_restrictive(fresh_home):
|
||||
c = Config()
|
||||
assert c.get("apikey_ark") == "sk-old"
|
||||
assert c.get("path_study") == "/tmp/x"
|
||||
# env.json renamed out of the way so migration does not re-run.
|
||||
assert not (fresh_home / "env.json").exists()
|
||||
assert (fresh_home / "env.json.migrated").exists()
|
||||
|
||||
|
||||
def test_legacy_env_merges_without_clobbering_existing_keys(fresh_home):
|
||||
fresh_home.mkdir(parents=True)
|
||||
(fresh_home / "config.json").write_text(
|
||||
json.dumps({"keys": {"apikey_ark": "sk-new"}})
|
||||
)
|
||||
(fresh_home / "env.json").write_text(
|
||||
json.dumps({"vars": {"apikey_ark": "sk-old", "extra": "e"}})
|
||||
)
|
||||
c = Config()
|
||||
# Unknown legacy vars are merged in; known keys keep the existing mytoolkit value.
|
||||
assert c.get("extra") == "e"
|
||||
assert c.get("apikey_ark") == "sk-new"
|
||||
c.set("k", "v")
|
||||
assert (fresh_home / "config.json").stat().st_mode & 0o777 == 0o600
|
||||
|
||||
Reference in New Issue
Block a user