"""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//``; 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, "cache/generated/images/x.png": True, "profiles/sage/cache/generated/videos/v.mp4": 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(" 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//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_negative_keep_is_rejected_by_backup_parser_and_snapshot_prune(self, capsys): """A negative keep slices away the NEWEST archives/snapshots, so both the ``backup --keep`` parser and ``/snapshot prune N`` refuse it.""" import argparse from hermes_cli.cli_commands_mixin import CLICommandsMixin from hermes_cli.subcommands.backup import build_backup_parser parser = argparse.ArgumentParser() build_backup_parser(parser.add_subparsers(dest="command"), cmd_backup=lambda args: None) for bad in ("-1", "x"): with pytest.raises(SystemExit) as exc: parser.parse_args(["backup", "--keep", bad]) assert exc.value.code == 2 assert [parser.parse_args(["backup", *a]).keep for a in (["--keep", "0"], ["-k", "1"], [])] == [0, 1, 3] with patch("hermes_cli.backup.prune_quick_snapshots") as prune: CLICommandsMixin._snapshot_prune(object(), ["/snapshot", "prune", "-1"]) prune.assert_not_called() CLICommandsMixin._snapshot_prune(object(), ["/snapshot", "prune", "2"]) prune.assert_called_once_with(keep=2) 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 /kanban/boards//kanban.db, not /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 # --------------------------------------------------------------------------- # Cron agent-job prompt field degradation across update (issue #82990) # --------------------------------------------------------------------------- class TestRestoreCronPromptFieldsIfDegraded: """A writer in the update's mutation window replaced every agent-job prompt with the job's own name while the count stayed identical — the count-based net passed it undetected. `restore_cron_prompt_fields_if_degraded` is the field-level net.""" @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_prompts_when_count_unchanged_but_clobbered(self, tmp_path): """The reported incident: 6 jobs before, 6 after, every prompt == name.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [ {"id": f"job-{i}", "name": f"Morning brief {i}", "prompt": f"Read the news and write brief {i}."} for i in range(6) ]) snap_id = self._make_snapshot(hermes_home) assert snap_id # The mutation window clobbers prompts to the name; count unchanged. self._seed_jobs(jobs_path, [ {"id": f"job-{i}", "name": f"Morning brief {i}", "prompt": f"Morning brief {i}"} for i in range(6) ]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is not None assert result["restored"] is True assert result["prompts"] == 6 assert result["snapshot_id"] == snap_id restored = json.loads(jobs_path.read_text()) for i, job in enumerate(restored["jobs"]): assert job["prompt"] == f"Read the news and write brief {i}." # Only the prompt was touched — name still the live one. assert job["name"] == f"Morning brief {i}" def test_blank_prompt_is_restored_too(self, tmp_path): """Missing/blank prompt on a live agent job is equally unusable.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [{"id": "a", "name": "Cleanup", "prompt": "Clean the tmp dir."}]) snap_id = self._make_snapshot(hermes_home) assert snap_id self._seed_jobs(jobs_path, [{"id": "a", "name": "Cleanup", "prompt": ""}]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is not None assert result["prompts"] == 1 assert json.loads(jobs_path.read_text())["jobs"][0]["prompt"] == "Clean the tmp dir." def test_no_agent_jobs_are_untouched(self, tmp_path): """Script jobs have no prompt — never restored, never stomped.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [ {"id": "agent", "name": "Brief", "prompt": "Write the brief."}, {"id": "script", "name": "Backup", "no_agent": True, "script": "echo hi", "prompt": "Backup"}, ]) snap_id = self._make_snapshot(hermes_home) assert snap_id self._seed_jobs(jobs_path, [ {"id": "agent", "name": "Brief", "prompt": "Brief"}, {"id": "script", "name": "Backup", "no_agent": True, "script": "echo hi", "prompt": "Backup"}, ]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is not None assert result["prompts"] == 1 assert result["job_ids"] == ["agent"] jobs = json.loads(jobs_path.read_text())["jobs"] assert jobs[0]["prompt"] == "Write the brief." assert jobs[1]["prompt"] == "Backup" # untouched def test_a_legitimate_user_edit_is_not_stomped(self, tmp_path): """Only prompt==name or blank restores — a genuinely different prompt is the user's edit during the window, not degradation.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [{"id": "a", "name": "Old", "prompt": "Old prompt text."}]) snap_id = self._make_snapshot(hermes_home) assert snap_id self._seed_jobs(jobs_path, [{"id": "a", "name": "New name", "prompt": "Deliberately rewritten prompt."}]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is None assert json.loads(jobs_path.read_text())["jobs"][0]["prompt"] == "Deliberately rewritten prompt." def test_a_legit_prompt_equal_to_name_stays(self, tmp_path): """A user may legitimately set prompt == name; a blank SNAPSHOT prompt means the snapshot has nothing better — restore nothing.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" # Snapshot job had a blank prompt (hand-edited oddity). self._seed_jobs(jobs_path, [{"id": "a", "name": "Ping", "prompt": ""}]) snap_id = self._make_snapshot(hermes_home) assert snap_id # Live prompt collapsed to the name — but the snapshot has nothing to give back. self._seed_jobs(jobs_path, [{"id": "a", "name": "Ping", "prompt": "Ping"}]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is None def test_unknown_job_ids_are_left_alone(self, tmp_path): """Jobs the snapshot does not know are new since the snapshot — theirs.""" from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [{"id": "a", "name": "A", "prompt": "A prompt."}]) snap_id = self._make_snapshot(hermes_home) assert snap_id self._seed_jobs(jobs_path, [ {"id": "a", "name": "A", "prompt": "A prompt."}, {"id": "new", "name": "New", "prompt": "New"}, # created during the window ]) result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is None def test_healthy_file_is_untouched(self, tmp_path): from hermes_cli.backup import restore_cron_prompt_fields_if_degraded hermes_home = tmp_path / ".hermes" jobs_path = hermes_home / "cron" / "jobs.json" self._seed_jobs(jobs_path, [{"id": "a", "name": "A", "prompt": "A prompt."}]) snap_id = self._make_snapshot(hermes_home) assert snap_id before = jobs_path.read_text() result = restore_cron_prompt_fields_if_degraded(snap_id, hermes_home=hermes_home) assert result is None assert jobs_path.read_text() == before # --------------------------------------------------------------------------- # 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 ~/, 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 (/profiles/) back to : 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=/profiles/ must restore INTO the profile, not into (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()