mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
feat: add hermes backup and hermes import commands (#7997)
* feat: add `hermes backup` and `hermes import` commands hermes backup — creates a zip of ~/.hermes/ (config, skills, sessions, profiles, memories, skins, cron jobs, etc.) excluding the hermes-agent codebase, __pycache__, and runtime PID files. Defaults to ~/hermes-backup-<timestamp>.zip, customizable with -o. hermes import <zipfile> — restores from a backup zip, validating it looks like a hermes backup before extracting. Handles .hermes/ prefix stripping, path traversal protection, and confirmation prompts (skip with --force). 29 tests covering exclusion rules, backup creation, import validation, prefix detection, path traversal blocking, confirmation flow, and a full round-trip test. * test: improve backup/import coverage to 97% Add 17 additional tests covering: - _format_size helper (bytes through terabytes) - Nonexistent hermes home error exit - Output path is a directory (auto-names inside it) - Output without .zip suffix (auto-appends) - Empty hermes home (all files excluded) - Permission errors during backup and import - Output zip inside hermes root (skips itself) - Not-a-zip file rejection - EOFError and KeyboardInterrupt during confirmation - 500+ file progress display - Directory-only zip prefix detection Remove dead code branch in _detect_prefix (unreachable guard). * feat: auto-restore profile wrapper scripts on import After extracting backup files, hermes import now scans profiles/ for subdirectories with config.yaml or .env and recreates the ~/.local/bin wrapper scripts so profile aliases (e.g. 'coder chat') work immediately. Also prints guidance for re-installing gateway services per profile. Handles edge cases: - Skips profile dirs without config (not real profiles) - Skips aliases that collide with existing commands - Gracefully degrades if hermes_cli.profiles isn't available (fresh install) - Shows PATH hint if ~/.local/bin isn't in PATH 3 new profile restoration tests (49 total).
This commit is contained in:
Teknium
GitHub
parent
50d86b3c71
commit
fa7cd44b92
3 changed files with 1345 additions and 1 deletions
399
hermes_cli/backup.py
Normal file
399
hermes_cli/backup.py
Normal file
|
|
@ -0,0 +1,399 @@
|
|||
"""
|
||||
Backup and import commands for hermes CLI.
|
||||
|
||||
`hermes backup` creates a zip archive of the entire ~/.hermes/ directory
|
||||
(excluding the hermes-agent repo and transient files).
|
||||
|
||||
`hermes import` restores from a backup zip, overlaying onto the current
|
||||
HERMES_HOME root.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from hermes_constants import get_default_hermes_root, display_hermes_home
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Exclusion rules
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Directory names to skip entirely (matched against each path component)
|
||||
_EXCLUDED_DIRS = {
|
||||
"hermes-agent", # the codebase repo — re-clone instead
|
||||
"__pycache__", # bytecode caches — regenerated on import
|
||||
".git", # nested git dirs (profiles shouldn't have these, but safety)
|
||||
"node_modules", # js deps if website/ somehow leaks in
|
||||
}
|
||||
|
||||
# File-name suffixes to skip
|
||||
_EXCLUDED_SUFFIXES = (
|
||||
".pyc",
|
||||
".pyo",
|
||||
)
|
||||
|
||||
# File names to skip (runtime state that's meaningless on another machine)
|
||||
_EXCLUDED_NAMES = {
|
||||
"gateway.pid",
|
||||
"cron.pid",
|
||||
}
|
||||
|
||||
|
||||
def _should_exclude(rel_path: Path) -> bool:
|
||||
"""Return True if *rel_path* (relative to hermes root) should be skipped."""
|
||||
parts = rel_path.parts
|
||||
|
||||
# Any path component matches an excluded dir name
|
||||
for part in parts:
|
||||
if part in _EXCLUDED_DIRS:
|
||||
return True
|
||||
|
||||
name = rel_path.name
|
||||
|
||||
if name in _EXCLUDED_NAMES:
|
||||
return True
|
||||
|
||||
if name.endswith(_EXCLUDED_SUFFIXES):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _format_size(nbytes: int) -> str:
|
||||
"""Human-readable file size."""
|
||||
for unit in ("B", "KB", "MB", "GB"):
|
||||
if nbytes < 1024:
|
||||
return f"{nbytes:.1f} {unit}" if unit != "B" else f"{nbytes} {unit}"
|
||||
nbytes /= 1024
|
||||
return f"{nbytes:.1f} TB"
|
||||
|
||||
|
||||
def run_backup(args) -> None:
|
||||
"""Create a zip backup of the Hermes home directory."""
|
||||
hermes_root = get_default_hermes_root()
|
||||
|
||||
if not hermes_root.is_dir():
|
||||
print(f"Error: Hermes home directory not found at {hermes_root}")
|
||||
sys.exit(1)
|
||||
|
||||
# Determine output path
|
||||
if args.output:
|
||||
out_path = Path(args.output).expanduser().resolve()
|
||||
# If user gave a directory, put the zip inside it
|
||||
if out_path.is_dir():
|
||||
stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
|
||||
out_path = out_path / f"hermes-backup-{stamp}.zip"
|
||||
else:
|
||||
stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
|
||||
out_path = Path.home() / f"hermes-backup-{stamp}.zip"
|
||||
|
||||
# Ensure the suffix is .zip
|
||||
if out_path.suffix.lower() != ".zip":
|
||||
out_path = out_path.with_suffix(out_path.suffix + ".zip")
|
||||
|
||||
# Ensure parent directory exists
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Collect files
|
||||
print(f"Scanning {display_hermes_home()} ...")
|
||||
files_to_add: list[tuple[Path, Path]] = [] # (absolute, relative)
|
||||
skipped_dirs = set()
|
||||
|
||||
for dirpath, dirnames, filenames in os.walk(hermes_root, followlinks=False):
|
||||
dp = Path(dirpath)
|
||||
rel_dir = dp.relative_to(hermes_root)
|
||||
|
||||
# Prune excluded directories in-place so os.walk doesn't descend
|
||||
orig_dirnames = dirnames[:]
|
||||
dirnames[:] = [
|
||||
d for d in dirnames
|
||||
if d not in _EXCLUDED_DIRS
|
||||
]
|
||||
for removed in set(orig_dirnames) - set(dirnames):
|
||||
skipped_dirs.add(str(rel_dir / removed))
|
||||
|
||||
for fname in filenames:
|
||||
fpath = dp / fname
|
||||
rel = fpath.relative_to(hermes_root)
|
||||
|
||||
if _should_exclude(rel):
|
||||
continue
|
||||
|
||||
# Skip the output zip itself if it happens to be inside hermes root
|
||||
try:
|
||||
if fpath.resolve() == out_path.resolve():
|
||||
continue
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
|
||||
files_to_add.append((fpath, rel))
|
||||
|
||||
if not files_to_add:
|
||||
print("No files to back up.")
|
||||
return
|
||||
|
||||
# Create the zip
|
||||
file_count = len(files_to_add)
|
||||
print(f"Backing up {file_count} files ...")
|
||||
|
||||
total_bytes = 0
|
||||
errors = []
|
||||
t0 = time.monotonic()
|
||||
|
||||
with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf:
|
||||
for i, (abs_path, rel_path) in enumerate(files_to_add, 1):
|
||||
try:
|
||||
zf.write(abs_path, arcname=str(rel_path))
|
||||
total_bytes += abs_path.stat().st_size
|
||||
except (PermissionError, OSError) as exc:
|
||||
errors.append(f" {rel_path}: {exc}")
|
||||
continue
|
||||
|
||||
# Progress every 500 files
|
||||
if i % 500 == 0:
|
||||
print(f" {i}/{file_count} files ...")
|
||||
|
||||
elapsed = time.monotonic() - t0
|
||||
zip_size = out_path.stat().st_size
|
||||
|
||||
# Summary
|
||||
print()
|
||||
print(f"Backup complete: {out_path}")
|
||||
print(f" Files: {file_count}")
|
||||
print(f" Original: {_format_size(total_bytes)}")
|
||||
print(f" Compressed: {_format_size(zip_size)}")
|
||||
print(f" Time: {elapsed:.1f}s")
|
||||
|
||||
if skipped_dirs:
|
||||
print(f"\n Excluded directories:")
|
||||
for d in sorted(skipped_dirs):
|
||||
print(f" {d}/")
|
||||
|
||||
if errors:
|
||||
print(f"\n Warnings ({len(errors)} files skipped):")
|
||||
for e in errors[:10]:
|
||||
print(e)
|
||||
if len(errors) > 10:
|
||||
print(f" ... and {len(errors) - 10} more")
|
||||
|
||||
print(f"\nRestore with: hermes import {out_path.name}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Import
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _validate_backup_zip(zf: zipfile.ZipFile) -> tuple[bool, str]:
|
||||
"""Check that a zip looks like a Hermes backup.
|
||||
|
||||
Returns (ok, reason).
|
||||
"""
|
||||
names = zf.namelist()
|
||||
if not names:
|
||||
return False, "zip archive is empty"
|
||||
|
||||
# Look for telltale files that a hermes home would have
|
||||
markers = {"config.yaml", ".env", "hermes_state.db", "memory_store.db"}
|
||||
found = set()
|
||||
for n in names:
|
||||
# Could be at the root or one level deep (if someone zipped the directory)
|
||||
basename = Path(n).name
|
||||
if basename in markers:
|
||||
found.add(basename)
|
||||
|
||||
if not found:
|
||||
return False, (
|
||||
"zip does not appear to be a Hermes backup "
|
||||
"(no config.yaml, .env, or state databases found)"
|
||||
)
|
||||
|
||||
return True, ""
|
||||
|
||||
|
||||
def _detect_prefix(zf: zipfile.ZipFile) -> str:
|
||||
"""Detect if the zip has a common directory prefix wrapping all entries.
|
||||
|
||||
Some tools zip as `.hermes/config.yaml` instead of `config.yaml`.
|
||||
Returns the prefix to strip (empty string if none).
|
||||
"""
|
||||
names = [n for n in zf.namelist() if not n.endswith("/")]
|
||||
if not names:
|
||||
return ""
|
||||
|
||||
# Find common prefix
|
||||
parts_list = [Path(n).parts for n in names]
|
||||
|
||||
# Check if all entries share a common first directory
|
||||
first_parts = {p[0] for p in parts_list if len(p) > 1}
|
||||
if len(first_parts) == 1:
|
||||
prefix = first_parts.pop()
|
||||
# Only strip if it looks like a hermes dir name
|
||||
if prefix in (".hermes", "hermes"):
|
||||
return prefix + "/"
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def run_import(args) -> None:
|
||||
"""Restore a Hermes backup from a zip file."""
|
||||
zip_path = Path(args.zipfile).expanduser().resolve()
|
||||
|
||||
if not zip_path.is_file():
|
||||
print(f"Error: File not found: {zip_path}")
|
||||
sys.exit(1)
|
||||
|
||||
if not zipfile.is_zipfile(zip_path):
|
||||
print(f"Error: Not a valid zip file: {zip_path}")
|
||||
sys.exit(1)
|
||||
|
||||
hermes_root = get_default_hermes_root()
|
||||
|
||||
with zipfile.ZipFile(zip_path, "r") as zf:
|
||||
# Validate
|
||||
ok, reason = _validate_backup_zip(zf)
|
||||
if not ok:
|
||||
print(f"Error: {reason}")
|
||||
sys.exit(1)
|
||||
|
||||
prefix = _detect_prefix(zf)
|
||||
members = [n for n in zf.namelist() if not n.endswith("/")]
|
||||
file_count = len(members)
|
||||
|
||||
print(f"Backup contains {file_count} files")
|
||||
print(f"Target: {display_hermes_home()}")
|
||||
|
||||
if prefix:
|
||||
print(f"Detected archive prefix: {prefix!r} (will be stripped)")
|
||||
|
||||
# Check for existing installation
|
||||
has_config = (hermes_root / "config.yaml").exists()
|
||||
has_env = (hermes_root / ".env").exists()
|
||||
|
||||
if (has_config or has_env) and not args.force:
|
||||
print()
|
||||
print("Warning: Target directory already has Hermes configuration.")
|
||||
print("Importing will overwrite existing files with backup contents.")
|
||||
print()
|
||||
try:
|
||||
answer = input("Continue? [y/N] ").strip().lower()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nAborted.")
|
||||
sys.exit(1)
|
||||
if answer not in ("y", "yes"):
|
||||
print("Aborted.")
|
||||
return
|
||||
|
||||
# Extract
|
||||
print(f"\nImporting {file_count} files ...")
|
||||
hermes_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
errors = []
|
||||
restored = 0
|
||||
t0 = time.monotonic()
|
||||
|
||||
for member in members:
|
||||
# Strip prefix if detected
|
||||
if prefix and member.startswith(prefix):
|
||||
rel = member[len(prefix):]
|
||||
else:
|
||||
rel = member
|
||||
|
||||
if not rel:
|
||||
continue
|
||||
|
||||
target = hermes_root / rel
|
||||
|
||||
# Security: reject absolute paths and traversals
|
||||
try:
|
||||
target.resolve().relative_to(hermes_root.resolve())
|
||||
except ValueError:
|
||||
errors.append(f" {rel}: path traversal blocked")
|
||||
continue
|
||||
|
||||
try:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
with zf.open(member) as src, open(target, "wb") as dst:
|
||||
dst.write(src.read())
|
||||
restored += 1
|
||||
except (PermissionError, OSError) as exc:
|
||||
errors.append(f" {rel}: {exc}")
|
||||
|
||||
if restored % 500 == 0:
|
||||
print(f" {restored}/{file_count} files ...")
|
||||
|
||||
elapsed = time.monotonic() - t0
|
||||
|
||||
# Summary
|
||||
print()
|
||||
print(f"Import complete: {restored} files restored in {elapsed:.1f}s")
|
||||
print(f" Target: {display_hermes_home()}")
|
||||
|
||||
if errors:
|
||||
print(f"\n Warnings ({len(errors)} files skipped):")
|
||||
for e in errors[:10]:
|
||||
print(e)
|
||||
if len(errors) > 10:
|
||||
print(f" ... and {len(errors) - 10} more")
|
||||
|
||||
# Post-import: restore profile wrapper scripts
|
||||
profiles_dir = hermes_root / "profiles"
|
||||
restored_profiles = []
|
||||
if profiles_dir.is_dir():
|
||||
try:
|
||||
from hermes_cli.profiles import (
|
||||
create_wrapper_script, check_alias_collision,
|
||||
_is_wrapper_dir_in_path, _get_wrapper_dir,
|
||||
)
|
||||
for entry in sorted(profiles_dir.iterdir()):
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
profile_name = entry.name
|
||||
# Only create wrappers for directories with config
|
||||
if not (entry / "config.yaml").exists() and not (entry / ".env").exists():
|
||||
continue
|
||||
collision = check_alias_collision(profile_name)
|
||||
if collision:
|
||||
print(f" Skipped alias '{profile_name}': {collision}")
|
||||
restored_profiles.append((profile_name, False))
|
||||
else:
|
||||
wrapper = create_wrapper_script(profile_name)
|
||||
restored_profiles.append((profile_name, wrapper is not None))
|
||||
|
||||
if restored_profiles:
|
||||
created = [n for n, ok in restored_profiles if ok]
|
||||
skipped = [n for n, ok in restored_profiles if not ok]
|
||||
if created:
|
||||
print(f"\n Profile aliases restored: {', '.join(created)}")
|
||||
if skipped:
|
||||
print(f" Profile aliases skipped: {', '.join(skipped)}")
|
||||
if not _is_wrapper_dir_in_path():
|
||||
print(f"\n Note: {_get_wrapper_dir()} is not in your PATH.")
|
||||
print(' Add to your shell config (~/.bashrc or ~/.zshrc):')
|
||||
print(' export PATH="$HOME/.local/bin:$PATH"')
|
||||
except ImportError:
|
||||
# hermes_cli.profiles might not be available (fresh install)
|
||||
if any(profiles_dir.iterdir()):
|
||||
print(f"\n Profiles detected but aliases could not be created.")
|
||||
print(f" Run: hermes profile list (after installing hermes)")
|
||||
|
||||
# Guidance
|
||||
print()
|
||||
if not (hermes_root / "hermes-agent").is_dir():
|
||||
print("Note: The hermes-agent codebase was not included in the backup.")
|
||||
print(" If this is a fresh install, run: hermes update")
|
||||
|
||||
if restored_profiles:
|
||||
gw_profiles = [n for n, _ in restored_profiles]
|
||||
print("\nTo re-enable gateway services for profiles:")
|
||||
for pname in gw_profiles:
|
||||
print(f" hermes -p {pname} gateway install")
|
||||
|
||||
print("Done. Your Hermes configuration has been restored.")
|
||||
|
|
@ -2818,6 +2818,18 @@ def cmd_config(args):
|
|||
config_command(args)
|
||||
|
||||
|
||||
def cmd_backup(args):
|
||||
"""Back up Hermes home directory to a zip file."""
|
||||
from hermes_cli.backup import run_backup
|
||||
run_backup(args)
|
||||
|
||||
|
||||
def cmd_import(args):
|
||||
"""Restore a Hermes backup from a zip file."""
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
|
||||
def cmd_version(args):
|
||||
"""Show version."""
|
||||
print(f"Hermes Agent v{__version__} ({__release_date__})")
|
||||
|
|
@ -4904,7 +4916,43 @@ For more help on a command:
|
|||
help="Show redacted API key prefixes (first/last 4 chars) instead of just set/not set"
|
||||
)
|
||||
dump_parser.set_defaults(func=cmd_dump)
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# backup command
|
||||
# =========================================================================
|
||||
backup_parser = subparsers.add_parser(
|
||||
"backup",
|
||||
help="Back up Hermes home directory to a zip file",
|
||||
description="Create a zip archive of your entire Hermes configuration, "
|
||||
"skills, sessions, and data (excludes the hermes-agent codebase)"
|
||||
)
|
||||
backup_parser.add_argument(
|
||||
"-o", "--output",
|
||||
help="Output path for the zip file (default: ~/hermes-backup-<timestamp>.zip)"
|
||||
)
|
||||
backup_parser.set_defaults(func=cmd_backup)
|
||||
|
||||
# =========================================================================
|
||||
# import command
|
||||
# =========================================================================
|
||||
import_parser = subparsers.add_parser(
|
||||
"import",
|
||||
help="Restore a Hermes backup from a zip file",
|
||||
description="Extract a previously created Hermes backup into your "
|
||||
"Hermes home directory, restoring configuration, skills, "
|
||||
"sessions, and data"
|
||||
)
|
||||
import_parser.add_argument(
|
||||
"zipfile",
|
||||
help="Path to the backup zip file"
|
||||
)
|
||||
import_parser.add_argument(
|
||||
"--force", "-f",
|
||||
action="store_true",
|
||||
help="Overwrite existing files without confirmation"
|
||||
)
|
||||
import_parser.set_defaults(func=cmd_import)
|
||||
|
||||
# =========================================================================
|
||||
# config command
|
||||
# =========================================================================
|
||||
|
|
|
|||
897
tests/hermes_cli/test_backup.py
Normal file
897
tests/hermes_cli/test_backup.py
Normal file
|
|
@ -0,0 +1,897 @@
|
|||
"""Tests for hermes backup and import commands."""
|
||||
|
||||
import os
|
||||
import zipfile
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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")
|
||||
(root / "memory_store.db").write_bytes(b"fake-sqlite")
|
||||
(root / "hermes_state.db").write_bytes(b"fake-state")
|
||||
|
||||
# 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")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _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_pycache(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert _should_exclude(Path("plugins/__pycache__/mod.cpython-312.pyc"))
|
||||
|
||||
def test_excludes_pyc_files(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert _should_exclude(Path("some/module.pyc"))
|
||||
|
||||
def test_excludes_pid_files(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert _should_exclude(Path("gateway.pid"))
|
||||
assert _should_exclude(Path("cron.pid"))
|
||||
|
||||
def test_includes_config(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path("config.yaml"))
|
||||
|
||||
def test_includes_env(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path(".env"))
|
||||
|
||||
def test_includes_skills(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path("skills/my-skill/SKILL.md"))
|
||||
|
||||
def test_includes_profiles(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path("profiles/coder/config.yaml"))
|
||||
|
||||
def test_includes_sessions(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path("sessions/abc.json"))
|
||||
|
||||
def test_includes_logs(self):
|
||||
from hermes_cli.backup import _should_exclude
|
||||
assert not _should_exclude(Path("logs/agent.log"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backup tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestBackup:
|
||||
def test_creates_zip(self, tmp_path, monkeypatch):
|
||||
"""Backup creates a valid zip containing expected files."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
_make_hermes_tree(hermes_home)
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
# get_default_hermes_root needs this
|
||||
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)
|
||||
|
||||
assert out_zip.exists()
|
||||
with zipfile.ZipFile(out_zip, "r") as zf:
|
||||
names = zf.namelist()
|
||||
# Config should be present
|
||||
assert "config.yaml" in names
|
||||
assert ".env" in names
|
||||
# Skills
|
||||
assert "skills/my-skill/SKILL.md" in names
|
||||
# Profiles
|
||||
assert "profiles/coder/config.yaml" in names
|
||||
assert "profiles/coder/.env" in names
|
||||
# Sessions
|
||||
assert "sessions/abc123.json" in names
|
||||
# Logs
|
||||
assert "logs/agent.log" in names
|
||||
# Skins
|
||||
assert "skins/cyber.yaml" in names
|
||||
|
||||
def test_excludes_hermes_agent(self, tmp_path, monkeypatch):
|
||||
"""Backup does NOT include hermes-agent/ directory."""
|
||||
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 = 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()
|
||||
agent_files = [n for n in names if "hermes-agent" in n]
|
||||
assert agent_files == [], f"hermes-agent files leaked into backup: {agent_files}"
|
||||
|
||||
def test_excludes_pycache(self, tmp_path, monkeypatch):
|
||||
"""Backup does NOT include __pycache__ dirs."""
|
||||
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 = 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()
|
||||
pycache_files = [n for n in names if "__pycache__" in n]
|
||||
assert pycache_files == []
|
||||
|
||||
def test_excludes_pid_files(self, tmp_path, monkeypatch):
|
||||
"""Backup does NOT include PID files."""
|
||||
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 = 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()
|
||||
pid_files = [n for n in names if n.endswith(".pid")]
|
||||
assert pid_files == []
|
||||
|
||||
def test_default_output_path(self, tmp_path, monkeypatch):
|
||||
"""When no output path given, zip goes to ~/hermes-backup-*.zip."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("model: test\n")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
args = Namespace(output=None)
|
||||
|
||||
from hermes_cli.backup import run_backup
|
||||
run_backup(args)
|
||||
|
||||
# Should exist in home dir
|
||||
zips = list(tmp_path.glob("hermes-backup-*.zip"))
|
||||
assert len(zips) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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_restores_files(self, tmp_path, monkeypatch):
|
||||
"""Import extracts files into hermes home."""
|
||||
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:\n provider: openrouter\n",
|
||||
".env": "OPENROUTER_API_KEY=sk-test\n",
|
||||
"skills/my-skill/SKILL.md": "# My Skill\n",
|
||||
"profiles/coder/config.yaml": "model:\n provider: anthropic\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
assert (hermes_home / "config.yaml").read_text() == "model:\n provider: openrouter\n"
|
||||
assert (hermes_home / ".env").read_text() == "OPENROUTER_API_KEY=sk-test\n"
|
||||
assert (hermes_home / "skills" / "my-skill" / "SKILL.md").read_text() == "# My Skill\n"
|
||||
assert (hermes_home / "profiles" / "coder" / "config.yaml").exists()
|
||||
|
||||
def test_strips_hermes_prefix(self, tmp_path, monkeypatch):
|
||||
"""Import strips .hermes/ prefix if all entries share it."""
|
||||
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, {
|
||||
".hermes/config.yaml": "model: test\n",
|
||||
".hermes/skills/a/SKILL.md": "# A\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
assert (hermes_home / "config.yaml").read_text() == "model: test\n"
|
||||
assert (hermes_home / "skills" / "a" / "SKILL.md").read_text() == "# A\n"
|
||||
|
||||
def test_rejects_empty_zip(self, tmp_path, monkeypatch):
|
||||
"""Import rejects an empty zip."""
|
||||
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 / "empty.zip"
|
||||
with zipfile.ZipFile(zip_path, "w"):
|
||||
pass # empty
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with pytest.raises(SystemExit):
|
||||
run_import(args)
|
||||
|
||||
def test_rejects_non_hermes_zip(self, tmp_path, monkeypatch):
|
||||
"""Import rejects a zip that doesn't look like a hermes backup."""
|
||||
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 / "random.zip"
|
||||
self._make_backup_zip(zip_path, {
|
||||
"some/random/file.txt": "hello",
|
||||
"another/thing.json": "{}",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with pytest.raises(SystemExit):
|
||||
run_import(args)
|
||||
|
||||
def test_blocks_path_traversal(self, tmp_path, monkeypatch):
|
||||
"""Import blocks zip entries with path traversal."""
|
||||
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 / "evil.zip"
|
||||
# Include a marker file so validation passes
|
||||
self._make_backup_zip(zip_path, {
|
||||
"config.yaml": "model: test\n",
|
||||
"../../etc/passwd": "root:x:0:0\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
# config.yaml should be restored
|
||||
assert (hermes_home / "config.yaml").exists()
|
||||
# traversal file should NOT exist outside hermes home
|
||||
assert not (tmp_path / "etc" / "passwd").exists()
|
||||
|
||||
def test_confirmation_prompt_abort(self, tmp_path, monkeypatch):
|
||||
"""Import aborts when user says no to confirmation."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
# Pre-existing config triggers the confirmation
|
||||
(hermes_home / "config.yaml").write_text("existing: true\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": "model: restored\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=False)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with patch("builtins.input", return_value="n"):
|
||||
run_import(args)
|
||||
|
||||
# Original config should be unchanged
|
||||
assert (hermes_home / "config.yaml").read_text() == "existing: true\n"
|
||||
|
||||
def test_force_skips_confirmation(self, tmp_path, monkeypatch):
|
||||
"""Import with --force skips confirmation and overwrites."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("existing: true\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": "model: restored\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
assert (hermes_home / "config.yaml").read_text() == "model: restored\n"
|
||||
|
||||
def test_missing_file_exits(self, tmp_path, monkeypatch):
|
||||
"""Import exits with error for nonexistent file."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
args = Namespace(zipfile=str(tmp_path / "nonexistent.zip"), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with pytest.raises(SystemExit):
|
||||
run_import(args)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 TestFormatSize:
|
||||
def test_bytes(self):
|
||||
from hermes_cli.backup import _format_size
|
||||
assert _format_size(512) == "512 B"
|
||||
|
||||
def test_kilobytes(self):
|
||||
from hermes_cli.backup import _format_size
|
||||
assert "KB" in _format_size(2048)
|
||||
|
||||
def test_megabytes(self):
|
||||
from hermes_cli.backup import _format_size
|
||||
assert "MB" in _format_size(5 * 1024 * 1024)
|
||||
|
||||
def test_gigabytes(self):
|
||||
from hermes_cli.backup import _format_size
|
||||
assert "GB" in _format_size(3 * 1024 ** 3)
|
||||
|
||||
def test_terabytes(self):
|
||||
from hermes_cli.backup import _format_size
|
||||
assert "TB" in _format_size(2 * 1024 ** 4)
|
||||
|
||||
|
||||
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_validate_with_env(self):
|
||||
"""Zip with .env passes validation."""
|
||||
import io
|
||||
from hermes_cli.backup import _validate_backup_zip
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr(".env", "KEY=val")
|
||||
buf.seek(0)
|
||||
with zipfile.ZipFile(buf, "r") as zf:
|
||||
ok, reason = _validate_backup_zip(zf)
|
||||
assert ok
|
||||
|
||||
def test_validate_rejects_random(self):
|
||||
"""Zip without hermes markers fails validation."""
|
||||
import io
|
||||
from hermes_cli.backup import _validate_backup_zip
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr("random/file.txt", "hello")
|
||||
buf.seek(0)
|
||||
with zipfile.ZipFile(buf, "r") as zf:
|
||||
ok, reason = _validate_backup_zip(zf)
|
||||
assert not ok
|
||||
|
||||
def test_detect_prefix_hermes(self):
|
||||
"""Detects .hermes/ prefix wrapping all entries."""
|
||||
import io
|
||||
from hermes_cli.backup import _detect_prefix
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr(".hermes/config.yaml", "test")
|
||||
zf.writestr(".hermes/skills/a/SKILL.md", "skill")
|
||||
buf.seek(0)
|
||||
with zipfile.ZipFile(buf, "r") as zf:
|
||||
assert _detect_prefix(zf) == ".hermes/"
|
||||
|
||||
def test_detect_prefix_none(self):
|
||||
"""No prefix when entries are at root."""
|
||||
import io
|
||||
from hermes_cli.backup import _detect_prefix
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr("config.yaml", "test")
|
||||
zf.writestr("skills/a/SKILL.md", "skill")
|
||||
buf.seek(0)
|
||||
with zipfile.ZipFile(buf, "r") as zf:
|
||||
assert _detect_prefix(zf) == ""
|
||||
|
||||
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_nonexistent_hermes_home(self, tmp_path, monkeypatch):
|
||||
"""Backup exits when hermes home doesn't exist."""
|
||||
fake_home = tmp_path / "nonexistent" / ".hermes"
|
||||
monkeypatch.setenv("HERMES_HOME", str(fake_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path / "nonexistent")
|
||||
|
||||
args = Namespace(output=str(tmp_path / "out.zip"))
|
||||
|
||||
from hermes_cli.backup import run_backup
|
||||
with pytest.raises(SystemExit):
|
||||
run_backup(args)
|
||||
|
||||
def test_output_is_directory(self, tmp_path, monkeypatch):
|
||||
"""When output path is a directory, zip is created inside it."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("model: test\n")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
out_dir = tmp_path / "backups"
|
||||
out_dir.mkdir()
|
||||
|
||||
args = Namespace(output=str(out_dir))
|
||||
|
||||
from hermes_cli.backup import run_backup
|
||||
run_backup(args)
|
||||
|
||||
zips = list(out_dir.glob("hermes-backup-*.zip"))
|
||||
assert len(zips) == 1
|
||||
|
||||
def test_output_without_zip_suffix(self, tmp_path, monkeypatch):
|
||||
"""Output path without .zip gets suffix appended."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("model: test\n")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
out_path = tmp_path / "mybackup.tar"
|
||||
args = Namespace(output=str(out_path))
|
||||
|
||||
from hermes_cli.backup import run_backup
|
||||
run_backup(args)
|
||||
|
||||
# Should have .tar.zip suffix
|
||||
assert (tmp_path / "mybackup.tar.zip").exists()
|
||||
|
||||
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_permission_error_during_backup(self, tmp_path, monkeypatch):
|
||||
"""Backup handles permission errors gracefully."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("model: test\n")
|
||||
|
||||
# Create an unreadable file
|
||||
bad_file = hermes_home / "secret.db"
|
||||
bad_file.write_text("data")
|
||||
bad_file.chmod(0o000)
|
||||
|
||||
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
|
||||
try:
|
||||
run_backup(args)
|
||||
finally:
|
||||
# Restore permissions for cleanup
|
||||
bad_file.chmod(0o644)
|
||||
|
||||
# Zip should still be created with the readable files
|
||||
assert out_zip.exists()
|
||||
|
||||
def test_skips_output_zip_inside_hermes(self, tmp_path, monkeypatch):
|
||||
"""Backup skips its own output zip if it's inside hermes root."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / "config.yaml").write_text("model: test\n")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
# Output inside hermes home
|
||||
out_zip = hermes_home / "backup.zip"
|
||||
args = Namespace(output=str(out_zip))
|
||||
|
||||
from hermes_cli.backup import run_backup
|
||||
run_backup(args)
|
||||
|
||||
# The zip should exist but not contain itself
|
||||
assert out_zip.exists()
|
||||
with zipfile.ZipFile(out_zip, "r") as zf:
|
||||
assert "backup.zip" not in zf.namelist()
|
||||
|
||||
|
||||
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_not_a_zip(self, tmp_path, monkeypatch):
|
||||
"""Import rejects a non-zip file."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
not_zip = tmp_path / "fake.zip"
|
||||
not_zip.write_text("this is not a zip")
|
||||
|
||||
args = Namespace(zipfile=str(not_zip), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with pytest.raises(SystemExit):
|
||||
run_import(args)
|
||||
|
||||
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)
|
||||
|
||||
def test_keyboard_interrupt_during_confirmation(self, tmp_path, monkeypatch):
|
||||
"""Import handles KeyboardInterrupt during confirmation prompt."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / ".env").write_text("KEY=val\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=KeyboardInterrupt):
|
||||
with pytest.raises(SystemExit):
|
||||
run_import(args)
|
||||
|
||||
def test_permission_error_during_import(self, tmp_path, monkeypatch):
|
||||
"""Import handles permission errors during extraction."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
# Create a read-only directory so extraction fails
|
||||
locked_dir = hermes_home / "locked"
|
||||
locked_dir.mkdir()
|
||||
locked_dir.chmod(0o555)
|
||||
|
||||
zip_path = tmp_path / "backup.zip"
|
||||
self._make_backup_zip(zip_path, {
|
||||
"config.yaml": "model: test\n",
|
||||
"locked/secret.txt": "data",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
try:
|
||||
run_import(args)
|
||||
finally:
|
||||
locked_dir.chmod(0o755)
|
||||
|
||||
# config.yaml should still be restored despite the error
|
||||
assert (hermes_home / "config.yaml").exists()
|
||||
|
||||
def test_progress_with_many_files(self, tmp_path, monkeypatch):
|
||||
"""Import shows progress with 500+ files."""
|
||||
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 / "big.zip"
|
||||
files = {"config.yaml": "model: test\n"}
|
||||
for i in range(600):
|
||||
files[f"sessions/s{i:04d}.json"] = "{}"
|
||||
|
||||
self._make_backup_zip(zip_path, files)
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
assert (hermes_home / "config.yaml").exists()
|
||||
assert (hermes_home / "sessions" / "s0599.json").exists()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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_creates_profile_wrappers(self, tmp_path, monkeypatch):
|
||||
"""Import auto-creates wrapper scripts for restored profiles."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
|
||||
# Mock the wrapper dir to be inside 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:\n provider: openrouter\n",
|
||||
"profiles/coder/config.yaml": "model:\n provider: anthropic\n",
|
||||
"profiles/coder/.env": "ANTHROPIC_API_KEY=sk-test\n",
|
||||
"profiles/researcher/config.yaml": "model:\n provider: deepseek\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
run_import(args)
|
||||
|
||||
# Profile directories should exist
|
||||
assert (hermes_home / "profiles" / "coder" / "config.yaml").exists()
|
||||
assert (hermes_home / "profiles" / "researcher" / "config.yaml").exists()
|
||||
|
||||
# Wrapper scripts should be created
|
||||
assert (wrapper_dir / "coder").exists()
|
||||
assert (wrapper_dir / "researcher").exists()
|
||||
|
||||
# Wrappers should contain the right content
|
||||
coder_wrapper = (wrapper_dir / "coder").read_text()
|
||||
assert "hermes -p coder" in coder_wrapper
|
||||
|
||||
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()
|
||||
|
||||
def test_import_without_profiles_module(self, tmp_path, monkeypatch):
|
||||
"""Import gracefully handles missing profiles module (fresh install)."""
|
||||
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: test\n",
|
||||
"profiles/coder/config.yaml": "model: test\n",
|
||||
})
|
||||
|
||||
args = Namespace(zipfile=str(zip_path), force=True)
|
||||
|
||||
# Simulate profiles module not being available
|
||||
import hermes_cli.backup as backup_mod
|
||||
original_import = __builtins__.__import__ if hasattr(__builtins__, '__import__') else __import__
|
||||
|
||||
def fake_import(name, *a, **kw):
|
||||
if name == "hermes_cli.profiles":
|
||||
raise ImportError("no profiles module")
|
||||
return original_import(name, *a, **kw)
|
||||
|
||||
from hermes_cli.backup import run_import
|
||||
with patch("builtins.__import__", side_effect=fake_import):
|
||||
run_import(args)
|
||||
|
||||
# Files should still be restored even if wrappers can't be created
|
||||
assert (hermes_home / "profiles" / "coder" / "config.yaml").exists()
|
||||
Loading…
Add table
Add a link
Reference in a new issue