feat(backup): bundle assets in ZIP + partial-write hardening + restart log

Auto-backups now produce a ZIP containing ledgrab.db plus every file
in the assets dir under assets/ — matching the manual
GET /api/v1/system/backup format, so restore accepts either output
interchangeably. Legacy .db backups remain listable, restorable, and
prunable; both extensions count toward max_backups.

Writes stage to <name>.partial then os.replace into place — a crash
mid-ZIP never leaves a half-written backup that masquerades as valid.
Stale .partials from prior crashes are swept on the next run.
Symlinks inside the assets dir are skipped so a hostile link can't
slurp a target outside the dir into every backup. Backups larger than
500 MB log a warning so operators notice unbounded asset growth before
disk fills up.

restart.py: redirect the spawned restart script's stdout/stderr to
restart.log and bail out early if the script is missing — silent
failures (PowerShell off PATH, restart.ps1 erroring) used to vanish
into a detached child with no diagnostic trail.

Tests cover happy path, asset bytes round-trip, partial cleanup,
None/missing assets_dir, failure rollback, stale-partial sweep,
symlink rejection, mixed legacy+new listing, and cross-format prune.
This commit is contained in:
2026-05-28 17:25:55 +03:00
parent e4d24a02da
commit 85da2e538d
5 changed files with 348 additions and 27 deletions
@@ -0,0 +1,210 @@
"""Regression coverage for :class:`AutoBackupEngine` ZIP format.
Pins behavior added when auto-backups switched from raw ``.db`` snapshots to
``.zip`` archives (DB + assets), and the partial-write hardening that goes
with it (stage at ``<name>.partial`` then ``os.replace`` so a crash mid-write
never leaves a corrupt file masquerading as a valid backup).
"""
import zipfile
from pathlib import Path
from unittest.mock import patch
import pytest
from ledgrab.core.backup.auto_backup import AutoBackupEngine
from ledgrab.storage.database import Database
@pytest.fixture()
def engine_with_assets(tmp_path):
"""Engine wired to a small DB + assets dir with two sample files."""
db_path = tmp_path / "ledgrab.db"
assets_dir = tmp_path / "assets"
assets_dir.mkdir()
(assets_dir / "alert.wav").write_bytes(b"WAV_PAYLOAD_1")
(assets_dir / "ping.wav").write_bytes(b"WAV_PAYLOAD_2")
db = Database(db_path)
db.upsert("devices", "dev_1", "Living Room", {"id": "dev_1", "type": "mock"})
backup_dir = tmp_path / "backups"
engine = AutoBackupEngine(backup_dir=backup_dir, db=db, assets_dir=assets_dir)
yield engine
db.close()
def _only_backup(backup_dir: Path) -> Path:
"""Return the single .zip backup in ``backup_dir``; assert exactly one."""
zips = sorted(backup_dir.glob("*.zip"))
assert len(zips) == 1, f"expected exactly one .zip, found {[p.name for p in zips]}"
return zips[0]
def test_backup_bundles_db_and_assets(engine_with_assets):
"""Happy path: ZIP contains ``ledgrab.db`` plus every assets file."""
engine_with_assets._perform_backup_sync()
backup = _only_backup(engine_with_assets._backup_dir)
with zipfile.ZipFile(backup) as zf:
names = sorted(zf.namelist())
assert names == ["assets/alert.wav", "assets/ping.wav", "ledgrab.db"]
def test_backup_preserves_asset_bytes(engine_with_assets):
"""The asset binary inside the ZIP matches the source byte-for-byte."""
engine_with_assets._perform_backup_sync()
backup = _only_backup(engine_with_assets._backup_dir)
with zipfile.ZipFile(backup) as zf:
assert zf.read("assets/alert.wav") == b"WAV_PAYLOAD_1"
assert zf.read("assets/ping.wav") == b"WAV_PAYLOAD_2"
def test_backup_no_partial_file_left_on_success(engine_with_assets):
"""After a successful backup, the staging ``.partial`` is renamed away."""
engine_with_assets._perform_backup_sync()
leftovers = list(engine_with_assets._backup_dir.glob("*.partial"))
assert leftovers == []
def test_backup_skips_assets_when_dir_none(tmp_path):
"""``assets_dir=None`` produces a DB-only ZIP, no error."""
db_path = tmp_path / "ledgrab.db"
db = Database(db_path)
db.upsert("devices", "dev_1", "A", {"id": "dev_1"})
backup_dir = tmp_path / "backups"
try:
engine = AutoBackupEngine(backup_dir=backup_dir, db=db, assets_dir=None)
engine._perform_backup_sync()
backup = _only_backup(backup_dir)
with zipfile.ZipFile(backup) as zf:
assert zf.namelist() == ["ledgrab.db"]
finally:
db.close()
def test_backup_skips_assets_when_dir_missing(tmp_path):
"""A non-existent ``assets_dir`` is silently skipped, not an error."""
db_path = tmp_path / "ledgrab.db"
db = Database(db_path)
backup_dir = tmp_path / "backups"
try:
engine = AutoBackupEngine(
backup_dir=backup_dir, db=db, assets_dir=tmp_path / "does-not-exist"
)
engine._perform_backup_sync()
backup = _only_backup(backup_dir)
with zipfile.ZipFile(backup) as zf:
assert zf.namelist() == ["ledgrab.db"]
finally:
db.close()
def test_backup_failure_leaves_no_zip_or_partial(engine_with_assets):
"""When ``db.backup_to`` raises, neither the final .zip nor the .partial
survives. The exception propagates so the caller can log it.
"""
with patch.object(engine_with_assets._db, "backup_to", side_effect=RuntimeError("boom")):
with pytest.raises(RuntimeError, match="boom"):
engine_with_assets._perform_backup_sync()
assert list(engine_with_assets._backup_dir.glob("*.zip")) == []
assert list(engine_with_assets._backup_dir.glob("*.partial")) == []
def test_backup_cleans_stale_partial_from_previous_crash(engine_with_assets):
"""A leftover ``.partial`` from a prior crash is swept on the next run."""
stale = engine_with_assets._backup_dir
stale.mkdir(parents=True, exist_ok=True)
(stale / "ledgrab-backup-2025-01-01T000000.zip.partial").write_bytes(b"corrupt")
engine_with_assets._perform_backup_sync()
assert list(stale.glob("*.partial")) == []
# The successful backup is also present.
assert len(list(stale.glob("*.zip"))) == 1
def test_backup_skips_symlinks_in_assets_dir(tmp_path):
"""Symlinked files in ``assets_dir`` are not bundled (security guard)."""
db_path = tmp_path / "ledgrab.db"
assets_dir = tmp_path / "assets"
assets_dir.mkdir()
(assets_dir / "real.wav").write_bytes(b"REAL")
# Some test environments (e.g. unprivileged Windows) can't create
# symlinks; skip rather than fail spuriously when that's the case.
secret_target = tmp_path / "secret.bin"
secret_target.write_bytes(b"PRIVATE")
link_path = assets_dir / "linked.wav"
try:
link_path.symlink_to(secret_target)
except (OSError, NotImplementedError):
pytest.skip("symlinks not supported in this environment")
db = Database(db_path)
try:
backup_dir = tmp_path / "backups"
engine = AutoBackupEngine(backup_dir=backup_dir, db=db, assets_dir=assets_dir)
engine._perform_backup_sync()
backup = _only_backup(backup_dir)
with zipfile.ZipFile(backup) as zf:
names = zf.namelist()
assert "assets/real.wav" in names
assert "assets/linked.wav" not in names
finally:
db.close()
def test_list_backups_unions_legacy_db_and_new_zip(engine_with_assets):
"""``list_backups`` returns both legacy ``.db`` and current ``.zip`` files
so users can still restore from auto-backups written by older versions.
"""
backup_dir = engine_with_assets._backup_dir
backup_dir.mkdir(parents=True, exist_ok=True)
(backup_dir / "ledgrab-backup-2024-12-31T000000.db").write_bytes(b"legacy")
engine_with_assets._perform_backup_sync()
names = {b["filename"] for b in engine_with_assets.list_backups()}
assert any(n.endswith(".db") for n in names)
assert any(n.endswith(".zip") for n in names)
def test_prune_honors_max_across_formats(tmp_path):
"""``_prune_old_backups`` enforces ``max_backups`` across both extensions."""
db_path = tmp_path / "ledgrab.db"
db = Database(db_path)
backup_dir = tmp_path / "backups"
backup_dir.mkdir()
# Two legacy .db files (older), two .zip files (newer) — max=3 should
# prune the single oldest .db and keep the rest.
import time
(backup_dir / "ledgrab-backup-2024-01-01T000000.db").write_bytes(b"a")
time.sleep(0.02)
(backup_dir / "ledgrab-backup-2024-02-01T000000.db").write_bytes(b"b")
time.sleep(0.02)
(backup_dir / "ledgrab-backup-2024-03-01T000000.zip").write_bytes(b"c")
time.sleep(0.02)
(backup_dir / "ledgrab-backup-2024-04-01T000000.zip").write_bytes(b"d")
try:
engine = AutoBackupEngine(backup_dir=backup_dir, db=db, assets_dir=None)
engine._settings["max_backups"] = 3
engine._prune_old_backups()
remaining = sorted(p.name for p in engine._iter_backup_files())
assert remaining == [
"ledgrab-backup-2024-02-01T000000.db",
"ledgrab-backup-2024-03-01T000000.zip",
"ledgrab-backup-2024-04-01T000000.zip",
]
finally:
db.close()