hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 6 Packages Projects Releases Wiki Activity
Files
main
hermes-agent/tests/hermes_cli/test_backup.py
Eva c2e2150d78 fix(sessions): close the identity gaps the review found and prove the recovery paths 
- Strip persistence-only fields from a select_context() selection: the request copy was stripped before
  the hook, but an engine handing back the conversation_messages clones put timestamp, _row_id and the
  new ids on the wire (pre-existing for the old fields).
- Every row carries a message_uid whoever wrote it: an AFTER INSERT trigger mints one for a row inserted
  without it (an older build writing into a v31 store); the one-time backfill is gated by its own
  state_meta marker so a store whose schema version cannot advance (no FTS5) never rescans.
- One merge-witness helper (record_absorbed_message) for every fold of a durable message: alternation
  repair's user and assistant merges, the compressor's in-flight restatement onto the carrier, the real
  user anchor folded into a scaffolding turn (the anchor's uid leads), micro-compaction's adjacent-user
  merge.
- tool_call_uids follows tool_calls: it is a payload column, so an assistant merge's unioned map is
  rewritten with the row instead of being overwritten by the stale stored map on adoption.
- A result pairs with the nearest call naming its provider id: each assistant row's ids shadow earlier
  occurrences in the batch, restore and live-list indexes, a call row without a map leaves its result
  unpaired, and nothing pairs across a user turn.
- A composite rewind returns and installs the replacement row's uid on the live scaffold.
- Tests: import_foreign_history, hermes sessions recover and backup import carry the ids; the ACP
  restore test accounts for the uid on restored rows.

(cherry picked from commit b30636a49cb29fdd21ba4100eef9d810897338be)
2026-09-29 04:15:21 +05:30

2698 lines
110 KiB
Python
Raw Permalink Blame History

"""Tests for hermes backup and import commands."""
import json
import os
import socket
import sqlite3
import stat
import struct
import zipfile
from argparse import Namespace
from pathlib import Path
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _no_real_gateway_service(monkeypatch):
"""run_import() auto-installs the gateway service post-restore; tests must
never touch the host's systemd/launchd. Individual tests re-patch these to
assert the wiring."""
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(gateway_mod, "ensure_gateway_service", lambda **kw: False)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
def _advance_backup_clock(seconds: float = 1.1) -> None:
"""Skew hermes_cli.backup's datetime forward instead of sleeping.
Snapshot ids have 1-second resolution; tests that need two distinct
timestamps previously slept >1s. This installs (once) a datetime shim in
the backup module whose now() adds a cumulative offset, then bumps it.
"""
import datetime as _dt
import hermes_cli.backup as _backup
shim = getattr(_backup.datetime, "_hermes_test_shim", None)
if shim is None:
class _ShimDatetime(_dt.datetime):
_hermes_test_shim = True
_offset = _dt.timedelta(0)
@classmethod
def now(cls, tz=None): # noqa: D102
return _dt.datetime.now(tz) + cls._offset
_backup.datetime = _ShimDatetime
shim = _ShimDatetime
else:
shim = _backup.datetime
shim._offset += _dt.timedelta(seconds=seconds)
def _make_hermes_tree(root: Path) -> None:
"""Create a realistic ~/.hermes directory structure for testing."""
(root / "config.yaml").write_text("model:\n provider: openrouter\n")
(root / ".env").write_text("OPENROUTER_API_KEY=sk-test-123\n")
for db_name in ("memory_store.db", "hermes_state.db"):
with sqlite3.connect(root / db_name) as conn:
conn.execute("CREATE TABLE sample (value TEXT)")
conn.execute("INSERT INTO sample VALUES ('test')")
# Sessions
(root / "sessions").mkdir(exist_ok=True)
(root / "sessions" / "abc123.json").write_text("{}")
# Skills
(root / "skills").mkdir(exist_ok=True)
(root / "skills" / "my-skill").mkdir()
(root / "skills" / "my-skill" / "SKILL.md").write_text("# My Skill\n")
# Skins
(root / "skins").mkdir(exist_ok=True)
(root / "skins" / "cyber.yaml").write_text("name: cyber\n")
# Cron
(root / "cron").mkdir(exist_ok=True)
(root / "cron" / "jobs.json").write_text("[]")
# Memories
(root / "memories").mkdir(exist_ok=True)
(root / "memories" / "notes.json").write_text("{}")
# Profiles
(root / "profiles").mkdir(exist_ok=True)
(root / "profiles" / "coder").mkdir()
(root / "profiles" / "coder" / "config.yaml").write_text("model:\n provider: anthropic\n")
(root / "profiles" / "coder" / ".env").write_text("ANTHROPIC_API_KEY=sk-ant-123\n")
# hermes-agent repo (should be EXCLUDED)
(root / "hermes-agent").mkdir(exist_ok=True)
(root / "hermes-agent" / "run_agent.py").write_text("# big file\n")
(root / "hermes-agent" / ".git").mkdir()
(root / "hermes-agent" / ".git" / "HEAD").write_text("ref: refs/heads/main\n")
# __pycache__ (should be EXCLUDED)
(root / "plugins").mkdir(exist_ok=True)
(root / "plugins" / "__pycache__").mkdir()
(root / "plugins" / "__pycache__" / "mod.cpython-312.pyc").write_bytes(b"\x00")
# PID files (should be EXCLUDED)
(root / "gateway.pid").write_text("12345")
# Logs (should be included)
(root / "logs").mkdir(exist_ok=True)
(root / "logs" / "agent.log").write_text("log line\n")
def _symlink_file_or_skip(link: Path, target: Path) -> None:
try:
link.symlink_to(target)
except OSError as exc:
pytest.skip(f"symlinks unavailable in test environment: {exc}")
# ---------------------------------------------------------------------------
# _should_exclude tests
# ---------------------------------------------------------------------------
class TestShouldExclude:
def test_excludes_hermes_agent(self):
from hermes_cli.backup import _should_exclude
assert _should_exclude(Path("hermes-agent/run_agent.py"))
assert _should_exclude(Path("hermes-agent/.git/HEAD"))
def test_excludes_backups_dir(self):
"""backups/ is excluded so pre-update backups don't nest exponentially."""
from hermes_cli.backup import _should_exclude
assert _should_exclude(Path("backups/pre-update-2026-04-27-063400.zip"))
def test_excludes_state_snapshots_dir(self):
"""state-snapshots/ is excluded for the same reason as backups/: every
quick / pre-update snapshot holds its own copy of state.db, so zipping
the tree would ship the DB once per retained snapshot."""
from hermes_cli.backup import _QUICK_SNAPSHOTS_DIR, _should_exclude
assert _should_exclude(Path(_QUICK_SNAPSHOTS_DIR) / "20260814-203829-2026-08-15" / "state.db")
assert _should_exclude(Path(_QUICK_SNAPSHOTS_DIR) / "20260814-203829-2026-08-15" / "manifest.json")
# Named profiles accumulate snapshots too.
assert _should_exclude(Path("profiles/coder") / _QUICK_SNAPSHOTS_DIR / "x" / "state.db")
# The live DB is still backed up.
assert not _should_exclude(Path("state.db"))
def test_excludes_sqlite_sidecars(self):
"""SQLite WAL/SHM/journal sidecars must not ship alongside the
safe-copied .db — pairing a fresh snapshot with stale sidecar state
produces a torn restore."""
from hermes_cli.backup import _should_exclude
assert _should_exclude(Path("state.db-wal"))
assert _should_exclude(Path("state.db-shm"))
assert _should_exclude(Path("state.db-journal"))
assert _should_exclude(Path("memory_store.db-wal"))
# The .db itself is still included (and safe-copied separately)
assert not _should_exclude(Path("state.db"))
def test_excludes_managed_runtime_trees_at_root(self):
"""models/, runtimes/, and node/ at a profile-home root hold
re-downloadable GGUF weights and runtime binaries that reach
hundreds of GB — zipping them is the 20-minute-hang symptom."""
from hermes_cli.backup import _should_exclude
assert _should_exclude(Path("models/Qwen3.6-27B-Q4_K_M.gguf"))
assert _should_exclude(Path("models/assets/mmproj.gguf"))
assert _should_exclude(Path("runtimes/llamacpp/b10362/cuda/ggml-cuda.dll"))
assert _should_exclude(Path("node/node.exe"))
# Named profiles download their own copies.
assert _should_exclude(Path("profiles/clean/models/big.gguf"))
assert _should_exclude(Path("profiles/clean/runtimes/llamacpp/x.dll"))
def test_excludes_regenerable_cache_but_keeps_durable_artifacts(self):
"""Catalogs and live browser profiles are rebuilt on demand; delivered media and the
citation ledger are not, so they stay in the archive."""
from hermes_cli.backup import _should_exclude
assert _should_exclude(Path("cache/model_catalog.json"))
assert _should_exclude(Path("cache/chrome-debug/Default/Cookies"))
assert _should_exclude(Path("profiles/sage/cache/chrome-debug/cache.db"))
assert not _should_exclude(Path("cache/images/x.png"))
assert not _should_exclude(Path("profiles/sage/cache/citations/ledger.json"))
assert not _should_exclude(Path("skills/example/cache/notes.md"))
def test_keeps_nested_dirs_named_like_runtime_trees(self):
"""A deeper directory that happens to be called models/ or node/ is
user data (a skill's assets, project files) and must survive."""
from hermes_cli.backup import _should_exclude
assert not _should_exclude(Path("skills/mlops/models/notes.md"))
assert not _should_exclude(Path("scratch/node/index.js"))
assert not _should_exclude(Path("profiles/clean/skills/x/models/a.txt"))
def test_excludes_desktop_emergency_state_db_baks(self):
"""The desktop updater's pre-flight drops timestamped
state.db.pre-update-emergency-*.bak files at the HERMES_HOME root —
backup artifacts in the same class as backups/, so a full backup
must not re-ship them."""
from hermes_cli.backup import _should_exclude
assert _should_exclude(
Path("state.db.pre-update-emergency-2026-08-15T04-55-33-619Z.bak")
)
assert _should_exclude(
Path("profiles/coder/state.db.pre-update-emergency-2026-08-15T04-55-33-619Z.bak")
)
# Other .bak files are user data and stay.
assert not _should_exclude(Path("config.yaml.bak"))
# ---------------------------------------------------------------------------
# _iter_backup_files tests
# ---------------------------------------------------------------------------
class TestIterBackupFiles:
def test_manual_and_automatic_paths_share_one_walk(self, tmp_path):
"""Both backup entry points must select the identical file set.
Before the walks were unified, the automatic pre-update path pruned
``hermes-agent`` at ANY depth, silently dropping nested skill dirs
like ``skills/autonomous-ai-agents/hermes-agent/`` that the manual
path preserved. One shared iterator makes that drift impossible;
this test pins the contract."""
from hermes_cli.backup import _iter_backup_files
root = tmp_path / ".hermes"
root.mkdir()
_make_hermes_tree(root)
# The case the old automatic walk got wrong: a nested dir named
# hermes-agent holding real skill content.
nested = root / "skills" / "autonomous-ai-agents" / "hermes-agent"
nested.mkdir(parents=True)
(nested / "SKILL.md").write_text("# nested skill\n")
# A root-level managed runtime tree that both paths must prune.
(root / "models").mkdir()
(root / "models" / "big.gguf").write_bytes(b"\x00" * 64)
out_path = tmp_path / "out.zip"
selected = {str(rel) for _, rel in _iter_backup_files(root, out_path)}
rel_nested = str(Path("skills/autonomous-ai-agents/hermes-agent/SKILL.md"))
assert rel_nested in selected
assert str(Path("models/big.gguf")) not in selected
assert not any(s.startswith("hermes-agent") for s in selected)
def test_prunes_browser_use_cli_profiles_at_home_roots_only(self, tmp_path):
"""The Browser Use CLI backend writes ``HERMES_HOME/browser_profiles/`` (underscore) — a
live Chromium user-data dir holding Login Data / Cookies. It must never enter an archive,
at the root or under ``profiles/<name>/``; a skill's same-named dir is user data (#117346)."""
from hermes_cli.backup import _iter_backup_files
root = tmp_path / ".hermes"
root.mkdir()
files = {
"browser_profiles/browser-use-default/Default/Login Data": False,
"browser_profiles/browser-use-default/Default/Network/Cookies": False,
"profiles/coder/browser_profiles/browser-use-default/Default/Cookies": False,
"skills/example/browser_profiles/notes.md": True,
}
for rel in files:
f = root / rel
f.parent.mkdir(parents=True, exist_ok=True)
f.write_text("x")
selected = {str(rel) for _, rel in _iter_backup_files(root, tmp_path / "out.zip")}
assert {rel for rel in files if str(Path(rel)) in selected} == {rel for rel, keep in files.items() if keep}
def test_prunes_regenerable_caches_but_keeps_durable_and_nested(self, tmp_path):
from hermes_cli.backup import _iter_backup_files
root = tmp_path / ".hermes"
root.mkdir()
files = {
"cache/model_catalog.json": False,
"cache/chrome-debug/Default/Cookies": False,
"profiles/sage/cache/chrome-debug/cache.db": False,
"cache/images/x.png": True,
"cache/citations/ledger.json": True,
"profiles/sage/cache/images/y.png": True,
"skills/example/cache/state.db": True,
}
for rel in files:
(root / rel).parent.mkdir(parents=True, exist_ok=True)
(root / rel).write_bytes(b"x")
skipped: set = set()
selected = {str(rel) for _, rel in _iter_backup_files(root, tmp_path / "out.zip", skipped)}
assert {rel for rel, keep in files.items() if keep} == {s.replace(os.sep, "/") for s in selected}
assert str(Path("cache/chrome-debug")) in skipped
assert str(Path("profiles/sage/cache/chrome-debug")) in skipped
assert "cache" not in skipped
def test_skipped_dirs_collected_for_summary(self, tmp_path):
from hermes_cli.backup import _iter_backup_files
root = tmp_path / ".hermes"
root.mkdir()
_make_hermes_tree(root)
(root / "models").mkdir()
(root / "models" / "big.gguf").write_bytes(b"\x00")
skipped: set = set()
list(_iter_backup_files(root, tmp_path / "out.zip", skipped))
assert "models" in skipped
assert "hermes-agent" in skipped
@pytest.mark.platforms("linux")
def test_skips_unix_sockets(self, tmp_path, monkeypatch):
from hermes_cli.backup import _iter_backup_files
root = tmp_path / ".hermes"
root.mkdir()
# AF_UNIX paths are capped at ~108 bytes; pytest's tmp_path overflows that under the
# test runner's deep temp root, so bind by a relative name from inside ``root``.
monkeypatch.chdir(root)
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as gateway_socket:
gateway_socket.bind("gateway.sock")
selected = {str(rel) for _, rel in _iter_backup_files(root, tmp_path / "out.zip")}
assert "gateway.sock" not in selected
# ---------------------------------------------------------------------------
# Backup tests
# ---------------------------------------------------------------------------
class TestBackup:
def test_db_snapshots_staged_beside_output_zip(self, tmp_path, monkeypatch):
"""SQLite staging temp files must be created on the output zip's
filesystem (dir=out_path.parent), NOT the system /tmp default — a
small tmpfs there silently drops large DBs from the backup (#35376)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out_dir = tmp_path / "external-drive"
out_dir.mkdir()
out_zip = out_dir / "backup.zip"
args = Namespace(output=str(out_zip))
import hermes_cli.backup as backup_mod
staged_dirs = []
real_ntf = backup_mod.tempfile.NamedTemporaryFile
def _spy(*a, **kw):
staged_dirs.append(kw.get("dir"))
return real_ntf(*a, **kw)
monkeypatch.setattr(backup_mod.tempfile, "NamedTemporaryFile", _spy)
backup_mod.run_backup(args)
# At least one .db was staged, and every staging call targeted the
# output zip's directory rather than the system temp default.
assert staged_dirs, "no SQLite snapshot was staged"
assert all(d == str(out_dir) for d in staged_dirs), staged_dirs
def test_pre_update_db_snapshots_staged_beside_output_zip(self, tmp_path, monkeypatch):
"""The pre-update/pre-migration zip path (_write_full_zip_backup) must
also stage SQLite snapshots beside its output zip, not in /tmp."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out_zip = hermes_home / "backups" / "pre-update-test.zip"
out_zip.parent.mkdir(parents=True, exist_ok=True)
import hermes_cli.backup as backup_mod
staged_dirs = []
real_ntf = backup_mod.tempfile.NamedTemporaryFile
def _spy(*a, **kw):
staged_dirs.append(kw.get("dir"))
return real_ntf(*a, **kw)
monkeypatch.setattr(backup_mod.tempfile, "NamedTemporaryFile", _spy)
result = backup_mod._write_full_zip_backup(out_zip, hermes_home)
assert result is not None
assert staged_dirs, "no SQLite snapshot was staged"
assert all(d == str(out_zip.parent) for d in staged_dirs), staged_dirs
def test_skips_symlinked_files(self, tmp_path, monkeypatch):
"""Backup must not dereference symlinks and leak files outside HERMES_HOME."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
outside = tmp_path / "outside-secret.txt"
outside.write_text("outside secret\n")
_symlink_file_or_skip(hermes_home / "skills" / "outside-link.txt", outside)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out_zip = tmp_path / "backup.zip"
args = Namespace(output=str(out_zip))
from hermes_cli.backup import run_backup
run_backup(args)
with zipfile.ZipFile(out_zip, "r") as zf:
names = zf.namelist()
assert "skills/outside-link.txt" not in names
assert all(zf.read(name) != b"outside secret\n" for name in names)
def test_state_snapshots_not_nested_into_backup(self, tmp_path, monkeypatch):
"""A quick snapshot left under state-snapshots/ must not be re-shipped
by the full backup — each snapshot already holds a copy of state.db, so
nesting them multiplies the archive by (1 + retained snapshots)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
with sqlite3.connect(hermes_home / "state.db") as conn:
conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)")
conn.execute("INSERT INTO sessions VALUES ('s1')")
from hermes_cli.backup import _QUICK_SNAPSHOTS_DIR, create_quick_snapshot, run_backup
# Real producer, so the layout under state-snapshots/ is whatever the
# code actually writes (manifest.json + state.db copy + ...).
snap_id = create_quick_snapshot(hermes_home=hermes_home)
assert snap_id and (hermes_home / _QUICK_SNAPSHOTS_DIR / snap_id / "state.db").exists()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out_zip = tmp_path / "backup.zip"
run_backup(Namespace(output=str(out_zip)))
with zipfile.ZipFile(out_zip, "r") as zf:
names = zf.namelist()
assert not any(n.startswith(_QUICK_SNAPSHOTS_DIR + "/") for n in names), names
# Exactly one state.db in the archive: the live one.
assert [n for n in names if n == "state.db" or n.endswith("/state.db")] == ["state.db"]
# ---------------------------------------------------------------------------
# _validate_backup_zip tests
# ---------------------------------------------------------------------------
class TestValidateBackupZip:
def _make_zip(self, zip_path: Path, filenames: list[str]) -> None:
with zipfile.ZipFile(zip_path, "w") as zf:
for name in filenames:
zf.writestr(name, "dummy")
def test_state_db_passes(self, tmp_path):
"""A zip containing state.db is accepted as a valid Hermes backup."""
from hermes_cli.backup import _validate_backup_zip
zip_path = tmp_path / "backup.zip"
self._make_zip(zip_path, ["state.db", "sessions/abc.json"])
with zipfile.ZipFile(zip_path, "r") as zf:
ok, reason = _validate_backup_zip(zf)
assert ok, reason
# ---------------------------------------------------------------------------
# Import tests
# ---------------------------------------------------------------------------
class TestImport:
def _make_backup_zip(self, zip_path: Path, files: dict[str, str | bytes]) -> None:
"""Create a test zip with given files."""
with zipfile.ZipFile(zip_path, "w") as zf:
for name, content in files.items():
if isinstance(content, bytes):
zf.writestr(name, content)
else:
zf.writestr(name, content)
def test_import_auto_installs_gateway_service(self, tmp_path, monkeypatch):
"""After a restore, run_import brings the gateway service up without
prompting — restored cron jobs and bot tokens must not sit dormant
(the install-then-import dead-gateway bug)."""
import hermes_cli.gateway as gateway_mod
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
calls = []
monkeypatch.setattr(
gateway_mod, "ensure_gateway_service",
lambda **kw: calls.append(kw) or True,
)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {"config.yaml": "model: test\n"})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert calls and calls[0].get("context") == "import"
def test_import_skips_service_when_already_running(self, tmp_path, monkeypatch):
"""A live gateway is left alone — no reinstall churn during import."""
import hermes_cli.gateway as gateway_mod
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
calls = []
monkeypatch.setattr(
gateway_mod, "ensure_gateway_service",
lambda **kw: calls.append(kw) or True,
)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: True)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {"config.yaml": "model: test\n"})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert not calls
def test_import_survives_service_layer_import_failure(self, tmp_path, monkeypatch, capsys):
"""If the service helpers can't even be reached, import still completes
and prints the manual fallback."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
import hermes_cli.gateway as gateway_mod
def boom():
raise RuntimeError("service layer unavailable")
monkeypatch.setattr(gateway_mod, "_is_service_running", boom)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {"config.yaml": "model: test\n"})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert (hermes_home / "config.yaml").read_text() == "model: test\n"
@staticmethod
def _corrupt_member(zip_path: Path, name: str, damage: str) -> None:
"""Damage *name*'s data in place, leaving ``is_zipfile``/``namelist`` happy.
``deflate``: reserved block type 3 (``zlib.error``); ``stored``: one payload byte
flipped (``Bad CRC-32``); ``bzip2``: block CRC flipped (``OSError`` Invalid data
stream); ``truncated``: the central directory's compressed size halved, so the
deflate stream is cut short (``Bad CRC-32``). A real archive cannot make the read hit
physical EOF (``EOFError``): zipfile refuses overlapping entries first, so that case
-- media going bad under the archive -- is covered by the rot-after-pre-flight test.
"""
with zipfile.ZipFile(zip_path) as zf:
info = zf.getinfo(name)
data_offset = info.header_offset + 30 + len(info.filename) + len(info.extra)
raw = bytearray(zip_path.read_bytes())
if damage == "truncated":
# Central record = local header fields shifted by 2 ("version made by"); the CRC keeps
# the match unique. compress_size sits at +20 in the central record.
local = raw[info.header_offset + 4:info.header_offset + 26]
entry = raw.index(local, info.header_offset + 26) - 6
struct.pack_into("<I", raw, entry + 20, info.compress_size // 2)
elif damage == "deflate":
raw[data_offset] |= 0b110
else:
raw[data_offset + (10 if damage == "bzip2" else 0)] ^= 0xFF
zip_path.write_bytes(bytes(raw))
def _home_for_corrupt_import(self, tmp_path, monkeypatch) -> Path:
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text("model: live\n")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
return hermes_home
@pytest.mark.parametrize("damage, compression", [
("deflate", zipfile.ZIP_DEFLATED),
("stored", zipfile.ZIP_STORED),
("bzip2", zipfile.ZIP_BZIP2),
("truncated", zipfile.ZIP_DEFLATED),
])
def test_import_refuses_damaged_archive_before_touching_home(
self, tmp_path, monkeypatch, capsys, damage, compression
):
"""#121258: members that fail to decompress, fail their CRC or are cut short must all be
found BEFORE the first file is replaced -- not surface as a traceback half-way through
-- listed (capped) in one run, and the command must exit 1 without "restored" wording."""
hermes_home = self._home_for_corrupt_import(tmp_path, monkeypatch)
zip_path = tmp_path / "backup.zip"
names = [f"skills/s{i}/SKILL.md" for i in range(12)]
with zipfile.ZipFile(zip_path, "w", compression=compression) as zf:
zf.writestr("config.yaml", "model: restored\n")
for name in names:
zf.writestr(name, f"# {name}\n" * 20)
zf.writestr("sessions/after.json", '{"restored": true}')
for name in names:
self._corrupt_member(zip_path, name, damage)
assert zipfile.is_zipfile(zip_path)
from hermes_cli.backup import run_import
from hermes_cli.main import cmd_import
assert run_import(Namespace(zipfile=str(zip_path), force=True)) == 1
out = capsys.readouterr().out
assert "backup archive is damaged (12 member(s)" in out
assert sum(1 for line in out.splitlines() if line.strip().startswith("skills/s")) == 10
assert "... and 2 more" in out
assert "Importing" not in out
assert "restored" not in out.replace("nothing was restored", "")
# The home is exactly as it was.
assert (hermes_home / "config.yaml").read_text() == "model: live\n"
assert sorted(p.name for p in hermes_home.iterdir()) == ["config.yaml"]
assert cmd_import(Namespace(zipfile=str(zip_path), force=True)) == 1
def test_import_ignores_damaged_pm_runtime_but_keeps_portable_data(
self, tmp_path, monkeypatch
):
"""Machine-local interpreter/dependency state is not a reason to refuse a backup."""
hermes_home = self._home_for_corrupt_import(tmp_path, monkeypatch)
zip_path = tmp_path / "backup.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("config.yaml", "model: restored\n")
zf.writestr("profiles/coder/installs/python.zip", "machine-specific\n" * 20)
zf.writestr("skills/demo/SKILL.md", "# portable\n")
self._corrupt_member(zip_path, "profiles/coder/installs/python.zip", "deflate")
from hermes_cli.backup import run_import
assert run_import(Namespace(zipfile=str(zip_path), force=True)) is None
assert (hermes_home / "config.yaml").read_text() == "model: restored\n"
assert (hermes_home / "skills/demo/SKILL.md").read_text() == "# portable\n"
assert not (hermes_home / "profiles/coder/installs").exists()
def test_import_skips_member_that_rots_after_preflight_and_reports_incomplete(
self, tmp_path, monkeypatch, capsys
):
"""A member that passes the integrity pass but fails while being written (media going
bad under the archive) warns, leaves the damaged target absent, restores the rest and
exits 1 -- rather than escaping as a traceback. A damaged member the restore skips
anyway (runtime ``gateway.pid``) does not make the pre-flight refuse the archive."""
hermes_home = self._home_for_corrupt_import(tmp_path, monkeypatch)
zip_path = tmp_path / "backup.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("config.yaml", "model: restored\n")
zf.writestr("damaged.json", '{"will": "fail"}')
zf.writestr("gateway.pid", "12345\n" * 20)
zf.writestr("sessions/after.json", '{"restored": true}')
self._corrupt_member(zip_path, "gateway.pid", "deflate")
import hermes_cli.backup as backup_mod
real_extract = backup_mod._extract_member_atomically
def flaky_extract(zf, member, target, new_file_mode=None):
if member == "damaged.json":
raise EOFError("Compressed file ended before the end-of-stream marker was reached")
return real_extract(zf, member, target, new_file_mode)
monkeypatch.setattr(backup_mod, "_extract_member_atomically", flaky_extract)
assert backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) == 1
out = capsys.readouterr().out
assert "damaged (" not in out
assert (hermes_home / "config.yaml").read_text() == "model: restored\n"
assert not (hermes_home / "damaged.json").exists()
assert not (hermes_home / "gateway.pid").exists()
assert not list(hermes_home.glob(".damaged.json.*.partial"))
assert (hermes_home / "sessions" / "after.json").read_text() == '{"restored": true}'
assert "Warnings (1 files skipped):" in out
assert "Import incomplete" in out
assert "has been restored" not in out
def test_skipped_member_reports_incomplete_and_exits_1(self, tmp_path, monkeypatch, capsys):
"""A member that could not be written is a partial restore: the CLI must not print
"restored" or exit 0, or a script and the dashboard's "done" badge carry on over it.
Runtime files the import deliberately keeps are not failures."""
hermes_home = tmp_path / ".hermes"
locked = hermes_home / "skills" / "demo"
locked.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {
"config.yaml": "model: test\n",
"gateway.pid": "123\n",
"skills/demo/SKILL.md": "# demo\n",
})
args = Namespace(zipfile=str(zip_path), force=True)
from hermes_cli.backup import run_import
from hermes_cli.main import cmd_import
locked.chmod(0o555)
try:
if os.access(locked, os.W_OK):
pytest.skip("root or Windows: chmod 0555 does not make the directory read-only")
assert run_import(args) == 1
out = capsys.readouterr().out
assert "Import incomplete" in out
assert "Import complete" not in out
assert "skills/demo/SKILL.md" in out
assert "has been restored" not in out
assert cmd_import(args) == 1
finally:
locked.chmod(0o755)
capsys.readouterr()
assert cmd_import(args) is None
out = capsys.readouterr().out
assert "Preserved 1 runtime state file(s)" in out
assert "Done. Your Hermes configuration has been restored." in out
def test_preserves_per_profile_gateway_state(self, tmp_path, monkeypatch):
"""The skip is matched by basename, so a named profile's
gateway_state.json (profiles/<name>/gateway_state.json) is preserved
the same way the root profile's is."""
hermes_home = tmp_path / ".hermes"
(hermes_home / "profiles" / "coder").mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
live_state = '{"gateway_state": "running"}'
(hermes_home / "profiles" / "coder" / "gateway_state.json").write_text(live_state)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {
"config.yaml": "model: test\n",
"profiles/coder/config.yaml": "model: anthropic\n",
"profiles/coder/gateway_state.json": '{"gateway_state": "stopped"}',
})
args = Namespace(zipfile=str(zip_path), force=True)
from hermes_cli.backup import run_import
run_import(args)
# Profile config is restored, but its live gateway state is preserved.
assert (hermes_home / "profiles" / "coder" / "config.yaml").read_text() == "model: anthropic\n"
assert (
hermes_home / "profiles" / "coder" / "gateway_state.json"
).read_text() == live_state
def test_preserves_runtime_pid_and_process_files(self, tmp_path, monkeypatch):
"""gateway.pid / cron.pid / gateway.lock / processes.json from a backup
reference the source machine's process namespace and must never be
written over the target's."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
# Live runtime files belonging to the target's own processes.
(hermes_home / "gateway.pid").write_text("4242")
(hermes_home / "processes.json").write_text('{"live": true}')
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {
"config.yaml": "model: test\n",
"gateway.pid": "9999",
"cron.pid": "8888",
"gateway.lock": "7777",
"processes.json": '{"stale": true}',
})
args = Namespace(zipfile=str(zip_path), force=True)
from hermes_cli.backup import run_import
run_import(args)
# Live runtime files are untouched; the backup's foreign ones never land.
assert (hermes_home / "gateway.pid").read_text() == "4242"
assert (hermes_home / "processes.json").read_text() == '{"live": true}'
# cron.pid / gateway.lock had no live copy and were not seeded.
assert not (hermes_home / "cron.pid").exists()
assert not (hermes_home / "gateway.lock").exists()
@pytest.mark.skipif(os.name != "posix", reason="POSIX file permissions only")
def test_restores_secret_files_with_0600_perms(self, tmp_path, monkeypatch):
"""Secret files must end up at 0600 after restore (zipfile drops mode bits)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {
"config.yaml": "model: openrouter\n",
".env": "OPENROUTER_API_KEY=sk-secret\n",
"auth.json": '{"providers": {"nous": "token"}}',
"state.db": b"SQLite format 3\x00",
"profiles/coder/.env": "ANTHROPIC_API_KEY=sk-ant-secret\n",
})
args = Namespace(zipfile=str(zip_path), force=True)
from hermes_cli.backup import run_import
run_import(args)
for rel in (".env", "auth.json", "state.db", "profiles/coder/.env"):
mode = (hermes_home / rel).stat().st_mode & 0o777
assert mode == 0o600, f"{rel} restored with mode {oct(mode)}, expected 0o600"
# ---------------------------------------------------------------------------
# Round-trip test
# ---------------------------------------------------------------------------
class TestRoundTrip:
def test_backup_then_import(self, tmp_path, monkeypatch):
"""Full round-trip: backup -> import to a new location -> verify."""
# Source
src_home = tmp_path / "source" / ".hermes"
src_home.mkdir(parents=True)
_make_hermes_tree(src_home)
monkeypatch.setenv("HERMES_HOME", str(src_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path / "source")
# Backup
out_zip = tmp_path / "roundtrip.zip"
from hermes_cli.backup import run_backup, run_import
run_backup(Namespace(output=str(out_zip)))
assert out_zip.exists()
# Import into a different location
dst_home = tmp_path / "dest" / ".hermes"
dst_home.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(dst_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path / "dest")
run_import(Namespace(zipfile=str(out_zip), force=True))
# Verify key files
assert (dst_home / "config.yaml").read_text() == "model:\n provider: openrouter\n"
assert (dst_home / ".env").read_text() == "OPENROUTER_API_KEY=sk-test-123\n"
assert (dst_home / "skills" / "my-skill" / "SKILL.md").exists()
assert (dst_home / "profiles" / "coder" / "config.yaml").exists()
assert (dst_home / "sessions" / "abc123.json").exists()
assert (dst_home / "logs" / "agent.log").exists()
# hermes-agent should NOT be present
assert not (dst_home / "hermes-agent").exists()
# __pycache__ should NOT be present
assert not (dst_home / "plugins" / "__pycache__").exists()
# PID files should NOT be present
assert not (dst_home / "gateway.pid").exists()
# ---------------------------------------------------------------------------
# Validate / detect-prefix unit tests
# ---------------------------------------------------------------------------
class TestValidation:
def test_validate_with_config(self):
"""Zip with config.yaml passes validation."""
import io
from hermes_cli.backup import _validate_backup_zip
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("config.yaml", "test")
buf.seek(0)
with zipfile.ZipFile(buf, "r") as zf:
ok, reason = _validate_backup_zip(zf)
assert ok
def test_detect_prefix_only_dirs(self):
"""Prefix detection returns empty for zip with only directory entries."""
import io
from hermes_cli.backup import _detect_prefix
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
# Only directory entries (trailing slash)
zf.writestr(".hermes/", "")
zf.writestr(".hermes/skills/", "")
buf.seek(0)
with zipfile.ZipFile(buf, "r") as zf:
assert _detect_prefix(zf) == ""
# ---------------------------------------------------------------------------
# Edge case tests for uncovered paths
# ---------------------------------------------------------------------------
class TestBackupEdgeCases:
def test_incomplete_archive_is_kept_but_reported_as_failure(self, tmp_path, monkeypatch, capsys):
"""A file that cannot be read is skipped, the zip still lands, and the CLI exits 1: a
cron/systemd timer must never see a partial archive as success (#101096)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
unreadable = hermes_home / "skills" / "locked.md"
unreadable.write_text("secret\n")
unreadable.chmod(0)
if os.access(unreadable, os.R_OK):
pytest.skip("running as root: chmod 0 does not make the file unreadable")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
from hermes_cli.backup import _RUN_BACKUP_PREFIX, run_backup
from hermes_cli.main import cmd_backup
out_dir = tmp_path / "b"
out_dir.mkdir()
good_old = out_dir / f"{_RUN_BACKUP_PREFIX}old.zip"
good_old.write_bytes(b"PK")
out_zip = out_dir / f"{_RUN_BACKUP_PREFIX}new.zip"
assert run_backup(Namespace(output=str(out_zip), keep=1)) is False
assert out_zip.exists()
assert good_old.exists(), "an incomplete run must not rotate the last complete backup out"
with pytest.raises(SystemExit) as exc:
cmd_backup(Namespace(output=str(tmp_path / "out2.zip"), quick=False))
assert exc.value.code == 1
unreadable.chmod(0o600)
assert run_backup(Namespace(output=str(tmp_path / "out4.zip"))) is True
def test_empty_hermes_home(self, tmp_path, monkeypatch):
"""Backup handles empty hermes home (no files to back up)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
# Only excluded dirs, no actual files
(hermes_home / "__pycache__").mkdir()
(hermes_home / "__pycache__" / "foo.pyc").write_bytes(b"\x00")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
args = Namespace(output=str(tmp_path / "out.zip"))
from hermes_cli.backup import run_backup
run_backup(args)
# No zip should be created
assert not (tmp_path / "out.zip").exists()
def test_pre1980_timestamp_skipped(self, tmp_path, monkeypatch):
"""Backup skips files with pre-1980 timestamps (ZIP limitation)."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text("model: test\n")
# Create a file with epoch timestamp (1970-01-01)
old_file = hermes_home / "ancient.txt"
old_file.write_text("old data")
os.utime(old_file, (0, 0))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
out_zip = tmp_path / "out.zip"
args = Namespace(output=str(out_zip))
from hermes_cli.backup import run_backup
run_backup(args)
# Zip should still be created with the valid files
assert out_zip.exists()
with zipfile.ZipFile(out_zip, "r") as zf:
names = zf.namelist()
assert "config.yaml" in names
# The pre-1980 file should be skipped, not crash the backup
assert "ancient.txt" not in names
class TestImportEdgeCases:
def _make_backup_zip(self, zip_path: Path, files: dict[str, str | bytes]) -> None:
with zipfile.ZipFile(zip_path, "w") as zf:
for name, content in files.items():
zf.writestr(name, content)
def test_eof_during_confirmation(self, tmp_path, monkeypatch):
"""Import handles EOFError during confirmation prompt."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text("existing\n")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {"config.yaml": "new\n"})
args = Namespace(zipfile=str(zip_path), force=False)
from hermes_cli.backup import run_import
with patch("builtins.input", side_effect=EOFError):
with pytest.raises(SystemExit):
run_import(args)
class _ExplodingMember:
"""Zip member whose stream dies mid-restore (ENOSPC / corrupt member).
Both the pre-fix ``dst.write(src.read())`` and the atomic
``shutil.copyfileobj`` path pull bytes through ``read()``, so injecting
here exercises whichever implementation is in the tree.
"""
def __enter__(self):
return self
def __exit__(self, *exc_info):
return False
def read(self, *args):
raise OSError(28, "No space left on device")
def close(self):
pass
def _break_member(monkeypatch, failing_member: str) -> None:
"""Make ``ZipFile.open`` hand back a dying stream for one member only.
The first open of that member (run_import's integrity pre-flight, which reads every
member before anything is written) is served for real; the restore's own open dies.
"""
real_open = zipfile.ZipFile.open
seen: list[str] = []
def _patched(self, name, *args, **kwargs):
filename = name.filename if isinstance(name, zipfile.ZipInfo) else name
if filename == failing_member:
seen.append(filename)
if len(seen) > 1:
return _ExplodingMember()
return real_open(self, name, *args, **kwargs)
monkeypatch.setattr(zipfile.ZipFile, "open", _patched)
class TestImportAtomicWrites:
"""`hermes import` must never leave a user's file truncated.
The pre-fix code did ``open(target, "wb")`` then ``dst.write(src.read())``,
which zeroes the existing file *before* any replacement bytes exist. These
tests pin the invariant for both restore branches: the HERMES_HOME branch
and the ``_external/`` branch that writes into third-party configs under
the user's home.
"""
def _zip(self, zip_path: Path, files: dict) -> None:
with zipfile.ZipFile(zip_path, "w") as zf:
for name, content in files.items():
zf.writestr(name, content)
def test_failed_member_leaves_existing_file_intact(self, tmp_path, monkeypatch):
"""A dying member must not destroy the file it was replacing."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
original = "model: original\napi_key: keep-me\n"
(hermes_home / "config.yaml").write_text(original)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {"config.yaml": "model: replacement\n", "state.db": ""})
_break_member(monkeypatch, "config.yaml")
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
# Pre-fix this file is 0 bytes: the truncate landed, the write did not.
assert (hermes_home / "config.yaml").read_text() == original
# And the aborted write must not litter the directory it staged in.
assert list(hermes_home.glob(".config.yaml.*")) == []
def test_failed_external_member_leaves_existing_file_intact(self, tmp_path, monkeypatch):
"""Same invariant on the `_external/` branch, which writes outside HERMES_HOME."""
dst_home = tmp_path / "dst"
dst_home.mkdir()
hermes_home = dst_home / ".hermes"
hermes_home.mkdir()
honcho = dst_home / ".honcho"
honcho.mkdir()
original = '{"peer":"original"}'
(honcho / "config.json").write_text(original)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {
"config.yaml": "model: {}\n",
"_external/.honcho/config.json": '{"peer":"replacement"}',
})
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: dst_home)
_break_member(monkeypatch, "_external/.honcho/config.json")
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert (honcho / "config.json").read_text() == original
assert list(honcho.glob(".config.json.*")) == []
@pytest.mark.skipif(os.name != "posix", reason="POSIX symlinks")
def test_symlinked_target_keeps_its_symlink(self, tmp_path, monkeypatch):
"""A symlinked target is written through, not replaced by a regular file.
Guards the atomic rewrite against a naive ``os.replace``, which would
detach dotfiles-managed deployments (GitHub #16743). ``atomic_replace``
resolves the link first.
"""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
store = hermes_home / "store"
store.mkdir()
real = store / "config.yaml"
real.write_text("model: original\n")
link = hermes_home / "config.yaml"
link.symlink_to(real)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {"config.yaml": "model: restored\n", "state.db": ""})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert link.is_symlink(), "import replaced the symlink with a regular file"
assert real.read_text() == "model: restored\n"
@pytest.mark.skipif(os.name != "posix", reason="POSIX symlinks")
def test_symlinked_external_target_keeps_its_symlink(self, tmp_path, monkeypatch):
"""Same guard on the `_external/` branch — the realistic dotfiles case."""
dst_home = tmp_path / "dst"
dst_home.mkdir()
hermes_home = dst_home / ".hermes"
hermes_home.mkdir()
dotfiles = dst_home / "dotfiles"
dotfiles.mkdir()
real = dotfiles / "honcho.json"
real.write_text('{"peer":"original"}')
honcho = dst_home / ".honcho"
honcho.mkdir()
link = honcho / "config.json"
link.symlink_to(real)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: dst_home)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {
"config.yaml": "model: {}\n",
"_external/.honcho/config.json": '{"peer":"restored"}',
})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert link.is_symlink(), "import replaced the symlink with a regular file"
assert real.read_text() == '{"peer":"restored"}'
@pytest.mark.skipif(os.name != "posix", reason="POSIX file modes")
def test_restore_preserves_existing_file_mode(self, tmp_path, monkeypatch):
"""Staging through mkstemp must not silently tighten restored files to 0600.
``tempfile.mkstemp`` creates at 0600; the mode of the file being
replaced has to survive the publish, or Docker/NAS installs that rely
on broader permissions break on restore.
"""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
target = hermes_home / "config.yaml"
target.write_text("model: original\n")
os.chmod(target, 0o644)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {"config.yaml": "model: restored\n", "state.db": ""})
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert target.read_text() == "model: restored\n"
assert (target.stat().st_mode & 0o777) == 0o644
@pytest.mark.platforms("posix")
def test_restore_preserves_existing_file_owner(self, tmp_path, monkeypatch):
"""A root-run import must not re-own the user's files to root.
``os.replace`` swaps in a temp file owned by the *writing* user, so a
``sudo hermes import`` onto a user-owned (or Docker/NAS volume-owned)
HERMES_HOME would hand every restored file to root. The uid/gid is
forced so the assertion does not require running as root.
"""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
target = hermes_home / "config.yaml"
target.write_text("model: original\n")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {"config.yaml": "model: restored\n", "state.db": ""})
chown_calls: list[tuple[Path, int, int]] = []
monkeypatch.setattr(
"hermes_cli.backup_restore._preserve_file_owner",
lambda p: (123, 456) if Path(p).exists() else None,
)
monkeypatch.setattr(
"utils.os.chown",
lambda path, uid, gid: chown_calls.append((Path(path), uid, gid)),
)
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert target.read_text() == "model: restored\n"
# config.yaml pre-existed, so its owner is captured and re-applied;
# state.db is newly created, so there is no prior owner to restore.
assert chown_calls == [(target, 123, 456)]
def test_mode_is_applied_before_the_replace(self, tmp_path, monkeypatch):
"""Publish the correct native permission bits with no permissive window."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
target = hermes_home / "config.yaml"
target.write_text("model: original\n")
os.chmod(target, 0o644)
expected_mode = target.stat().st_mode & 0o777
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(zip_path, {"config.yaml": "model: restored\n", "state.db": ""})
import hermes_cli.backup_restore as backup_mod
real_replace = backup_mod.atomic_replace
staged_modes: list[int] = []
def spying_replace(tmp, dst):
if Path(dst).name == "config.yaml":
staged_modes.append(os.stat(tmp).st_mode & 0o777)
return real_replace(tmp, dst)
monkeypatch.setattr(backup_mod, "atomic_replace", spying_replace)
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
assert staged_modes == [expected_mode]
assert (target.stat().st_mode & 0o777) == expected_mode
@pytest.mark.platforms("posix")
def test_restore_does_not_carry_setuid_onto_archive_content(
self, tmp_path, monkeypatch
):
"""An imported member must not inherit a privileged target's identity.
``_preserve_file_mode`` returns ``stat.S_IMODE``, i.e. all twelve bits,
so a target sitting at 0o6755 hands setuid/setgid straight back to a
file whose contents now come from the zip. Whoever produced the
archive would then get whatever that file executes as. The other
``utils`` writers can preserve the full mode safely because they
re-serialize content this process produced; ``hermes import`` is the
one write path where the bytes are untrusted, and it is also the path
that documents ``sudo`` use for owner preservation.
The sibling assertions in this class mask with ``& 0o777``, which
discards exactly the bits at issue, so this failure is invisible to
them.
"""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
target = hermes_home / "helper.sh"
target.write_text("#!/bin/sh\necho original\n")
os.chmod(target, 0o6755)
if stat.S_IMODE(target.stat().st_mode) != 0o6755:
pytest.skip("filesystem refuses setuid/setgid on a user-owned file")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
zip_path = tmp_path / "backup.zip"
self._zip(
zip_path,
{"helper.sh": "#!/bin/sh\necho attacker\n", "state.db": ""},
)
import hermes_cli.backup_restore as backup_mod
real_replace = backup_mod.atomic_replace
staged_modes: list[int] = []
def spying_replace(tmp, dst):
if Path(dst).name == "helper.sh":
staged_modes.append(stat.S_IMODE(os.stat(tmp).st_mode))
return real_replace(tmp, dst)
monkeypatch.setattr(backup_mod, "atomic_replace", spying_replace)
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
published = stat.S_IMODE(target.stat().st_mode)
assert target.read_text() == "#!/bin/sh\necho attacker\n"
assert not published & stat.S_ISUID, (
f"archive content kept the target's setuid bit (mode 0o{published:o})"
)
assert not published & stat.S_ISGID, (
f"archive content kept the target's setgid bit (mode 0o{published:o})"
)
# The ordinary permission bits are still preserved — this drops the
# elevated bits, it does not fall back to mkstemp's 0600.
assert published == 0o755
# And there must be no transient elevation either: the temp file is
# chmod'd before the replace, so it must never carry the bits.
assert staged_modes == [0o755], (
f"the staged temp file was elevated before publish: {staged_modes}"
)
# ---------------------------------------------------------------------------
# Profile restoration tests
# ---------------------------------------------------------------------------
class TestProfileRestoration:
def _make_backup_zip(self, zip_path: Path, files: dict[str, str | bytes]) -> None:
with zipfile.ZipFile(zip_path, "w") as zf:
for name, content in files.items():
zf.writestr(name, content)
def test_import_skips_profile_dirs_without_config(self, tmp_path, monkeypatch):
"""Import doesn't create wrappers for profile dirs without config."""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
wrapper_dir = tmp_path / ".local" / "bin"
wrapper_dir.mkdir(parents=True)
zip_path = tmp_path / "backup.zip"
self._make_backup_zip(zip_path, {
"config.yaml": "model: test\n",
"profiles/valid/config.yaml": "model: test\n",
"profiles/empty/readme.txt": "nothing here\n",
})
args = Namespace(zipfile=str(zip_path), force=True)
from hermes_cli.backup import run_import
run_import(args)
# Only valid profile should get a wrapper
assert (wrapper_dir / "valid").exists()
assert not (wrapper_dir / "empty").exists()
# ---------------------------------------------------------------------------
# SQLite safe copy tests
# ---------------------------------------------------------------------------
class TestSafeCopyDb:
def test_copies_valid_database(self, tmp_path):
from hermes_cli.backup import _safe_copy_db
src = tmp_path / "test.db"
dst = tmp_path / "copy.db"
conn = sqlite3.connect(str(src))
conn.execute("CREATE TABLE t (x INTEGER)")
conn.execute("INSERT INTO t VALUES (42)")
conn.commit()
conn.close()
result = _safe_copy_db(src, dst)
assert result is True
conn = sqlite3.connect(str(dst))
rows = conn.execute("SELECT x FROM t").fetchall()
conn.close()
assert rows == [(42,)]
def test_aborts_when_source_remains_busy_past_deadline(
self, tmp_path, monkeypatch
):
from hermes_cli import backup as backup_mod
src = tmp_path / "locked.db"
dst = tmp_path / "copy.db"
src.touch()
dst.write_bytes(b"partial")
clock = iter((100.0, 100.5, 101.1))
class FakeSourceConnection:
def backup(self, _destination, *, pages, progress, sleep):
assert pages > 0
assert sleep > 0
progress(sqlite3.SQLITE_BUSY, 0, 1)
progress(sqlite3.SQLITE_BUSY, 0, 1)
def close(self):
pass
destination_closed = []
class FakeDestinationConnection:
def close(self):
destination_closed.append(True)
connections = iter((FakeSourceConnection(), FakeDestinationConnection()))
real_unlink = Path.unlink
def assert_closed_before_unlink(path, *args, **kwargs):
assert destination_closed
return real_unlink(path, *args, **kwargs)
connect_calls = []
def fake_connect(*args, **kwargs):
connect_calls.append((args, kwargs))
return next(connections)
monkeypatch.setattr(backup_mod.sqlite3, "connect", fake_connect)
monkeypatch.setattr(backup_mod.time, "monotonic", lambda: next(clock))
monkeypatch.setattr(Path, "unlink", assert_closed_before_unlink)
assert backup_mod._safe_copy_db(src, dst, timeout_seconds=1.0) is False
assert connect_calls[0][1]["timeout"] == 0.0
assert not dst.exists()
def test_is_zeroed_sqlite_file_detects_nul_header(self, tmp_path):
from hermes_cli.backup import is_zeroed_sqlite_file
p = tmp_path / "state.db"
p.write_bytes(bytes(4096)) # all NULs
assert is_zeroed_sqlite_file(p) is True
# ---------------------------------------------------------------------------
# Quick state snapshot tests
# ---------------------------------------------------------------------------
class TestQuickSnapshot:
@pytest.fixture
def hermes_home(self, tmp_path):
"""Create a fake HERMES_HOME with critical state files."""
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text("model:\n provider: openrouter\n")
(home / ".env").write_text("OPENROUTER_API_KEY=test-key-123\n")
(home / "auth.json").write_text('{"providers": {}}\n')
(home / "channel_aliases.json").write_text(
'{"whatsapp": {"120363408391911677@g.us": "general"}}\n'
)
(home / "cron").mkdir()
(home / "cron" / "jobs.json").write_text('{"jobs": []}\n')
# Real SQLite database
db_path = home / "state.db"
conn = sqlite3.connect(str(db_path))
conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, data TEXT)")
conn.execute("INSERT INTO sessions VALUES ('s1', 'hello world')")
conn.commit()
conn.close()
return home
def test_state_db_safely_copied(self, hermes_home):
from hermes_cli.backup import create_quick_snapshot
snap_id = create_quick_snapshot(hermes_home=hermes_home)
db_copy = hermes_home / "state-snapshots" / snap_id / "state.db"
assert db_copy.exists()
conn = sqlite3.connect(str(db_copy))
rows = conn.execute("SELECT * FROM sessions").fetchall()
conn.close()
assert len(rows) == 1
assert rows[0] == ("s1", "hello world")
def test_failed_state_db_copy_is_loud(self, hermes_home, monkeypatch, capsys):
"""#68474: unreadable state.db must not look like a silent success."""
from hermes_cli import backup as backup_mod
def boom(src, dst):
return False
monkeypatch.setattr(backup_mod, "_safe_copy_db", boom)
snap_id = backup_mod.create_quick_snapshot(hermes_home=hermes_home)
# Other small files still snapshot; the failed DB is recorded, not silently dropped.
assert snap_id
manifest = (hermes_home / "state-snapshots" / snap_id / "manifest.json")
data = json.loads(manifest.read_text(encoding="utf-8"))
assert "state.db" not in data.get("files", {})
assert "state.db" in data.get("failed_dbs", [])
def test_restore_state_db_live_connection(self, hermes_home):
"""Restoring state.db must update data visible through a live connection.
Regression test for #65942: when state.db is open with a live SQLite
connection (as happens with the gateway, dashboard, or another CLI
session), the restore must write pages through the backup API so the
live connection sees the restored data instead of stale cached pages
from a replaced inode.
"""
from hermes_cli.backup import create_quick_snapshot, restore_quick_snapshot
snap_id = create_quick_snapshot(hermes_home=hermes_home)
# Open a live connection (simulating gateway/dashboard).
live_conn = sqlite3.connect(str(hermes_home / "state.db"))
live_conn.execute("PRAGMA journal_mode=wal")
# Insert data AFTER the snapshot — this is what must be reverted.
live_conn.execute("INSERT INTO sessions VALUES ('s2', 'new-data')")
live_conn.commit()
rows_before = live_conn.execute("SELECT * FROM sessions").fetchall()
assert len(rows_before) == 2
# Restore — the live connection stays open during restore.
result = restore_quick_snapshot(snap_id, hermes_home=hermes_home)
assert result is True
# The live connection must see the restored (single-row) state.
# A fresh connection would trivially work; the live one is the test.
rows_after = live_conn.execute("SELECT * FROM sessions").fetchall()
live_conn.close()
assert len(rows_after) == 1, (
f"Live connection still sees {len(rows_after)} rows after restore "
f"(expected 1); the extra row 's2' should have been reverted."
)
def test_snapshot_includes_pairing_directories(self, hermes_home):
"""Pairing JSONs live outside state.db — snapshot must capture them
recursively (generic + per-platform) so approved-user lists survive
disasters like #15733."""
from hermes_cli.backup import create_quick_snapshot
# Generic pairing store (new location)
(hermes_home / "platforms" / "pairing").mkdir(parents=True)
(hermes_home / "platforms" / "pairing" / "telegram-approved.json").write_text(
'{"12345": {"user_name": "alice"}}'
)
(hermes_home / "platforms" / "pairing" / "discord-approved.json").write_text(
'{"67890": {"user_name": "bob"}}'
)
# Legacy pairing store (old location)
(hermes_home / "pairing").mkdir()
(hermes_home / "pairing" / "matrix-approved.json").write_text(
'{"@charlie:server": {"user_name": "charlie"}}'
)
# Feishu's separate JSON
(hermes_home / "feishu_comment_pairing.json").write_text(
'{"doc_abc": {"allow_from": ["user_xyz"]}}'
)
snap_id = create_quick_snapshot(hermes_home=hermes_home)
assert snap_id is not None
snap_dir = hermes_home / "state-snapshots" / snap_id
assert (snap_dir / "platforms" / "pairing" / "telegram-approved.json").exists()
assert (snap_dir / "platforms" / "pairing" / "discord-approved.json").exists()
assert (snap_dir / "pairing" / "matrix-approved.json").exists()
assert (snap_dir / "feishu_comment_pairing.json").exists()
with open(snap_dir / "manifest.json") as f:
meta = json.load(f)
files = meta["files"]
assert "platforms/pairing/telegram-approved.json" in files
assert "platforms/pairing/discord-approved.json" in files
assert "pairing/matrix-approved.json" in files
assert "feishu_comment_pairing.json" in files
# ---------------------------------------------------------------------------
# Pre-update backup (hermes update safety net)
# ---------------------------------------------------------------------------
# -- security: path traversal regression coverage -----------------------
# Per @egilewski audit on PR #9217: restore_quick_snapshot must reject
# malicious snapshot_id values (the directory selector) AND malicious
# rel paths inside the manifest (the per-file selector). Both surfaces
# need explicit regression tests because they validate independent
# traversal vectors.
def test_oversized_db_suppresses_pruning(self, hermes_home, capsys):
"""#68805: an oversized state.db skipped for size must suppress
pruning so the older complete snapshot (containing the only
recoverable database) is preserved.
Reproduces the reviewer's scenario: keep=1 + a state.db exceeding
the size cap → the new snapshot omits state.db, failed_dbs stays
empty (the file wasn't unreadable, just too large), and without
tracking oversized_skipped the older complete snapshot would be
pruned — losing the only recovery copy.
"""
import json
from hermes_cli.backup import create_quick_snapshot, list_quick_snapshots
# First snapshot: complete (state.db is small, under any cap)
first_id = create_quick_snapshot(label="complete", hermes_home=hermes_home)
assert first_id is not None
first_dir = hermes_home / "state-snapshots" / first_id
assert (first_dir / "state.db").exists()
_advance_backup_clock()
# Second snapshot: state.db exceeds the 1024-byte cap → skipped for
# size, but small config files (32-54 bytes) still land in the manifest.
second_id = create_quick_snapshot(
label="oversized", hermes_home=hermes_home, max_file_size=1024, keep=1
)
assert second_id is not None
second_dir = hermes_home / "state-snapshots" / second_id
assert not (second_dir / "state.db").exists()
# Manifest must record the oversized skip
with open(second_dir / "manifest.json") as f:
meta = json.load(f)
assert "state.db" in meta.get("oversized_skipped", [])
# CRITICAL: the first (complete) snapshot must survive pruning
# because the second snapshot is incomplete (oversized state.db).
all_snaps = list_quick_snapshots(limit=100, hermes_home=hermes_home)
snap_ids = {s["id"] for s in all_snaps}
assert first_id in snap_ids, (
f"Complete snapshot {first_id} was pruned by an incomplete "
f"(oversized) snapshot — the recovery copy was lost!"
)
assert second_id in snap_ids
class TestQuickSnapshotProjectsKanban:
"""Regression for #52889: projects.db / kanban.db must survive an upgrade.
Both are per-profile user-created stores outside the git checkout. If they
are not in the pre-update snapshot, the post-update ``CREATE TABLE IF NOT
EXISTS`` runs against a missing file and every project / board row is lost.
"""
@pytest.fixture
def hermes_home(self, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
# Minimal critical file so the snapshot is non-empty.
(home / "config.yaml").write_text("model:\n provider: openrouter\n")
for name, table, row in (
("projects.db", "projects", ("p1", "demo")),
("kanban.db", "tasks", ("t1", "todo")),
):
conn = sqlite3.connect(str(home / name))
conn.execute(f"CREATE TABLE {table} (id TEXT PRIMARY KEY, data TEXT)")
conn.execute(f"INSERT INTO {table} VALUES (?, ?)", row)
conn.commit()
conn.close()
return home
def test_non_default_kanban_board_snapshotted(self, hermes_home):
"""#52889 completeness: non-default boards live at
<root>/kanban/boards/<slug>/kanban.db, not <root>/kanban.db. The
``kanban/boards`` dir entry must capture them too, or multi-board
users still lose every board except ``default`` on upgrade."""
from hermes_cli.backup import create_quick_snapshot, restore_quick_snapshot
board_dir = hermes_home / "kanban" / "boards" / "work"
board_dir.mkdir(parents=True)
conn = sqlite3.connect(str(board_dir / "kanban.db"))
conn.execute("CREATE TABLE tasks (id TEXT PRIMARY KEY, data TEXT)")
conn.execute("INSERT INTO tasks VALUES (?, ?)", ("w1", "ship"))
conn.commit()
conn.close()
snap_id = create_quick_snapshot(hermes_home=hermes_home)
copy = (
hermes_home / "state-snapshots" / snap_id
/ "kanban" / "boards" / "work" / "kanban.db"
)
assert copy.exists(), "non-default board kanban.db was not snapshotted"
# Simulate the upgrade wiping the board, then restore it.
conn = sqlite3.connect(str(board_dir / "kanban.db"))
conn.execute("DELETE FROM tasks")
conn.commit()
conn.close()
assert restore_quick_snapshot(snap_id, hermes_home=hermes_home) is True
conn = sqlite3.connect(str(board_dir / "kanban.db"))
rows = conn.execute("SELECT * FROM tasks").fetchall()
conn.close()
assert rows == [("w1", "ship")]
def test_board_db_copied_wal_safely(self, hermes_home, monkeypatch):
"""#52889 W2: a non-default board's .db (dir-branch) must go through the
WAL-safe _safe_copy_db, not a raw shutil.copy2, so an open WAL doesn't
produce an inconsistent copy."""
import hermes_cli.backup as bk
from hermes_cli.backup import create_quick_snapshot
board = hermes_home / "kanban" / "boards" / "work"
board.mkdir(parents=True)
conn = sqlite3.connect(str(board / "kanban.db"))
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("CREATE TABLE tasks (id TEXT PRIMARY KEY, data TEXT)")
conn.execute("INSERT INTO tasks VALUES ('w1', 'ship')")
conn.commit()
conn.close()
called = {"db": []}
real = bk._safe_copy_db
def _spy(src, dst):
called["db"].append(str(src))
return real(src, dst)
monkeypatch.setattr(bk, "_safe_copy_db", _spy)
snap_id = create_quick_snapshot(hermes_home=hermes_home)
# The board db was copied via _safe_copy_db (not raw copy).
assert any(s.endswith("boards/work/kanban.db") for s in called["db"]), called["db"]
copy = hermes_home / "state-snapshots" / snap_id / "kanban" / "boards" / "work" / "kanban.db"
rows = sqlite3.connect(str(copy)).execute("SELECT * FROM tasks").fetchall()
assert rows == [("w1", "ship")]
def _fail_zip_write_after(monkeypatch, member: str, byte_count: int = 120_000) -> None:
"""Make ZipFile.write finalize a large partial member, then surface a source-read failure."""
real_write = zipfile.ZipFile.write
def flaky_write(self, filename, arcname=None, compress_type=None, compresslevel=None):
if str(arcname) == member:
with Path(filename).open("rb") as src, self.open(str(arcname), "w") as dst:
dst.write(src.read(byte_count))
raise OSError(5, "simulated source read failure")
return real_write(self, filename, arcname, compress_type, compresslevel)
monkeypatch.setattr(zipfile.ZipFile, "write", flaky_write)
class TestFailedZipMemberRecovery:
def test_automatic_backup_omits_crc_valid_partial_member(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text("model: test\n")
(hermes_home / "flaky.bin").write_bytes(os.urandom(300_000))
archive = tmp_path / "automatic.zip"
_fail_zip_write_after(monkeypatch, "flaky.bin")
from hermes_cli.backup import _write_full_zip_backup
assert _write_full_zip_backup(archive, hermes_home) is None
assert not archive.exists()
salvage = tmp_path / "automatic.incomplete.zip"
# A streaming reader scans local headers, so the dropped member's bytes must be gone too.
assert b"flaky.bin" not in salvage.read_bytes()
with zipfile.ZipFile(salvage) as zf:
assert "flaky.bin" not in zf.namelist()
assert zf.read("config.yaml") == b"model: test\n"
assert zf.testzip() is None
def test_incomplete_pre_update_backup_does_not_rotate_last_complete(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
(hermes_home / "config.yaml").write_text("model: test\n")
from hermes_cli.backup import create_pre_update_backup
good = create_pre_update_backup(hermes_home=hermes_home, keep=2)
assert good is not None and good.exists()
real_write = zipfile.ZipFile.write
(hermes_home / "flaky.bin").write_bytes(os.urandom(300_000))
_fail_zip_write_after(monkeypatch, "flaky.bin")
for _ in range(2):
_advance_backup_clock()
assert create_pre_update_backup(hermes_home=hermes_home, keep=2) is None
assert good.exists(), "an incomplete generation rotated out the last complete backup"
backup_dir = hermes_home / "backups"
salvages = list(backup_dir.glob("pre-update-*.incomplete.zip"))
assert len(salvages) == 1, "repeated incomplete runs must not pile up"
with zipfile.ZipFile(salvages[0]) as zf:
assert "flaky.bin" not in zf.namelist()
assert zf.read("config.yaml") == b"model: test\n"
assert zf.testzip() is None
# Salvage archives must not count toward retention on the next complete run.
monkeypatch.setattr(zipfile.ZipFile, "write", real_write)
(hermes_home / "flaky.bin").unlink()
_advance_backup_clock()
newest = create_pre_update_backup(hermes_home=hermes_home, keep=2)
assert newest is not None and newest.exists()
assert good.exists(), "a complete run pruned complete backups in favour of salvage"
assert len(list(backup_dir.glob("pre-update-*.incomplete.zip"))) == 1
class TestPreUpdateBackup:
"""Tests for create_pre_update_backup — the auto-backup ``hermes update``
runs before touching anything."""
@pytest.fixture
def hermes_home(self, tmp_path):
root = tmp_path / ".hermes"
root.mkdir()
_make_hermes_tree(root)
return root
def test_backup_contents_match_full_backup(self, hermes_home):
"""Pre-update backup should include the same user data that
``hermes backup`` would, and should exclude the same directories."""
from hermes_cli.backup import create_pre_update_backup
out = create_pre_update_backup(hermes_home=hermes_home)
assert out is not None
with zipfile.ZipFile(out) as zf:
names = set(zf.namelist())
# User data present
assert "config.yaml" in names
assert ".env" in names
assert "sessions/abc123.json" in names
assert "skills/my-skill/SKILL.md" in names
assert "profiles/coder/config.yaml" in names
# hermes-agent repo excluded
assert not any(n.startswith("hermes-agent/") for n in names)
# __pycache__ excluded
assert not any("__pycache__" in n for n in names)
# pid files excluded
assert "gateway.pid" not in names
def test_pre_update_zip_does_not_nest_the_pre_update_snapshot(self, hermes_home):
"""``hermes update`` in ``full`` mode takes the quick snapshot *before*
the full zip, so the zip walk sees the snapshot it just made. It must
skip it — otherwise every pre-update zip ships state.db twice."""
from hermes_cli.backup import (
_QUICK_SNAPSHOTS_DIR,
create_pre_update_backup,
create_quick_snapshot,
)
with sqlite3.connect(hermes_home / "state.db") as conn:
conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)")
snap_id = create_quick_snapshot(label="pre-update", hermes_home=hermes_home)
assert snap_id and (hermes_home / _QUICK_SNAPSHOTS_DIR / snap_id / "state.db").exists()
out = create_pre_update_backup(hermes_home=hermes_home)
assert out is not None
with zipfile.ZipFile(out) as zf:
names = zf.namelist()
assert "state.db" in names
assert not any(n.startswith(_QUICK_SNAPSHOTS_DIR + "/") for n in names), names
def test_rotation_keeps_only_n(self, hermes_home):
"""After more than ``keep`` backups are created, older ones are
pruned automatically."""
from hermes_cli.backup import create_pre_update_backup
created = []
for _ in range(5):
out = create_pre_update_backup(hermes_home=hermes_home, keep=3)
created.append(out)
_advance_backup_clock()
remaining = sorted(
p.name for p in (hermes_home / "backups").iterdir()
if p.name.startswith("pre-update-")
)
assert len(remaining) == 3
# Oldest two should have been pruned
assert created[0].name not in remaining
assert created[1].name not in remaining
# Newest three should remain
assert created[4].name in remaining
def test_skips_symlinked_files(self, hermes_home, tmp_path):
"""Pre-update backups must not dereference symlinks outside HERMES_HOME."""
from hermes_cli.backup import create_pre_update_backup
outside = tmp_path / "outside-secret.txt"
outside.write_text("outside secret\n")
_symlink_file_or_skip(hermes_home / "skills" / "outside-link.txt", outside)
out = create_pre_update_backup(hermes_home=hermes_home)
assert out is not None
with zipfile.ZipFile(out) as zf:
names = zf.namelist()
assert "skills/outside-link.txt" not in names
assert all(zf.read(name) != b"outside secret\n" for name in names)
class TestRunPreUpdateBackup:
"""Tests for the ``_run_pre_update_backup`` wrapper in main.py —
covers the consolidated off/quick/full mode gate, CLI flags, and
user-facing output."""
@pytest.fixture
def hermes_home(self, tmp_path, monkeypatch):
root = tmp_path / ".hermes"
root.mkdir()
_make_hermes_tree(root)
# Point HERMES_HOME at the temp dir so config + backup paths resolve here
monkeypatch.setenv("HERMES_HOME", str(root))
# Make Path.home() point at tmp_path for anything that uses it
monkeypatch.setattr(Path, "home", lambda: tmp_path)
# Config reads resolve HERMES_HOME dynamically and their caches are
# keyed by config path. Do not remove shared modules from sys.modules:
# other test modules may retain imports from the existing module object.
return root
@staticmethod
def _set_mode(hermes_home, value):
import hermes_yaml as yaml
(hermes_home / "config.yaml").write_text(yaml.safe_dump({
"_config_version": 22,
"updates": {"pre_update_backup": value},
}))
@staticmethod
def _zips(hermes_home):
d = hermes_home / "backups"
return list(d.glob("pre-update-*.zip")) if d.exists() else []
@staticmethod
def _snaps(hermes_home):
d = hermes_home / "state-snapshots"
return [p for p in d.iterdir() if p.is_dir()] if d.exists() else []
def test_config_off_disables_everything_silently(self, hermes_home, capsys):
"""pre_update_backup: off — an explicit opt-out disables the quick
snapshot too (it previously ran unconditionally), with no output."""
self._set_mode(hermes_home, "off")
from hermes_cli.update_cmd import _run_pre_update_backup
snap_id = _run_pre_update_backup(Namespace(no_backup=False, backup=False))
out = capsys.readouterr().out
assert snap_id is None
assert out == ""
assert not self._snaps(hermes_home)
assert not self._zips(hermes_home)
def test_config_full_mode(self, hermes_home, capsys):
self._set_mode(hermes_home, "full")
from hermes_cli.update_cmd import _run_pre_update_backup
snap_id = _run_pre_update_backup(Namespace(no_backup=False, backup=False))
assert snap_id is not None
assert len(self._zips(hermes_home)) == 1
# ---------------------------------------------------------------------------
# Pre-migration backup (hermes claw migrate safety net)
# ---------------------------------------------------------------------------
class TestPreMigrationBackup:
"""Tests for create_pre_migration_backup — the auto-backup
``hermes claw migrate`` runs before mutating ~/.hermes/."""
@pytest.fixture
def hermes_home(self, tmp_path):
root = tmp_path / ".hermes"
root.mkdir()
_make_hermes_tree(root)
return root
def test_restorable_with_hermes_import(self, hermes_home, tmp_path):
"""The zip produced by pre-migration backup must be a valid Hermes
backup — `hermes import` should accept it."""
from hermes_cli.backup import create_pre_migration_backup, _validate_backup_zip
out = create_pre_migration_backup(hermes_home=hermes_home)
assert out is not None
with zipfile.ZipFile(out) as zf:
valid, _reason = _validate_backup_zip(zf)
assert valid, "pre-migration zip failed _validate_backup_zip"
def test_does_not_touch_pre_update_backups(self, hermes_home):
"""Pre-migration rotation must only prune pre-migration-*.zip files,
leaving pre-update-*.zip backups untouched."""
from hermes_cli.backup import create_pre_update_backup, create_pre_migration_backup
update_backup = create_pre_update_backup(hermes_home=hermes_home, keep=5)
assert update_backup is not None and update_backup.exists()
# Spin up a lot of migration backups with keep=1
for _ in range(3):
out = create_pre_migration_backup(hermes_home=hermes_home, keep=1)
assert out is not None
_advance_backup_clock()
# Update backup must still be there
assert update_backup.exists(), "pre-migration rotation wrongly pruned the pre-update backup"
# ---------------------------------------------------------------------------
# Cron jobs auto-restore after silent migration loss (issue #34600)
# ---------------------------------------------------------------------------
class TestRestoreCronJobsIfEmptied:
"""`hermes update` config migration can leave cron/jobs.json valid-but-empty,
silently dropping every scheduled job. `restore_cron_jobs_if_emptied` is the
post-migration safety net that restores from the pre-update snapshot."""
@staticmethod
def _seed_jobs(path: Path, jobs):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"jobs": jobs}))
def _make_snapshot(self, hermes_home: Path, label="pre-update"):
from hermes_cli.backup import create_quick_snapshot
return create_quick_snapshot(label=label, hermes_home=hermes_home, keep=5)
def test_restores_when_emptied_after_migration(self, tmp_path):
from hermes_cli.backup import restore_cron_jobs_if_emptied
hermes_home = tmp_path / ".hermes"
jobs_path = hermes_home / "cron" / "jobs.json"
# Pre-update: 3 real jobs.
self._seed_jobs(jobs_path, [{"id": "a"}, {"id": "b"}, {"id": "c"}])
snap_id = self._make_snapshot(hermes_home)
assert snap_id
# Migration silently empties the file (valid JSON, zero jobs).
jobs_path.write_text(json.dumps({"jobs": []}))
result = restore_cron_jobs_if_emptied(snap_id, hermes_home=hermes_home)
assert result is not None
assert result["restored"] is True
assert result["job_count"] == 3
assert result["snapshot_id"] == snap_id
# The live file now has the jobs back.
restored = json.loads(jobs_path.read_text())
assert len(restored["jobs"]) == 3
def test_restores_when_partial_job_loss(self, tmp_path):
"""Desktop scheduler overwrites jobs.json with its own small set,
losing tool-created crons while keeping desktop-tracked ones."""
from hermes_cli.backup import restore_cron_jobs_if_emptied
hermes_home = tmp_path / ".hermes"
jobs_path = hermes_home / "cron" / "jobs.json"
# Pre-update: 19 jobs (18 tool-created + 1 desktop watchdog).
self._seed_jobs(
jobs_path,
[{"id": f"job-{i}"} for i in range(19)],
)
snap_id = self._make_snapshot(hermes_home)
assert snap_id
# Desktop scheduler overwrites with only its own 1 job.
jobs_path.write_text(json.dumps({"jobs": [{"id": "desktop-watchdog"}]}))
result = restore_cron_jobs_if_emptied(snap_id, hermes_home=hermes_home)
assert result is not None
assert result["restored"] is True
assert result["job_count"] == 19
# The live file now has all 19 jobs back.
restored = json.loads(jobs_path.read_text())
assert len(restored["jobs"]) == 19
# ---------------------------------------------------------------------------
# config.yaml model/provider + MoA auto-restore after silent update rewrite
# (issue #64160)
# ---------------------------------------------------------------------------
class TestRestoreConfigModelSettingsIfRewritten:
"""Desktop update/repair cycles have rewritten user-set model.provider /
model.default and dropped the moa: section (#64160).
`restore_config_model_settings_if_rewritten` is the post-update safety net
that restores only the protected keys from the pre-update snapshot."""
USER_CONFIG = (
"_config_version: 39\n"
"model:\n"
" provider: custom\n"
" default: zyphra/zamba-3-large\n"
" base_url: https://api.zyphra.example/v1\n"
"moa:\n"
" enabled: true\n"
" presets:\n"
" council:\n"
" aggregator: {provider: custom, model: zyphra/zamba-3-large}\n"
"custom_unknown_key:\n"
" hello: world\n"
)
def _make_snapshot(self, hermes_home: Path, label="pre-update"):
from hermes_cli.backup import create_quick_snapshot
return create_quick_snapshot(label=label, hermes_home=hermes_home, keep=5)
def _seed(self, hermes_home: Path) -> Path:
hermes_home.mkdir(parents=True, exist_ok=True)
cfg = hermes_home / "config.yaml"
cfg.write_text(self.USER_CONFIG, encoding="utf-8")
return cfg
def test_restores_rewritten_provider_and_dropped_moa(self, tmp_path):
import hermes_yaml as yaml
from hermes_cli.backup import restore_config_model_settings_if_rewritten
hermes_home = tmp_path / ".hermes"
cfg = self._seed(hermes_home)
snap_id = self._make_snapshot(hermes_home)
assert snap_id
# The update flow rewrites config.yaml with defaults: provider flips
# to deepseek, MoA section is gone (the #64160 field report).
cfg.write_text(
"_config_version: 39\nmodel:\n provider: deepseek\n default: deepseek-chat\n",
encoding="utf-8",
)
result = restore_config_model_settings_if_rewritten(
snap_id, hermes_home=hermes_home
)
assert result is not None
assert result["restored"] is True
assert result["snapshot_id"] == snap_id
assert "model.provider" in result["keys"]
assert "model.default" in result["keys"]
assert "moa" in result["keys"]
after = yaml.safe_load(cfg.read_text(encoding="utf-8"))
assert after["model"]["provider"] == "custom"
assert after["model"]["default"] == "zyphra/zamba-3-large"
assert after["model"]["base_url"] == "https://api.zyphra.example/v1"
assert after["moa"]["enabled"] is True
assert after["moa"]["presets"]["council"]["aggregator"]["model"] == (
"zyphra/zamba-3-large"
)
def test_noop_when_config_untouched(self, tmp_path):
from hermes_cli.backup import restore_config_model_settings_if_rewritten
hermes_home = tmp_path / ".hermes"
cfg = self._seed(hermes_home)
snap_id = self._make_snapshot(hermes_home)
before = cfg.read_text(encoding="utf-8")
result = restore_config_model_settings_if_rewritten(
snap_id, hermes_home=hermes_home
)
assert result is None
assert cfg.read_text(encoding="utf-8") == before
def test_preserves_legitimate_update_writes(self, tmp_path):
"""Only protected keys are restored — a version bump or a new section
the migration legitimately wrote must survive the restore."""
import hermes_yaml as yaml
from hermes_cli.backup import restore_config_model_settings_if_rewritten
hermes_home = tmp_path / ".hermes"
cfg = self._seed(hermes_home)
snap_id = self._make_snapshot(hermes_home)
cfg.write_text(
"_config_version: 40\n" # legitimate migration bump
"new_section:\n added: true\n" # legitimate new default
"model:\n provider: deepseek\n", # illegitimate rewrite
encoding="utf-8",
)
result = restore_config_model_settings_if_rewritten(
snap_id, hermes_home=hermes_home
)
assert result is not None
after = yaml.safe_load(cfg.read_text(encoding="utf-8"))
assert after["model"]["provider"] == "custom" # restored
assert after["_config_version"] == 40 # kept
assert after["new_section"] == {"added": True} # kept
def test_noop_when_user_never_set_protected_keys(self, tmp_path):
"""A config that never had model.provider/moa set gets no restore even
if the update writes those keys fresh — nothing of the user's was lost."""
from hermes_cli.backup import restore_config_model_settings_if_rewritten
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir(parents=True)
cfg = hermes_home / "config.yaml"
cfg.write_text("_config_version: 39\nagent: {}\n", encoding="utf-8")
snap_id = self._make_snapshot(hermes_home)
cfg.write_text(
"_config_version: 39\nmodel:\n provider: deepseek\n", encoding="utf-8"
)
result = restore_config_model_settings_if_rewritten(
snap_id, hermes_home=hermes_home
)
assert result is None
def test_noop_on_unreadable_live_config(self, tmp_path):
"""An unparseable live config is a different failure the user must see;
the safety net leaves it alone (mirrors the cron net's posture)."""
from hermes_cli.backup import restore_config_model_settings_if_rewritten
hermes_home = tmp_path / ".hermes"
cfg = self._seed(hermes_home)
snap_id = self._make_snapshot(hermes_home)
cfg.write_text(": not [valid yaml", encoding="utf-8")
result = restore_config_model_settings_if_rewritten(
snap_id, hermes_home=hermes_home
)
assert result is None
assert cfg.read_text(encoding="utf-8") == ": not [valid yaml"
def test_noop_without_snapshot_id(self, tmp_path):
from hermes_cli.backup import restore_config_model_settings_if_rewritten
assert restore_config_model_settings_if_rewritten(
"", hermes_home=tmp_path / ".hermes"
) is None
# ---------------------------------------------------------------------------
# Memory-provider external paths (~/.honcho, ~/.hindsight, ...) — captured via
# MemoryProvider.backup_paths() and restored to their original home-relative
# location, NOT under HERMES_HOME. (backup/import cycle data-loss fix)
# ---------------------------------------------------------------------------
class TestMemoryProviderExternalPaths:
def _make_min_tree(self, hermes_home: Path) -> None:
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "config.yaml").write_text("model:\n provider: openrouter\n")
(hermes_home / ".env").write_text("OPENROUTER_API_KEY=sk-test\n")
(hermes_home / "state.db").write_bytes(b"x")
def test_backup_skips_external_paths_outside_home(self, tmp_path, monkeypatch):
"""A declared path outside the home dir is not portable and must be
skipped, never archived."""
hermes_home = tmp_path / ".hermes"
self._make_min_tree(hermes_home)
outside = tmp_path.parent / "outside-home-secret"
outside.mkdir(exist_ok=True)
(outside / "leak.json").write_text('{"secret":1}')
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
import hermes_cli.backup as backup_mod
monkeypatch.setattr(
backup_mod, "_collect_memory_provider_external_paths", lambda: [outside]
)
out_zip = tmp_path / "backup.zip"
backup_mod.run_backup(Namespace(output=str(out_zip)))
with zipfile.ZipFile(out_zip) as zf:
names = set(zf.namelist())
assert not any(n.startswith("_external/") for n in names)
assert not any("leak.json" in n for n in names)
(outside / "leak.json").unlink()
outside.rmdir()
def test_import_restores_external_to_home_relative_location(self, tmp_path, monkeypatch):
"""_external/ members restore to ~/<relpath>, not under HERMES_HOME,
and credential-shaped files get 0600."""
dst_home = tmp_path / "dst"
dst_home.mkdir()
hermes_home = dst_home / ".hermes"
hermes_home.mkdir()
zip_path = tmp_path / "backup.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.writestr("config.yaml", "model: {}\n")
zf.writestr(".env", "X=1\n")
zf.writestr("state.db", "")
zf.writestr("_external/.honcho/config.json", '{"peer":"bob"}')
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: dst_home)
from hermes_cli.backup import run_import
run_import(Namespace(zipfile=str(zip_path), force=True))
restored = dst_home / ".honcho" / "config.json"
assert restored.exists()
assert restored.read_text() == '{"peer":"bob"}'
# Credential-shaped file tightened.
assert (restored.stat().st_mode & 0o777) == 0o600
# External state did NOT leak into HERMES_HOME.
assert not (hermes_home / "_external").exists()
# ---------------------------------------------------------------------------
# run_import: HERMES_HOME override handling (issue #99839)
# ---------------------------------------------------------------------------
class TestImportHonorsHermesHomeOverride:
"""`hermes import` must restore into the home the command runs under.
Resolving the target through get_default_hermes_root() maps a profile
home (<root>/profiles/<name>) back to <root>: the import then overwrites
the live root's config.yaml while the profile directory stays empty —
exactly what "Target:" printed it would NOT do.
"""
def _make_backup_zip(self, tmp_path):
import zipfile
src_root = tmp_path / "src-home"
src_root.mkdir()
(src_root / "config.yaml").write_text("model:\n provider: anthropic\n")
(src_root / ".env").write_text("ANTHROPIC_API_KEY=sk-test\n")
zip_path = tmp_path / "backup.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.write(src_root / "config.yaml", "config.yaml")
zf.write(src_root / ".env", ".env")
return zip_path
def test_import_targets_named_profile_home(self, tmp_path, monkeypatch):
"""HERMES_HOME=<root>/profiles/<name> must restore INTO the profile,
not into <root> (which would clobber the live root config)."""
root = tmp_path / "hermes-root"
profile = root / "profiles" / "coder"
profile.mkdir(parents=True)
# Live root config that must survive untouched.
(root / "config.yaml").write_text("model:\n provider: openai\n")
monkeypatch.setenv("HERMES_HOME", str(profile))
from hermes_constants import get_hermes_home
assert get_hermes_home() == profile
zip_path = self._make_backup_zip(tmp_path)
import argparse
from hermes_cli.backup import run_import
args = argparse.Namespace(zipfile=str(zip_path), force=True)
run_import(args)
assert (profile / "config.yaml").read_text() == (
"model:\n provider: anthropic\n"
)
assert (root / "config.yaml").read_text() == "model:\n provider: openai\n"
def test_import_skips_gateway_install_for_non_default_home(
self, tmp_path, monkeypatch
):
"""A restore into a sandbox must not silently start a second gateway
pointed at it — the profile/sandbox gateway would shadow the default
service installed by the primary install."""
native_default = tmp_path / "native-default"
sandbox = tmp_path / "sandbox-home"
sandbox.mkdir(parents=True)
# Live default install markers.
native_default.mkdir()
(native_default / "config.yaml").write_text("model:\n provider: openai\n")
monkeypatch.setenv("HERMES_HOME", str(sandbox))
import argparse
from hermes_cli import backup as backup_mod
monkeypatch.setattr(
backup_mod,
"_get_platform_default_hermes_home",
lambda: native_default,
)
calls = []
monkeypatch.setattr(
"hermes_cli.gateway.ensure_gateway_service",
lambda *a, **kw: calls.append(kw),
)
monkeypatch.setattr(
"hermes_cli.gateway._is_service_running",
lambda: False,
)
zip_path = self._make_backup_zip(tmp_path)
args = argparse.Namespace(zipfile=str(zip_path), force=True)
backup_mod.run_import(args)
assert calls == [], "gateway must not be auto-installed for a sandbox restore"
def test_import_installs_gateway_when_default_home_is_target(
self, tmp_path, monkeypatch
):
"""Restoring into the default home keeps the auto-install behavior."""
native_default = tmp_path / "native-default"
native_default.mkdir()
monkeypatch.setenv("HERMES_HOME", str(native_default))
import argparse
from hermes_cli import backup as backup_mod
monkeypatch.setattr(
backup_mod,
"_get_platform_default_hermes_home",
lambda: native_default,
)
calls = []
monkeypatch.setattr(
"hermes_cli.gateway.ensure_gateway_service",
lambda *a, **kw: calls.append(kw),
)
monkeypatch.setattr(
"hermes_cli.gateway._is_service_running",
lambda: False,
)
zip_path = self._make_backup_zip(tmp_path)
args = argparse.Namespace(zipfile=str(zip_path), force=True)
backup_mod.run_import(args)
assert calls and calls[0].get("context") == "import"
# ---------------------------------------------------------------------------
# Live session database import (issue #100960)
# ---------------------------------------------------------------------------
def _write_session_db(path: Path, sessions: int, messages_per_session: int) -> None:
"""Create a minimal Hermes-shaped session database at *path*."""
conn = sqlite3.connect(str(path))
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS sessions "
"(session_id TEXT PRIMARY KEY, message_count INTEGER)"
)
conn.execute(
"CREATE TABLE IF NOT EXISTS messages "
"(id INTEGER PRIMARY KEY, session_id TEXT, content TEXT)"
)
for s in range(sessions):
sid = f"sess-{s}"
conn.execute(
"INSERT INTO sessions VALUES (?, ?)", (sid, messages_per_session)
)
for m in range(messages_per_session):
conn.execute(
"INSERT INTO messages (session_id, content) VALUES (?, ?)",
(sid, f"{sid}-msg-{m}"),
)
conn.commit()
finally:
conn.close()
class TestImportLiveSessionDatabase:
"""`hermes import` must not swap the inode of a database Hermes holds open.
Publishing state.db with a rename leaves any live gateway/dashboard/WebUI
connection reading and writing the unlinked inode, so its sessions vanish
from the database everyone else opens and nothing is logged (#100960).
"""
def _zip_with_db(self, zip_path: Path, db_path: Path) -> None:
with zipfile.ZipFile(zip_path, "w") as zf:
zf.write(db_path, "state.db")
def _prepare(self, tmp_path, monkeypatch, live=(3, 4), backup=(2, 2)):
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
live_db = home / "state.db"
_write_session_db(live_db, *live)
staged = tmp_path / "backup-state.db"
_write_session_db(staged, *backup)
zip_path = tmp_path / "backup.zip"
self._zip_with_db(zip_path, staged)
return home, live_db, zip_path
def test_live_holder_sees_imported_rows(self, tmp_path, monkeypatch):
"""A connection open across the import converges on the imported data."""
from hermes_cli.backup import run_import
home, live_db, zip_path = self._prepare(tmp_path, monkeypatch)
holder = sqlite3.connect(str(live_db))
# Read first so the connection has cached pages of the pre-import file.
assert holder.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 12
inode_before = os.stat(live_db).st_ino
try:
run_import(Namespace(zipfile=str(zip_path), force=True))
assert holder.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 4
finally:
holder.close()
assert os.stat(live_db).st_ino == inode_before
assert _count_rows(live_db) == (2, 4)
def test_refused_restore_is_reported_and_leaves_db_intact(
self, tmp_path, monkeypatch, capsys
):
"""A refused live-safe restore is a warning, not a counted success."""
import hermes_cli.backup as backup_mod
# _import_db_member (the run_import .db publish path) lives in
# hermes_cli.backup_restore and resolves _safe_restore_db there.
import hermes_cli.backup_restore as backup_restore_mod
home, live_db, zip_path = self._prepare(tmp_path, monkeypatch)
monkeypatch.setattr(backup_restore_mod, "_safe_restore_db", lambda src, dst: False)
assert backup_mod.run_import(Namespace(zipfile=str(zip_path), force=True)) == 1
out = capsys.readouterr().out
assert "files skipped" in out
assert "state.db" in out
assert "has been restored" not in out
# The pre-import database is still the one on disk.
assert _count_rows(live_db) == (3, 12)
def test_sidecar_members_are_not_installed_beside_a_restored_db(
self, tmp_path, monkeypatch
):
"""A `state.db-wal` member from an old/hand-built archive must not be
os.replace'd next to the page-restored database: it describes a
different image and SQLite would replay it on the next open."""
from hermes_cli.backup import run_import
home, live_db, zip_path = self._prepare(tmp_path, monkeypatch)
with zipfile.ZipFile(zip_path, "a") as zf:
zf.writestr("state.db-wal", b"foreign-wal-from-archive")
zf.writestr("state.db-shm", b"foreign-shm")
zf.writestr("state.db-journal", b"foreign-journal")
run_import(Namespace(zipfile=str(zip_path), force=True))
for suffix, payload in (
("-wal", b"foreign-wal-from-archive"),
("-shm", b"foreign-shm"),
("-journal", b"foreign-journal"),
):
sidecar = live_db.with_name("state.db" + suffix)
assert not sidecar.exists() or sidecar.read_bytes() != payload, suffix
assert _count_rows(live_db) == (2, 4)
def test_missing_target_takes_the_plain_publish(self, tmp_path, monkeypatch):
"""A fresh install has no inode to preserve; the member still lands."""
from hermes_cli.backup import run_import
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
staged = tmp_path / "backup-state.db"
_write_session_db(staged, 2, 3)
zip_path = tmp_path / "backup.zip"
self._zip_with_db(zip_path, staged)
run_import(Namespace(zipfile=str(zip_path), force=True))
assert _count_rows(home / "state.db") == (2, 6)
def _count_rows(db_path: Path) -> tuple[int, int]:
conn = sqlite3.connect(str(db_path))
try:
return (
conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0],
conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0],
)
finally:
conn.close()
def test_run_backup_prunes_older_default_named_zips_but_not_others(tmp_path, monkeypatch):
"""Hourly `hermes backup` callers accumulated 150+ zips; --keep bounds the default-named
ones and leaves custom-named or foreign zips alone (#81317)."""
from argparse import Namespace
from hermes_cli import backup as backup_mod
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text("model: x\n")
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
for i in range(4):
(tmp_path / f"hermes-backup-2026-01-0{i + 1}-000000.zip").write_bytes(b"old")
(tmp_path / "my-archive.zip").write_bytes(b"mine")
backup_mod.run_backup(Namespace(output=None, keep=2))
kept = sorted(p.name for p in tmp_path.glob("hermes-backup-*.zip"))
assert len(kept) == 2 and kept[0] == "hermes-backup-2026-01-04-000000.zip"
assert (tmp_path / "my-archive.zip").exists()
def test_import_restores_the_session_store_with_its_message_uids(tmp_path, monkeypatch):
"""A backup ships state.db as a SQLite snapshot and an import puts it back byte-for-byte: the durable
message ids come back with the rows."""
from hermes_state import SessionDB
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_make_hermes_tree(hermes_home)
db = SessionDB(db_path=hermes_home / "state.db")
try:
db.create_session("s", "cli", model="m")
db.append_message(session_id="s", role="user", content="q")
db.append_message(session_id="s", role="assistant", content="a")
uids = [m["message_uid"] for m in db.get_messages_as_conversation("s")]
finally:
db.close()
assert len(uids) == 2 and all(len(u) == 32 for u in uids)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
from hermes_cli.backup import run_backup, run_import
out_zip = tmp_path / "backup.zip"
run_backup(Namespace(output=str(out_zip)))
for name in ("state.db", "state.db-wal", "state.db-shm"):
(hermes_home / name).unlink(missing_ok=True)
assert run_import(Namespace(zipfile=str(out_zip), force=True)) is None
restored = SessionDB(db_path=hermes_home / "state.db")
try:
assert [m["message_uid"] for m in restored.get_messages_as_conversation("s")] == uids
finally:
restored.close()
Reference in New Issue View Git Blame Copy Permalink