GodopuandClaude Opus 5.5
811b080966
R1: tomllib accepting a file is not herdr accepting it (a schema error makes
herdr drop the whole file, restore included). The shim passes REAL_HERDR and
the written file must pass `herdr config check` ("parse error"/"using defaults"
reject; unknown keys and sound-file notices do not), else the key-only file is
written and checked. A non-table `session` and a source that is not UTF-8 reach
that fallback instead of raising.
R2: the line edit is replaced by parse -> set -> serialize; the result must
parse back to the source plus the key, so dotted keys and inline tables keep
the user's other settings.
R3: the output is never the source (aliases included) nor a file not generated
for the same source; then a per-source name beside it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
293 lines
14 KiB
Python
293 lines
14 KiB
Python
"""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),
|
|
("session_not_a_table", "session = 1\n[ui]\nconfirm_close = false\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),
|
|
# valid spellings a line edit cannot reach: the rest must survive (review R2)
|
|
("dotted_key", "session.resume_agents_on_restore = true\n[ui]\nconfirm_close = false\n", True, False),
|
|
("inline_table", "session = { resume_agents_on_restore = true }\n[ui]\nconfirm_close = false\n", True, False),
|
|
("header_inside_string", 'banner = """\n[session]\n"""\n[ui]\nconfirm_close = false\n', True, False),
|
|
("value_kinds", 'top = "a\\"b\\u007f\\u00e9"\n"odd key" = [1, 2.5, inf, true]\n'
|
|
'when = 1979-05-27T07:32:00Z\n[ui.sound]\npath = "s/done.wav"\n'
|
|
'[[keys.bind]]\nkey = "a"\n[[keys.bind]]\nkey = "b"\n', True, False),
|
|
]
|
|
|
|
|
|
def _without_key(d):
|
|
d = dict(d)
|
|
session = dict(d.get("session") or {})
|
|
session.pop("resume_agents_on_restore", None)
|
|
if session:
|
|
d["session"] = session
|
|
else:
|
|
d.pop("session", None)
|
|
return 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_a_non_bool_restore_value_is_replaced():
|
|
# herdr would reject the whole file for this; the key is ours to set anyway.
|
|
text, warnings = hc.server_config_text('[session]\nresume_agents_on_restore = "no"\n')
|
|
assert tomllib.loads(text) == {"session": {"resume_agents_on_restore": False}} and warnings == []
|
|
|
|
|
|
def test_without_tomllib_restore_is_still_off(monkeypatch):
|
|
# The shim runs a bare `python3`; on one older than 3.11 an import error
|
|
# would leave the server with restore on. Off wins over the user's settings.
|
|
monkeypatch.setattr(hc, "tomllib", None)
|
|
text, warnings = hc.server_config_text("[ui]\nconfirm_close = false\n")
|
|
assert text == hc.MINIMAL and "no tomllib" in warnings[0]
|
|
|
|
|
|
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):
|
|
if kept and user: # a source herdr itself takes must stay one it takes
|
|
src = tmp_path / "source.toml"
|
|
src.write_text(user)
|
|
assert hc.herdr_rejects(str(src), HERDR) is None
|
|
text, _ = hc.server_config_text(user)
|
|
cfg = tmp_path / "server.toml"
|
|
cfg.write_text(hc.MARKER + "\n" + text)
|
|
assert hc.herdr_rejects(str(cfg), HERDR) is None
|
|
|
|
|
|
@pytest.mark.skipif(not HERDR, reason="herdr is not installed")
|
|
def test_herdr_rejects_tells_schema_errors_from_harmless_issues(tmp_path):
|
|
bad = tmp_path / "bad.toml"
|
|
bad.write_text('[ui]\nconfirm_close = "no"\n')
|
|
assert "using defaults" in hc.herdr_rejects(str(bad), HERDR)
|
|
soft = tmp_path / "soft.toml"
|
|
soft.write_text('made_up = 1\n[ui.sound]\npath = "x.wav"\n')
|
|
assert hc.herdr_rejects(str(soft), HERDR) is None
|
|
|
|
|
|
def _cli(tmp_path, source: Path, *extra) -> subprocess.CompletedProcess:
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_") and k != "XDG_CONFIG_HOME"}
|
|
env.update(HOME=str(tmp_path), HERDR_CONFIG_PATH=str(source),
|
|
PYTHONPATH=str(REPO_ROOT / ".agents" / "skills"))
|
|
return subprocess.run([sys.executable, "-m", "lib_py.herdr_config", "server-config", *extra],
|
|
capture_output=True, text=True, env=env)
|
|
|
|
|
|
@pytest.mark.skipif(not HERDR, reason="herdr is not installed")
|
|
@pytest.mark.parametrize("body,kept", [
|
|
(b"[ui]\nconfirm_close = false\n", True), # control: valid, kept
|
|
(b'[ui]\nconfirm_close = "no"\n', False), # tomllib yes, herdr's schema no (review R1)
|
|
(b"session = 1\n[ui]\nconfirm_close = false\n", False), # non-table session (R1)
|
|
(b"[ui]\nconfirm_close = false\n# \xff\xfe\n", False), # not UTF-8 (R1)
|
|
], ids=["valid", "schema-error", "session-scalar", "invalid-utf8"])
|
|
def test_cli_always_hands_herdr_a_restore_off_config_it_accepts(tmp_path, body, kept):
|
|
source = tmp_path / "custom.toml"
|
|
source.write_bytes(body)
|
|
res = _cli(tmp_path, source, "--herdr", HERDR)
|
|
assert res.returncode == 0, res.stderr
|
|
out = Path(res.stdout.strip())
|
|
assert out.parent == tmp_path and out != source
|
|
assert hc.herdr_rejects(str(out), HERDR) is None
|
|
parsed = tomllib.loads(out.read_text())
|
|
assert parsed["session"]["resume_agents_on_restore"] is False
|
|
if kept:
|
|
assert parsed["ui"]["confirm_close"] is False and res.stderr == ""
|
|
else:
|
|
assert parsed == {"session": {"resume_agents_on_restore": False}} and "WARNING" in res.stderr
|
|
assert source.read_bytes() == body
|
|
|
|
|
|
@pytest.mark.skipif(not HERDR, reason="herdr is not installed")
|
|
def test_a_config_herdr_rejects_is_never_left_in_place(tmp_path):
|
|
# the merge itself is fine for tomllib; only herdr's schema says no
|
|
source = tmp_path / "config.toml"
|
|
source.write_text('[ui]\nconfirm_close = "no"\n[session]\nresume_agents_on_restore = true\n')
|
|
path, warnings = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}, herdr=HERDR)
|
|
assert tomllib.loads(Path(path).read_text()) == {"session": {"resume_agents_on_restore": False}}
|
|
assert any("herdr rejects" in w for w in warnings), warnings
|
|
|
|
|
|
# --- the user's file is never the output (review R3) -----------------------
|
|
|
|
def test_source_named_like_the_output_is_not_overwritten(tmp_path):
|
|
source = tmp_path / hc.OUTPUT_NAME
|
|
source.write_bytes(b"[ui]\nconfirm_close = false\n")
|
|
path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)})
|
|
assert source.read_bytes() == b"[ui]\nconfirm_close = false\n"
|
|
assert Path(path).parent == tmp_path and Path(path) != source
|
|
assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False
|
|
# and the next start finds the same file again
|
|
assert hc.write_server_config({"HERDR_CONFIG_PATH": str(source)})[0] == path
|
|
|
|
|
|
def test_an_alias_of_the_source_is_not_overwritten(tmp_path):
|
|
source = tmp_path / "config.toml"
|
|
source.write_bytes(b"[ui]\nconfirm_close = false\n")
|
|
(tmp_path / hc.OUTPUT_NAME).symlink_to(source)
|
|
path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)})
|
|
assert source.read_bytes() == b"[ui]\nconfirm_close = false\n"
|
|
assert os.path.islink(tmp_path / hc.OUTPUT_NAME)
|
|
assert not os.path.samefile(path, source)
|
|
|
|
|
|
def test_a_foreign_file_with_the_output_name_is_kept(tmp_path):
|
|
source = tmp_path / "config.toml"
|
|
source.write_text("[ui]\nconfirm_close = false\n")
|
|
foreign = tmp_path / hc.OUTPUT_NAME
|
|
foreign.write_bytes(b"# someone else's\n")
|
|
path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)})
|
|
assert foreign.read_bytes() == b"# someone else's\n" and Path(path) != foreign
|
|
# a file generated for another source is someone else's too
|
|
other = tmp_path / "other.toml"
|
|
other.write_text("")
|
|
other_path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(other)})
|
|
assert other_path not in (path, str(foreign))
|
|
assert hc.user_config_path({"HERDR_CONFIG_PATH": path}) == str(source)
|