multi-agent-mux/tests/test_issue45_herdr_config.py

"""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
from pathlib import Path

try:  # Python 3.11+; on older Pythons herdr_config writes the key-only config
    import tomllib
except ImportError:  # pragma: no cover - depends on the interpreter
    tomllib = None

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

# Python < 3.11 has no tomllib: herdr_config then writes the key-only config
# (restore still off, the user's other settings dropped, with a warning), so
# the tests that parse a result or expect the user's settings kept need it.
# test_without_tomllib_restore_is_still_off covers that fallback everywhere.
needs_tomllib = pytest.mark.skipif(tomllib is None, reason="needs tomllib (Python 3.11+)")

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


@needs_tomllib
@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


@needs_tomllib
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")


@needs_tomllib
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")] == []


@needs_tomllib
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}}


@needs_tomllib
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


@needs_tomllib
@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
        if HERDR:
            assert hc.herdr_rejects(path, HERDR) is None
        assert hc.write_server_config(_env(tmp_path))[0] is None  # no fallback given
    finally:
        cfg_dir.chmod(0o755)


# --- the fallback obeys the same rules (review round 2, R3) ----------------

def _tag(source: Path) -> str:
    import hashlib
    return hashlib.sha256(os.path.realpath(source).encode()).hexdigest()[:12]


@needs_tomllib
@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions")
def test_a_foreign_fallback_file_is_kept(tmp_path):
    cfg_dir = tmp_path / "cfg"
    cfg_dir.mkdir()
    source = cfg_dir / "config.toml"
    source.write_text("[ui]\nconfirm_close = false\n")
    cfg_dir.chmod(0o555)
    mam = tmp_path / ".mam"
    mam.mkdir()
    foreign = mam / hc.FALLBACK_NAME
    foreign.write_bytes(b"# user-owned\n[ui]\nconfirm_close = true\n")
    try:
        path, warnings = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}, str(mam), HERDR)
    finally:
        cfg_dir.chmod(0o755)
    assert foreign.read_bytes() == b"# user-owned\n[ui]\nconfirm_close = true\n"
    assert path == str(mam / f"herdr-server-config-{_tag(source)}.toml"), warnings
    assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False


@needs_tomllib
@pytest.mark.parametrize("alias", [False, True], ids=["same-path", "symlink-alias"])
def test_the_fallback_is_never_the_source(tmp_path, alias):
    mam = tmp_path / ".mam"
    mam.mkdir()
    body = b"[ui]\nconfirm_close = false\n"
    if alias:
        source = tmp_path / "real.toml"
        source.write_bytes(body)
        (mam / hc.FALLBACK_NAME).symlink_to(source)
        source_dir = tmp_path
    else:
        source = mam / hc.FALLBACK_NAME
        source.write_bytes(body)
        source_dir = mam
    # both names beside the source are someone else's, so the fallback is used
    for name in (hc.OUTPUT_NAME, f"mam-server-config-{_tag(source)}.toml"):
        (source_dir / name).write_bytes(b"# foreign\n")
    path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}, str(mam), HERDR)
    assert source.read_bytes() == body
    if alias:  # os.replace would swap out the user's link itself
        assert (mam / hc.FALLBACK_NAME).is_symlink()
    assert path and not os.path.samefile(path, source)
    assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False


def test_no_write_when_every_name_is_taken(tmp_path):
    mam = tmp_path / ".mam"
    mam.mkdir()
    source = tmp_path / "config.toml"
    source.write_text("[ui]\nconfirm_close = false\n")
    taken = [tmp_path / hc.OUTPUT_NAME, tmp_path / f"mam-server-config-{_tag(source)}.toml",
             mam / hc.FALLBACK_NAME, mam / f"herdr-server-config-{_tag(source)}.toml"]
    for f in taken:
        f.write_bytes(b"# foreign\n")
    path, warnings = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}, str(mam))
    assert path is None and any("no free output name" in w for w in warnings), warnings
    assert all(f.read_bytes() == b"# foreign\n" for f in taken)


@needs_tomllib
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"])
@needs_tomllib
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


@needs_tomllib
@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) -----------------------

@needs_tomllib
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)