Implement a restore command (#13)
* Implement restore command Add a restore subcommand alongside the existing backup command: - curateipsum/restore.py: restore_snapshot() and verify_restored(), reusing fs.scantree/copy_direntry/rm_direntry so metadata handling (mode, ownership, timestamps, symlinks) matches backup exactly. - backup.find_backup(): resolve a snapshot by name or latest. - cli.py: argparse subcommands (backup/restore), with backup implied when no subcommand is given so existing invocations keep working. Restore supports dry-run (creates nothing), partial restore by snapshot-relative path, an overwrite policy (never/always), a destination-inside-snapshot boundary check, and manifest-checksum verification of restored content. It takes the same backups_lock a backup does, since retention could otherwise delete the snapshot being restored out from under it. Closes P1 'Provide a safe restoration workflow' in IMPROVEMENTS.md. * Address Copilot review: symlink-escape boundary check, docstring fix - restore_snapshot() and verify_restored() now canonicalize each destination path and refuse to operate on it if it resolves outside dest_dir - a symlink planted under dest_dir (by another process, or left over from a prior run) could otherwise redirect a write, delete, or read outside the intended destination. - Fixed restore_snapshot()'s docstring: the skipped-paths list reports what would be skipped on dry-run too, it isn't always empty. * Address Copilot suppressed findings: stream plan_restore, restore-specific lock message - plan_restore() is now a generator instead of building a full list in memory, matching fs.scantree()'s streaming style everywhere else in the codebase. - restore CLI logs a restore-specific message on lock-acquire failure, since set_backups_lock()'s own log line talks about a 'previous backup', which read wrong for a restore. * Fix PATH docs: snapshot-relative means source-basename-prefixed README and CLI help examples for restore's PATH argument omitted the source directory's basename, which backup.initiate_backup() always prefixes onto snapshot-relative paths - the given examples would have matched nothing in an actual snapshot. --------- Co-authored-by: Maxim Shulga <maxim.shulga@alludo.com>
This commit is contained in:
snegovandMaxim Shulga authored and 
GitHub committed
1 parent
248204fe41
commit
33dcbe6c58
7 files changed
+799
-41
No files matched your search
@@ -0,0 +1,249 @@
|
||||
"""Tests for the restore module (restore.py)."""
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from curateipsum import backup as bk
|
||||
from curateipsum import restore
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def restore_dirs(tmp_path):
|
||||
backups_dir = tmp_path / "backups"
|
||||
source_dir = tmp_path / "source"
|
||||
dest_dir = tmp_path / "dest"
|
||||
backups_dir.mkdir()
|
||||
source_dir.mkdir()
|
||||
return backups_dir, source_dir, dest_dir
|
||||
|
||||
|
||||
def _make_snapshot(backups_dir, source_dir):
|
||||
entry = bk.initiate_backup(
|
||||
sources=[str(source_dir)], backups_dir=str(backups_dir))
|
||||
assert entry is not None
|
||||
return bk.find_backup(str(backups_dir))
|
||||
|
||||
|
||||
class TestRestoreSnapshot:
|
||||
def test_restores_files_and_dirs(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "subdir").mkdir()
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
(source_dir / "subdir" / "file2.txt").write_text("content2")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
skipped = restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
assert skipped == []
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
restored_root = dest_dir / src_name
|
||||
assert (restored_root / "file1.txt").read_text() == "content1"
|
||||
assert (restored_root / "subdir" / "file2.txt").read_text() == (
|
||||
"content2")
|
||||
|
||||
def test_preserves_symlinks(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "real.txt").write_text("data")
|
||||
os.symlink("real.txt", str(source_dir / "link.txt"))
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
link_path = dest_dir / src_name / "link.txt"
|
||||
assert os.path.islink(str(link_path))
|
||||
assert os.readlink(str(link_path)) == "real.txt"
|
||||
|
||||
def test_excludes_snapshot_bookkeeping(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
restored_names = os.listdir(str(dest_dir))
|
||||
assert bk.DELTA_DIR not in restored_names
|
||||
assert not any(n.startswith(bk.BACKUP_MARKER)
|
||||
for n in restored_names)
|
||||
|
||||
def test_overwrite_never_skips_existing(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("new-content")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
existing = dest_dir / src_name / "file1.txt"
|
||||
existing.parent.mkdir(parents=True)
|
||||
existing.write_text("old-content")
|
||||
|
||||
skipped = restore.restore_snapshot(
|
||||
snapshot, str(dest_dir), overwrite=restore.OVERWRITE_NEVER)
|
||||
|
||||
assert os.path.join(src_name, "file1.txt") in skipped
|
||||
assert existing.read_text() == "old-content"
|
||||
|
||||
def test_overwrite_always_replaces_existing(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("new-content")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
existing = dest_dir / src_name / "file1.txt"
|
||||
existing.parent.mkdir(parents=True)
|
||||
existing.write_text("old-content")
|
||||
|
||||
skipped = restore.restore_snapshot(
|
||||
snapshot, str(dest_dir), overwrite=restore.OVERWRITE_ALWAYS)
|
||||
|
||||
assert skipped == []
|
||||
assert existing.read_text() == "new-content"
|
||||
|
||||
def test_dry_run_writes_nothing(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
dest_dir.mkdir()
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
skipped = restore.restore_snapshot(
|
||||
snapshot, str(dest_dir), dry_run=True)
|
||||
|
||||
assert skipped == []
|
||||
assert os.listdir(str(dest_dir)) == []
|
||||
|
||||
def test_rel_paths_restricts_restore(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "keep.txt").write_text("keep")
|
||||
(source_dir / "skip.txt").write_text("skip")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
restore.restore_snapshot(
|
||||
snapshot, str(dest_dir),
|
||||
rel_paths=[os.path.join(src_name, "keep.txt")])
|
||||
|
||||
restored_root = dest_dir / src_name
|
||||
assert (restored_root / "keep.txt").read_text() == "keep"
|
||||
assert not (restored_root / "skip.txt").exists()
|
||||
|
||||
def test_refuses_to_restore_into_snapshot_itself(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
with pytest.raises(restore.RestoreError):
|
||||
restore.restore_snapshot(snapshot, snapshot.path)
|
||||
|
||||
def test_refuses_symlinked_dest_subdir_escape(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "subdir").mkdir()
|
||||
(source_dir / "subdir" / "file.txt").write_text("content")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
|
||||
outside = restore_dirs[2].parent / "outside"
|
||||
outside.mkdir()
|
||||
dest_dir.mkdir()
|
||||
(dest_dir / src_name).symlink_to(outside, target_is_directory=True)
|
||||
|
||||
with pytest.raises(restore.RestoreError):
|
||||
restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
assert os.listdir(str(outside)) == []
|
||||
|
||||
def test_unknown_overwrite_policy_rejected(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
with pytest.raises(ValueError):
|
||||
restore.restore_snapshot(
|
||||
snapshot, str(dest_dir), overwrite="sometimes")
|
||||
|
||||
|
||||
class TestVerifyRestored:
|
||||
def test_verify_matches_after_clean_restore(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
mismatches = restore.verify_restored(snapshot, str(dest_dir))
|
||||
assert mismatches == []
|
||||
|
||||
def test_verify_detects_corrupted_restore(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
restore.restore_snapshot(snapshot, str(dest_dir))
|
||||
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
(dest_dir / src_name / "file1.txt").write_text("corrupted")
|
||||
|
||||
mismatches = restore.verify_restored(snapshot, str(dest_dir))
|
||||
assert mismatches == [os.path.join(src_name, "file1.txt")]
|
||||
|
||||
def test_verify_refuses_symlinked_dest_escape(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
src_name = os.path.basename(str(source_dir))
|
||||
|
||||
outside = restore_dirs[2].parent / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "file1.txt").write_text("content1")
|
||||
dest_dir.mkdir()
|
||||
(dest_dir / src_name).symlink_to(outside, target_is_directory=True)
|
||||
|
||||
mismatches = restore.verify_restored(snapshot, str(dest_dir))
|
||||
assert mismatches == [os.path.join(src_name, "file1.txt")]
|
||||
|
||||
def test_verify_detects_missing_restore(self, restore_dirs):
|
||||
backups_dir, source_dir, dest_dir = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("content1")
|
||||
|
||||
snapshot = _make_snapshot(backups_dir, source_dir)
|
||||
dest_dir.mkdir()
|
||||
|
||||
mismatches = restore.verify_restored(snapshot, str(dest_dir))
|
||||
assert mismatches == [
|
||||
os.path.join(os.path.basename(str(source_dir)), "file1.txt")
|
||||
]
|
||||
|
||||
|
||||
class TestFindBackup:
|
||||
def test_finds_latest_by_default(self, restore_dirs):
|
||||
backups_dir, source_dir, _ = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("v1")
|
||||
first = _make_snapshot(backups_dir, source_dir)
|
||||
|
||||
(source_dir / "file1.txt").write_text("v2")
|
||||
bk.initiate_backup(sources=[str(source_dir)],
|
||||
backups_dir=str(backups_dir))
|
||||
|
||||
latest = bk.find_backup(str(backups_dir))
|
||||
assert latest.name != first.name
|
||||
|
||||
def test_finds_by_explicit_name(self, restore_dirs):
|
||||
backups_dir, source_dir, _ = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("v1")
|
||||
first = _make_snapshot(backups_dir, source_dir)
|
||||
|
||||
found = bk.find_backup(str(backups_dir), first.name)
|
||||
assert found.name == first.name
|
||||
|
||||
def test_raises_on_unknown_name(self, restore_dirs):
|
||||
backups_dir, source_dir, _ = restore_dirs
|
||||
(source_dir / "file1.txt").write_text("v1")
|
||||
_make_snapshot(backups_dir, source_dir)
|
||||
|
||||
with pytest.raises(bk.BackupFailedError):
|
||||
bk.find_backup(str(backups_dir), "20000101_000000")
|
||||
|
||||
def test_raises_when_no_backups_exist(self, restore_dirs):
|
||||
backups_dir, _, _ = restore_dirs
|
||||
with pytest.raises(bk.BackupFailedError):
|
||||
bk.find_backup(str(backups_dir))
|
||||
Reference in new issue
Block a user