feat(issue-45): herdr config for MAM servers with restore-on-restart off
lib_py/herdr_config.py builds the user's herdr config (HERDR_CONFIG_PATH, else XDG_CONFIG_HOME, else ~/.config, as herdr reads it) with only [session] resume_agents_on_restore forced to false, written next to the source so relative paths resolve the same. The text merge is re-parsed; a merge that does not leave the rest equal falls back to the key alone, since herdr drops a config that does not parse and would restore again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed
1 parent
c5c7eff6f7
commit
a640b0f072
2 files changed
+388
No files matched your search
@@ -0,0 +1,192 @@
|
||||
"""Issue #45 (A): the herdr config MAM's isolated servers start with.
|
||||
|
||||
lib_py/herdr_config.py builds the user's herdr config with only
|
||||
[session] resume_agents_on_restore forced to false. herdr drops a config that
|
||||
does not parse and runs on defaults - restore on - so every result here is
|
||||
re-parsed, and fed to the real `herdr config check` when herdr is installed.
|
||||
That restore really is off on a live server is test_issue45_restore.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / ".agents" / "skills"))
|
||||
|
||||
from lib_py import herdr_config as hc # noqa: E402
|
||||
|
||||
HERDR = shutil.which("herdr")
|
||||
|
||||
# (name, user config text or None, expect the user's other settings kept, expect a warning)
|
||||
CASES = [
|
||||
("missing", None, True, False),
|
||||
("empty", "", True, False),
|
||||
("broken", "[session]\n[session]\n", False, True),
|
||||
("no_session", "[ui]\nconfirm_close = false\n", True, False),
|
||||
("no_session_no_newline", "[ui]\nconfirm_close = false", True, False),
|
||||
("session_without_key", "[session]\n# a comment\n\n[ui]\nconfirm_close = false\n", True, False),
|
||||
("key_true", "[ui]\nconfirm_close = false\n[session]\nresume_agents_on_restore = true # mine\n", True, False),
|
||||
("key_false", "[session]\nresume_agents_on_restore=false\n", True, False),
|
||||
("sub_table_only", "[session.extra]\na = 1\n", True, False),
|
||||
("dotted_key", "session.resume_agents_on_restore = true\n", False, True),
|
||||
("inline_table", "session = { resume_agents_on_restore = true }\n", False, True),
|
||||
("header_inside_string", 'banner = """\n[session]\n"""\n', False, True),
|
||||
]
|
||||
|
||||
|
||||
def _without_key(d):
|
||||
return hc._without_key(d)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name,user,kept,warned", CASES, ids=[c[0] for c in CASES])
|
||||
def test_merge_always_parses_with_restore_off(name, user, kept, warned):
|
||||
text, warnings = hc.server_config_text(user)
|
||||
parsed = tomllib.loads(text)
|
||||
assert parsed["session"]["resume_agents_on_restore"] is False
|
||||
assert bool(warnings) == warned, warnings
|
||||
if kept and user:
|
||||
assert _without_key(parsed) == _without_key(tomllib.loads(user))
|
||||
if not kept:
|
||||
assert text == hc.MINIMAL
|
||||
|
||||
|
||||
def test_existing_false_is_left_byte_for_byte():
|
||||
user = "[session]\nresume_agents_on_restore=false # keep\n"
|
||||
assert hc.server_config_text(user) == (user, [])
|
||||
|
||||
|
||||
def test_a_bad_value_type_falls_back_to_minimal():
|
||||
# herdr rejects a non-bool here as a parse error of the whole file.
|
||||
text, warnings = hc.server_config_text('[session]\nresume_agents_on_restore = "no"\n')
|
||||
assert tomllib.loads(text)["session"]["resume_agents_on_restore"] is False
|
||||
# tomllib reads the string fine, so the merge replaces the line instead.
|
||||
assert warnings == []
|
||||
|
||||
|
||||
def _env(tmp_path, **kw):
|
||||
env = {"HOME": str(tmp_path / "home")}
|
||||
env.update(kw)
|
||||
return env
|
||||
|
||||
|
||||
def test_source_is_home_config_by_default(tmp_path):
|
||||
assert hc.user_config_path(_env(tmp_path)) == str(tmp_path / "home" / ".config" / "herdr" / "config.toml")
|
||||
|
||||
|
||||
def test_source_follows_xdg_config_home(tmp_path, monkeypatch):
|
||||
assert hc.user_config_path(_env(tmp_path, XDG_CONFIG_HOME=str(tmp_path / "x"))) == \
|
||||
str(tmp_path / "x" / "herdr" / "config.toml")
|
||||
monkeypatch.chdir(tmp_path) # herdr takes a relative one against its cwd
|
||||
assert hc.user_config_path(_env(tmp_path, XDG_CONFIG_HOME="rel")) == \
|
||||
str(tmp_path / "rel" / "herdr" / "config.toml")
|
||||
|
||||
|
||||
def test_source_is_herdr_config_path_unless_generated(tmp_path):
|
||||
user = tmp_path / "custom.toml"
|
||||
user.write_text("[ui]\nconfirm_close = false\n")
|
||||
env = _env(tmp_path, HERDR_CONFIG_PATH=str(user), XDG_CONFIG_HOME=str(tmp_path / "x"))
|
||||
assert hc.user_config_path(env) == str(user)
|
||||
|
||||
# A MAM pane inherits HERDR_CONFIG_PATH=<generated>; a server bootstrapped
|
||||
# from there must still build from the user's file, not from the copy.
|
||||
generated, _ = hc.write_server_config(env)
|
||||
assert generated == str(tmp_path / hc.OUTPUT_NAME)
|
||||
assert hc.user_config_path(dict(env, HERDR_CONFIG_PATH=generated)) == str(user)
|
||||
|
||||
# A generated file without a source line falls through to XDG / HOME.
|
||||
bare = tmp_path / "bare.toml"
|
||||
bare.write_text(hc.MARKER + "\n" + hc.MINIMAL)
|
||||
assert hc.user_config_path(dict(env, HERDR_CONFIG_PATH=str(bare))) == \
|
||||
str(tmp_path / "x" / "herdr" / "config.toml")
|
||||
|
||||
|
||||
def test_written_next_to_source_with_its_mode(tmp_path):
|
||||
cfg_dir = tmp_path / "home" / ".config" / "herdr"
|
||||
cfg_dir.mkdir(parents=True)
|
||||
user = cfg_dir / "config.toml"
|
||||
user.write_text('[ui.sound]\npath = "sounds/done.wav"\n')
|
||||
user.chmod(0o600)
|
||||
path, warnings = hc.write_server_config(_env(tmp_path))
|
||||
assert path == str(cfg_dir / hc.OUTPUT_NAME) and warnings == []
|
||||
text = Path(path).read_text()
|
||||
assert text.splitlines()[:2] == [hc.MARKER, hc.SOURCE_PREFIX + str(user)]
|
||||
parsed = tomllib.loads(text)
|
||||
# same directory, so the relative sound path resolves to the same file
|
||||
assert parsed["ui"]["sound"]["path"] == "sounds/done.wav"
|
||||
assert parsed["session"]["resume_agents_on_restore"] is False
|
||||
assert stat.S_IMODE(os.stat(path).st_mode) == 0o600
|
||||
assert user.read_text() == '[ui.sound]\npath = "sounds/done.wav"\n' # untouched
|
||||
assert [p.name for p in cfg_dir.iterdir() if p.name.endswith(".tmp")] == []
|
||||
|
||||
|
||||
def test_written_without_any_user_config(tmp_path):
|
||||
path, warnings = hc.write_server_config(_env(tmp_path))
|
||||
assert path == str(tmp_path / "home" / ".config" / "herdr" / hc.OUTPUT_NAME) and warnings == []
|
||||
assert stat.S_IMODE(os.stat(path).st_mode) == 0o644
|
||||
assert tomllib.loads(Path(path).read_text()) == {"session": {"resume_agents_on_restore": False}}
|
||||
|
||||
|
||||
def test_rewritten_on_every_call(tmp_path):
|
||||
cfg_dir = tmp_path / "home" / ".config" / "herdr"
|
||||
cfg_dir.mkdir(parents=True)
|
||||
(cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = true\n")
|
||||
path, _ = hc.write_server_config(_env(tmp_path))
|
||||
(cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = false\n")
|
||||
hc.write_server_config(_env(tmp_path))
|
||||
assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions")
|
||||
def test_unwritable_config_dir_falls_back_to_mam_dir(tmp_path):
|
||||
cfg_dir = tmp_path / "home" / ".config" / "herdr"
|
||||
cfg_dir.mkdir(parents=True)
|
||||
(cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = false\n")
|
||||
cfg_dir.chmod(0o555)
|
||||
try:
|
||||
mam = tmp_path / ".mam"
|
||||
mam.mkdir()
|
||||
path, warnings = hc.write_server_config(_env(tmp_path), str(mam))
|
||||
assert path == str(mam / hc.FALLBACK_NAME)
|
||||
assert any("relative paths" in w for w in warnings), warnings
|
||||
assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False
|
||||
assert hc.write_server_config(_env(tmp_path))[0] is None # no fallback given
|
||||
finally:
|
||||
cfg_dir.chmod(0o755)
|
||||
|
||||
|
||||
def test_cli_prints_the_path(tmp_path):
|
||||
env = dict(os.environ, HOME=str(tmp_path), PYTHONPATH=str(REPO_ROOT / ".agents" / "skills"))
|
||||
env.pop("HERDR_CONFIG_PATH", None)
|
||||
env.pop("XDG_CONFIG_HOME", None)
|
||||
cfg_dir = tmp_path / ".config" / "herdr"
|
||||
cfg_dir.mkdir(parents=True)
|
||||
(cfg_dir / "config.toml").write_text("[session]\n[session]\n")
|
||||
res = subprocess.run([sys.executable, "-m", "lib_py.herdr_config", "server-config"],
|
||||
capture_output=True, text=True, env=env)
|
||||
assert res.returncode == 0, res.stderr
|
||||
assert res.stdout == str(cfg_dir / hc.OUTPUT_NAME) + "\n"
|
||||
assert "WARNING: herdr config does not parse" in res.stderr
|
||||
|
||||
|
||||
@pytest.mark.skipif(not HERDR, reason="herdr is not installed")
|
||||
@pytest.mark.parametrize("name,user,kept,warned", CASES, ids=[c[0] for c in CASES])
|
||||
def test_real_herdr_accepts_every_result(tmp_path, name, user, kept, warned):
|
||||
text, _ = hc.server_config_text(user)
|
||||
cfg = tmp_path / "server.toml"
|
||||
cfg.write_text(hc.MARKER + "\n" + text)
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}
|
||||
env.update(HOME=str(tmp_path), HERDR_CONFIG_PATH=str(cfg))
|
||||
res = subprocess.run([HERDR, "config", "check"], capture_output=True, text=True, env=env)
|
||||
out = res.stdout + res.stderr
|
||||
assert "parse error" not in out, out
|
||||
# the fixtures use only keys herdr knows, apart from the made-up ones below
|
||||
issues = [l for l in out.splitlines()[1:] if l.strip()]
|
||||
assert all(l.startswith("unknown config key") and ("extra" in l or "banner" in l) for l in issues), out
|
||||
Reference in new issue
Block a user