59624b7ace
The OS lanes are marker-driven: list_os_marked_tests.py picks the files
a lane imports from their platforms() specs and the lane selects with
-m platforms. A test gated with skipif(sys.platform != "win32") is
therefore never imported on the Windows lane and skipped everywhere else
— it runs on no host. skipif(sys.platform == "win32") tests were merely
invisible to the lane bookkeeping, but the rule the tree now follows is
one host marker, never a bare skipif.
Mechanical mapping, semantics preserved: skip-on-Windows → "posix",
skip-off-Windows → "windows", skip-off-Linux → "linux", skip-on-macOS →
"not macos". The former skip reasons stay as trailing comments. A
non-host condition (os.geteuid() == 0) stays a separate skipif beside
the marker, spelled getattr(os, "geteuid", ...) so the decorator still
imports on Windows.
Where the conversion would stack two platforms() marks on one test (the
conftest rejects that at collection) the narrower mark wins:
- test_update_wedged_gateway: the class is already platforms("linux");
its per-test "needs UNIX sockets" marks were redundant and are gone.
- test_process_registry.TestSystemdCgroupIsolation: the class-level
skip-on-Windows moves onto the 11 methods that had no host mark; the
11 platforms("linux") methods keep theirs.
- test_file_ops_single_roundtrip: the two fifo tests drop their
platforms("linux") in favour of the module's "posix" (mkfifo exists on
macOS; both tests already skip when it does not).
- test_linux_desktop_entry / test_gateway_job_teardown_live: duplicate
or wider marks removed.
142 lines
4.8 KiB
Python
142 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from hermes_cli.backup import (
|
|
BackupInProgressError,
|
|
_atomic_output_path,
|
|
_backup_operation_lock,
|
|
_write_full_zip_backup,
|
|
create_quick_snapshot,
|
|
list_quick_snapshots,
|
|
)
|
|
|
|
|
|
def test_backup_lock_rejects_a_second_operation(tmp_path) -> None:
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
|
|
with _backup_operation_lock(home):
|
|
with pytest.raises(BackupInProgressError):
|
|
with _backup_operation_lock(home, timeout_seconds=0):
|
|
raise AssertionError("second backup unexpectedly acquired the lock")
|
|
|
|
|
|
def test_atomic_output_publishes_only_after_clean_close(tmp_path) -> None:
|
|
final = tmp_path / "backup.zip"
|
|
final.write_bytes(b"previous")
|
|
|
|
with _atomic_output_path(final) as partial:
|
|
partial.write_bytes(b"complete")
|
|
assert final.read_bytes() == b"previous"
|
|
|
|
assert final.read_bytes() == b"complete"
|
|
assert not partial.exists()
|
|
|
|
|
|
def test_atomic_output_keeps_previous_file_after_failure(tmp_path) -> None:
|
|
final = tmp_path / "backup.zip"
|
|
final.write_bytes(b"previous")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
with _atomic_output_path(final) as partial:
|
|
partial.write_bytes(b"incomplete")
|
|
raise RuntimeError("compression failed")
|
|
|
|
assert final.read_bytes() == b"previous"
|
|
assert not partial.exists()
|
|
|
|
|
|
def test_quick_snapshot_is_published_with_manifest(tmp_path, monkeypatch) -> None:
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
(home / "config.yaml").write_bytes(b"model: {}\n")
|
|
published: list[tuple[Path, Path]] = []
|
|
|
|
from hermes_cli import backup
|
|
|
|
real_replace = backup.os.replace
|
|
|
|
def replace(source, destination) -> None:
|
|
source_path = Path(source)
|
|
destination_path = Path(destination)
|
|
if destination_path.parent == home / "state-snapshots":
|
|
assert source_path.name.endswith(".partial")
|
|
assert (source_path / "manifest.json").is_file()
|
|
assert not destination_path.exists()
|
|
published.append((source_path, destination_path))
|
|
real_replace(source, destination)
|
|
|
|
monkeypatch.setattr(backup.os, "replace", replace)
|
|
snapshot_id = create_quick_snapshot(hermes_home=home)
|
|
|
|
assert snapshot_id is not None
|
|
assert len(published) == 1
|
|
manifest = json.loads(
|
|
(home / "state-snapshots" / snapshot_id / "manifest.json").read_text(encoding="utf-8")
|
|
)
|
|
assert manifest["id"] == snapshot_id
|
|
assert manifest["files"] == {"config.yaml": 10}
|
|
|
|
|
|
@pytest.mark.platforms("posix") # POSIX permission bits
|
|
def test_quick_snapshot_tree_is_owner_only_under_permissive_umask(tmp_path) -> None:
|
|
"""Recovery snapshots must never inherit world-readable default modes.
|
|
|
|
A normal 0022 umask creates SQLite databases and JSON files as 0644 and
|
|
directories as 0755. Quick snapshots contain session state, credentials,
|
|
pairing records, and cron data, so every published file must be 0600 and
|
|
every directory 0700 regardless of the caller's umask or source modes.
|
|
"""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
(home / "config.yaml").write_text("model: {}\n", encoding="utf-8")
|
|
with sqlite3.connect(home / "state.db") as conn:
|
|
conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)")
|
|
|
|
old_umask = os.umask(0o022)
|
|
try:
|
|
snapshot_id = create_quick_snapshot(hermes_home=home)
|
|
finally:
|
|
os.umask(old_umask)
|
|
|
|
assert snapshot_id is not None
|
|
root = home / "state-snapshots"
|
|
snapshot = root / snapshot_id
|
|
directories = [root, snapshot, *(p for p in snapshot.rglob("*") if p.is_dir())]
|
|
files = [p for p in snapshot.rglob("*") if p.is_file()]
|
|
|
|
assert directories
|
|
assert files
|
|
assert all(stat.S_IMODE(path.stat().st_mode) == 0o700 for path in directories)
|
|
assert all(stat.S_IMODE(path.stat().st_mode) == 0o600 for path in files)
|
|
|
|
|
|
def test_quick_snapshot_listing_ignores_partial_directories(tmp_path) -> None:
|
|
home = tmp_path / ".hermes"
|
|
partial = home / "state-snapshots" / ".unfinished.1.partial"
|
|
partial.mkdir(parents=True)
|
|
(partial / "manifest.json").write_text('{"id":"unfinished"}', encoding="utf-8")
|
|
|
|
assert list_quick_snapshots(hermes_home=home) == []
|
|
|
|
|
|
def test_failed_automatic_backup_preserves_previous_archive(tmp_path, monkeypatch) -> None:
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
(home / "state.db").write_bytes(b"not-a-database")
|
|
archive = tmp_path / "automatic.zip"
|
|
archive.write_bytes(b"previous-valid-backup")
|
|
|
|
monkeypatch.setattr("hermes_cli.backup._safe_copy_db", lambda _src, _dst: False)
|
|
|
|
assert _write_full_zip_backup(archive, home) is None
|
|
assert archive.read_bytes() == b"previous-valid-backup"
|
|
assert list(tmp_path.glob(".*.partial")) == []
|