Plan 3: load/save ~/assistant/state/config.json with mode 0600
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
"""Read / write the device config file at ~/assistant/state/config.json."""
|
||||
import json
|
||||
import os
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class ConfigMissingError(RuntimeError):
|
||||
"""Raised when config.json is missing or malformed."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
device_id: str
|
||||
device_token: str
|
||||
backend_ws: str
|
||||
|
||||
|
||||
def config_path() -> Path:
|
||||
return Path(os.environ["HOME"]) / "assistant" / "state" / "config.json"
|
||||
|
||||
|
||||
def load_config() -> Config:
|
||||
path = config_path()
|
||||
if not path.exists():
|
||||
raise ConfigMissingError(f"{path} not found; run `python -m client.pair` first")
|
||||
try:
|
||||
raw = json.loads(path.read_text())
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ConfigMissingError(f"{path} is not valid JSON: {exc}") from exc
|
||||
try:
|
||||
return Config(
|
||||
device_id=str(raw["device_id"]),
|
||||
device_token=str(raw["device_token"]),
|
||||
backend_ws=str(raw["backend_ws"]),
|
||||
)
|
||||
except KeyError as exc:
|
||||
raise ConfigMissingError(f"{path} missing field: {exc}") from exc
|
||||
|
||||
|
||||
def save_config(cfg: Config) -> None:
|
||||
path = config_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
# Write to a temp file then atomically rename, so a partial write never
|
||||
# leaves a half-baked config on disk.
|
||||
tmp = path.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(asdict(cfg), indent=2) + "\n")
|
||||
os.chmod(tmp, 0o600)
|
||||
os.replace(tmp, path)
|
||||
@@ -0,0 +1,59 @@
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
|
||||
import pytest
|
||||
|
||||
from client.config import (
|
||||
Config,
|
||||
ConfigMissingError,
|
||||
config_path,
|
||||
load_config,
|
||||
save_config,
|
||||
)
|
||||
|
||||
|
||||
def test_config_path_under_assistant_state(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
assert config_path() == tmp_path / "assistant" / "state" / "config.json"
|
||||
|
||||
|
||||
def test_load_missing_raises(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
with pytest.raises(ConfigMissingError):
|
||||
load_config()
|
||||
|
||||
|
||||
def test_save_then_load_round_trip(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
cfg = Config(
|
||||
device_id="11111111-1111-1111-1111-111111111111",
|
||||
device_token="t0k3n",
|
||||
backend_ws="wss://example/device",
|
||||
)
|
||||
save_config(cfg)
|
||||
loaded = load_config()
|
||||
assert loaded == cfg
|
||||
|
||||
|
||||
def test_save_creates_parent_dirs(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
cfg = Config(device_id="x", device_token="y", backend_ws="wss://z")
|
||||
save_config(cfg)
|
||||
assert (tmp_path / "assistant" / "state").is_dir()
|
||||
|
||||
|
||||
def test_save_sets_mode_0600(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
cfg = Config(device_id="x", device_token="y", backend_ws="wss://z")
|
||||
save_config(cfg)
|
||||
mode = stat.S_IMODE(os.stat(config_path()).st_mode)
|
||||
assert mode == 0o600
|
||||
|
||||
|
||||
def test_load_rejects_missing_fields(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
config_path().parent.mkdir(parents=True, exist_ok=True)
|
||||
config_path().write_text(json.dumps({"device_id": "x"}))
|
||||
with pytest.raises(ConfigMissingError):
|
||||
load_config()
|
||||
Reference in New Issue
Block a user