hermes-agent/tests/test_atomic_replace_symlinks.py
Teknium b61d9b297a refactor: consolidate symlink-safe atomic replace into shared helper
Extract the islink/realpath guard from the 16743 fix into a single
atomic_replace() helper in utils.py, then migrate every os.replace()
call site in the codebase to use it.

The original PR #16777 correctly identified and fixed the bug, but
only patched 9 of ~24 call sites. The same bug class (managed
deployments that symlink state files silently losing the link on
every write) still existed at auth.json, sessions file, gateway
config, env_loader, webhook subscriptions, debug store, model
catalog, pairing, google OAuth, nous rate guard, and more.

Rather than add another 10+ copies of the same three-line guard,
consolidate into atomic_replace(tmp, target) which:
- resolves symlinks via os.path.realpath before os.replace
- returns the resolved real path so callers can re-apply permissions
- is a drop-in replacement for os.replace at the use sites

Changes:
- utils.py: new atomic_replace() helper + atomic_json_write /
  atomic_yaml_write now call it instead of inlining the guard
- 16 files: all os.replace() call sites migrated to atomic_replace()
  - agent/{google_oauth, nous_rate_guard, shell_hooks}.py
  - cron/jobs.py
  - gateway/{pairing, session, platforms/telegram}.py
  - hermes_cli/{auth, config, debug, env_loader, model_catalog, webhook}.py
  - tools/{memory_tool, skill_manager_tool, skills_sync}.py

Tests: tests/test_atomic_replace_symlinks.py pins the invariant for
atomic_replace + atomic_json_write + atomic_yaml_write, covers plain
files, first-time creates, broken symlinks, and permission preservation.

Refs #16743
Builds on #16777 by @vominh1919.
2026-04-28 04:58:22 -07:00

160 lines
5.5 KiB
Python

"""Regression tests for GitHub #16743 — atomic writes must preserve symlinks.
``os.replace(tmp, target)`` replaces whatever exists at ``target`` — including
symlinks, which it swaps for a regular file. Managed deployments that
symlink ``~/.hermes/config.yaml`` (and other state files) to a git-tracked
profile package were silently detached on every config write.
The fix: a shared ``atomic_replace`` helper in ``utils.py`` that resolves the
target through ``os.path.realpath`` when it is a symlink, so the real file is
overwritten in-place while the symlink survives. All atomic-write sites in
the codebase were migrated to the helper; these tests pin that invariant.
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
import pytest
import yaml
# Ensure the repo root is importable when running via `pytest tests/...`.
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
from utils import atomic_json_write, atomic_replace, atomic_yaml_write
# ─── Direct helper ────────────────────────────────────────────────────────────
def _write_tmp(dir_: Path, content: str) -> Path:
tmp = dir_ / ".src.tmp"
tmp.write_text(content, encoding="utf-8")
return tmp
def test_atomic_replace_preserves_symlink(tmp_path: Path) -> None:
real = tmp_path / "real.yaml"
link = tmp_path / "link.yaml"
real.write_text("original\n", encoding="utf-8")
link.symlink_to(real)
tmp = _write_tmp(tmp_path, "updated\n")
returned = atomic_replace(tmp, link)
assert link.is_symlink(), "symlink must not be replaced with a regular file"
assert real.read_text(encoding="utf-8") == "updated\n"
assert Path(returned) == real
# Follow the symlink — same content.
assert link.read_text(encoding="utf-8") == "updated\n"
def test_atomic_replace_regular_file(tmp_path: Path) -> None:
target = tmp_path / "plain.yaml"
target.write_text("old\n", encoding="utf-8")
tmp = _write_tmp(tmp_path, "fresh\n")
returned = atomic_replace(tmp, target)
assert Path(returned) == target
assert target.read_text(encoding="utf-8") == "fresh\n"
assert not target.is_symlink()
def test_atomic_replace_first_time_create(tmp_path: Path) -> None:
target = tmp_path / "new.yaml"
assert not target.exists()
tmp = _write_tmp(tmp_path, "brand new\n")
returned = atomic_replace(tmp, target)
assert Path(returned) == target
assert target.read_text(encoding="utf-8") == "brand new\n"
def test_atomic_replace_accepts_pathlike_and_str(tmp_path: Path) -> None:
target = tmp_path / "dual.json"
target.write_text("{}", encoding="utf-8")
# str inputs
tmp1 = _write_tmp(tmp_path, "1")
atomic_replace(str(tmp1), str(target))
assert target.read_text(encoding="utf-8") == "1"
# Path inputs
tmp2 = _write_tmp(tmp_path, "2")
atomic_replace(tmp2, target)
assert target.read_text(encoding="utf-8") == "2"
# ─── atomic_json_write / atomic_yaml_write wiring ──────────────────────────
def test_atomic_json_write_preserves_symlink(tmp_path: Path) -> None:
real = tmp_path / "real.json"
link = tmp_path / "link.json"
real.write_text("{}", encoding="utf-8")
link.symlink_to(real)
atomic_json_write(link, {"hello": "world"})
assert link.is_symlink()
loaded = json.loads(real.read_text(encoding="utf-8"))
assert loaded == {"hello": "world"}
def test_atomic_yaml_write_preserves_symlink(tmp_path: Path) -> None:
real = tmp_path / "real.yaml"
link = tmp_path / "link.yaml"
real.write_text("placeholder: true\n", encoding="utf-8")
link.symlink_to(real)
atomic_yaml_write(link, {"model": {"provider": "openrouter"}})
assert link.is_symlink()
data = yaml.safe_load(real.read_text(encoding="utf-8"))
assert data == {"model": {"provider": "openrouter"}}
def test_atomic_json_write_preserves_symlink_permissions(tmp_path: Path) -> None:
"""Symlinked targets keep the real file's permission bits."""
if os.name != "posix":
pytest.skip("POSIX-only")
real = tmp_path / "real.json"
link = tmp_path / "link.json"
real.write_text("{}", encoding="utf-8")
os.chmod(real, 0o644)
link.symlink_to(real)
atomic_json_write(link, {"x": 1})
import stat as _stat
mode = _stat.S_IMODE(real.stat().st_mode)
assert mode == 0o644, f"permissions drifted after symlinked write: {oct(mode)}"
# ─── Broken-symlink edge case ─────────────────────────────────────────────
def test_atomic_replace_broken_symlink_creates_target(tmp_path: Path) -> None:
"""A symlink pointing at a missing file: the write should create the
real target (resolving via realpath) rather than leaving the dangling
link in place as a regular file.
"""
missing = tmp_path / "does_not_exist_yet.yaml"
link = tmp_path / "link.yaml"
link.symlink_to(missing)
assert link.is_symlink()
assert not missing.exists()
tmp = _write_tmp(tmp_path, "created-through-link\n")
atomic_replace(tmp, link)
assert link.is_symlink(), "symlink must be preserved"
assert missing.exists(), "real target should now exist"
assert missing.read_text(encoding="utf-8") == "created-through-link\n"