diff --git a/IMPROVEMENTS.md b/IMPROVEMENTS.md index 638428f..2d79225 100644 --- a/IMPROVEMENTS.md +++ b/IMPROVEMENTS.md @@ -129,12 +129,12 @@ Acceptance criteria: ### Provide a safe restoration workflow -- [ ] Implement a restore command or a complete tested restore procedure. -- [ ] Support dry-run, partial restore, overwrite policy, and destination +- [x] Implement a restore command or a complete tested restore procedure. +- [x] Support dry-run, partial restore, overwrite policy, and destination boundary checks. -- [ ] Preserve supported metadata and verify restored content against the +- [x] Preserve supported metadata and verify restored content against the manifest. -- [ ] Test restoration from current and older snapshots. +- [x] Test restoration from current and older snapshots. ### Define filesystem compatibility diff --git a/README.md b/README.md index a89d250..2a09eb7 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,11 @@ pip install . ## Usage +cura-te-ipsum has two subcommands, `backup` and `restore`. For backwards +compatibility, `backup` is implied when the first argument isn't a +recognized subcommand - so `cura-te-ipsum -b DIR SOURCE` and +`cura-te-ipsum backup -b DIR SOURCE` are equivalent. + ### Basic Backup ```bash @@ -85,6 +90,53 @@ cura-te-ipsum -b /backups /home/user/data --verbose cura-te-ipsum -b /backups /home/user/data --external-rsync ``` +### Restoring a Backup + +``` +cura-te-ipsum restore -b BACKUPS_DIR DEST [PATH ...] + +Required Arguments: + -b BACKUPS_DIR Directory where backups are stored + DEST Directory to restore into (created if missing) + +Optional Arguments: + PATH Restore only these snapshot-relative paths - each + starts with the source directory's basename (e.g. + "data/documents/notes.txt" for a source backed up + from /home/user/data); default is everything + --snapshot NAME Snapshot to restore, e.g. 20260101_120000 + (default: the latest snapshot) + --overwrite {never,always} + What to do when a destination path already exists + (default: never - existing paths are left alone + and reported as skipped) + --verify Verify restored content against the snapshot's + manifest checksums afterwards + -n, --dry-run Show what would be restored without copying + anything or creating DEST + -f, --force Wait for a running backup/restore instead of + exiting immediately + -v, --verbose Enable debug logging +``` + +Restoring the latest snapshot in full: +```bash +cura-te-ipsum restore -b /backups /tmp/recovered +``` + +Restoring one file from a specific snapshot, verifying it afterwards +(here the source was backed up from `/home/user/data`, so the path +starts with `data/`): +```bash +cura-te-ipsum restore -b /backups --snapshot 20260101_120000 --verify \ + /tmp/recovered data/documents/notes.txt +``` + +A restore takes the same `.backups_lock` a backup does, so retention +cleanup can never delete a snapshot out from under a restore in progress. +Restored files keep the source's mode, ownership (when running as root), +and timestamps; symlinks are recreated as symlinks, never followed. + ## How It Works ### Hardlink-Based Snapshots @@ -139,10 +191,10 @@ backups/ truncating a file in one snapshot changes that same file's content in every other snapshot sharing its inode. Treat every snapshot directory as read-only; copy files out before editing them. -- **No dedicated restore command.** Recovering data means manually - copying files out of a snapshot directory - there is no restore - workflow with dry-run, partial restore, or overwrite-policy support - yet. +- **Restore preserves data content and standard metadata only.** The + `restore` command copies file content, mode, ownership, timestamps, + and symlinks; it does not preserve ACLs, extended attributes, or + filesystem-specific flags. - **Special filesystem entries (FIFOs, sockets, device nodes) are excluded, not backed up** - but only with the default Python implementation, which records them in the snapshot manifest as diff --git a/curateipsum/backup.py b/curateipsum/backup.py index 9514c45..383b981 100644 --- a/curateipsum/backup.py +++ b/curateipsum/backup.py @@ -338,6 +338,29 @@ def _date_from_backup(backup_entry: os.DirEntry) -> datetime: return datetime.strptime(backup_entry.name, BACKUP_ENT_FMT) +def find_backup(backups_dir: str, + snapshot_name: Optional[str] = None) -> os.DirEntry: + """ + Return a specific snapshot by name, or the latest one if + snapshot_name is None. + + :raises BackupFailedError: if snapshot_name doesn't match any backup + in backups_dir, or no backups exist at all. + """ + if snapshot_name is None: + latest = _get_latest_backup(backups_dir) + if latest is None: + raise BackupFailedError("No backups found in %s" % backups_dir) + return latest + + for entry in _iterate_backups(backups_dir): + if entry.name == snapshot_name: + return entry + raise BackupFailedError( + "No such backup %r in %s" % (snapshot_name, backups_dir) + ) + + # Open lock file descriptors currently held by this process, keyed by the # absolute path of the backups directory they lock. The kernel associates # an flock() with this exact open file description, so the descriptor must diff --git a/curateipsum/cli.py b/curateipsum/cli.py index 6eb20a4..6d6182e 100755 --- a/curateipsum/cli.py +++ b/curateipsum/cli.py @@ -8,18 +8,28 @@ import sys import time from datetime import timedelta -from curateipsum import backup +from curateipsum import backup, restore from curateipsum._version import version _lg = logging.getLogger("curateipsum") SUPPORTED_PLATFORMS = ("linux", "darwin") +_SUBCOMMANDS = ("backup", "restore") +_TOP_LEVEL_FLAGS = ("-h", "--help", "-V", "--version") -def main(): - formatter = logging.Formatter("{asctime}|{levelname}|{message}", style="{") - console_handler = logging.StreamHandler() - console_handler.setFormatter(formatter) +def _normalize_argv(argv): + """ + Insert the implicit "backup" subcommand when argv doesn't already + start with a known subcommand or a top-level flag (-h/-V), so + existing invocations like `cura-te-ipsum -b DIR SRC` keep working + unchanged now that `restore` exists alongside `backup`. + """ + if argv and (argv[0] in _SUBCOMMANDS or argv[0] in _TOP_LEVEL_FLAGS): + return argv + return ["backup", *argv] + +def _build_parser(): parser = argparse.ArgumentParser( prog="cura-te-ipsum", description="cura-te-ipsum, my personal backup software.", @@ -27,50 +37,83 @@ def main(): parser.add_argument("-V", "--version", action="version", version=f"%(prog)s v{version}") - parser.add_argument("-v", "--verbose", + + common = argparse.ArgumentParser(add_help=False) + common.add_argument("-v", "--verbose", action="store_true", default=False, help="print verbose information") - parser.add_argument("-b", + common.add_argument("-b", dest="backups_dir", metavar="BACKUPS_DIR", type=str, required=True, help="directory, where all backups will be stored") - parser.add_argument("-n", "--dry-run", + common.add_argument("-f", "--force", action="store_true", default=False, - help="Do not do create backup") - parser.add_argument("-f", "--force", + help="Wait for a running backup/restore to finish " + "instead of exiting immediately") + common.add_argument("-n", "--dry-run", action="store_true", default=False, - help="Wait for previous backup to finish instead of " - "exiting immediately if one is still running") - parser.add_argument("--external-rsync", - action="store_true", - default=False, - help="Use external rsync for copying") - parser.add_argument("--external-hardlink", - action="store_true", - default=False, - help="Use cp command for creating hardlink copies") - parser.add_argument("sources", - nargs="+", - metavar="SOURCE", - type=str, - help="backup source (file/dir/smth else)") - args = parser.parse_args() + help="show what would be done, without doing it") - loglevel = logging.DEBUG if args.verbose else logging.INFO - logging.basicConfig(level=loglevel, handlers=[console_handler]) + subparsers = parser.add_subparsers(dest="command", required=True) - _lg.info("Starting %s: %s", parser.prog, args) + backup_parser = subparsers.add_parser( + "backup", parents=[common], help="create a new backup") + backup_parser.add_argument("--external-rsync", + action="store_true", + default=False, + help="Use external rsync for copying") + backup_parser.add_argument("--external-hardlink", + action="store_true", + default=False, + help="Use cp command for creating hardlink " + "copies") + backup_parser.add_argument("sources", + nargs="+", + metavar="SOURCE", + type=str, + help="backup source (file/dir/smth else)") - if sys.platform not in SUPPORTED_PLATFORMS: - _lg.error("Not supported platform: %s. Supported platforms: %s", - sys.platform, SUPPORTED_PLATFORMS) - return 1 + restore_parser = subparsers.add_parser( + "restore", parents=[common], help="restore a backed-up snapshot") + restore_parser.add_argument("--snapshot", + dest="snapshot_name", + metavar="NAME", + default=None, + help="snapshot to restore, e.g. " + "20260101_120000 (default: latest)") + restore_parser.add_argument("--overwrite", + choices=restore.OVERWRITE_POLICIES, + default=restore.OVERWRITE_NEVER, + help="what to do when a destination path " + "already exists (default: %(default)s)") + restore_parser.add_argument("--verify", + action="store_true", + default=False, + help="verify restored content against the " + "snapshot manifest afterwards") + restore_parser.add_argument("dest", + metavar="DEST", + type=str, + help="directory to restore into") + restore_parser.add_argument("paths", + nargs="*", + metavar="PATH", + help="restore only these snapshot-relative " + "paths - each starts with the source " + "directory's basename, e.g. " + "'mydir/subdir/file.txt' for a source " + "backed up from /home/user/mydir " + "(default: restore everything)") + return parser + + +def _run_backup(args) -> int: if args.external_rsync and not shutil.which("rsync"): _lg.error("rsync should be installed to use --external-rsync option.") return 1 @@ -132,5 +175,101 @@ def main(): return exit_code +def _run_restore(args) -> int: + backups_dir_abs = os.path.abspath(args.backups_dir) + if not os.path.isdir(backups_dir_abs): + _lg.error("Backup directory %s does not exist, exiting", + args.backups_dir) + return 1 + + dest_abs = os.path.abspath(args.dest) + + # Locked for the same reason a backup is: retention could otherwise + # delete or quarantine the very snapshot this restore is reading from + # while it's still in progress. set_backups_lock() already logs why + # (e.g. "Previous backup is still in progress") - add restore-specific + # context here so a restore failure doesn't read like a backup one. + if not backup.set_backups_lock(backups_dir_abs, args.force): + _lg.error("Could not acquire the backups lock, restore aborted " + "(use --force to wait for it instead)") + return 1 + + exit_code = 0 + try: + try: + snapshot_entry = backup.find_backup(backups_dir_abs, + args.snapshot_name) + except backup.BackupFailedError as err: + _lg.error("%s", err) + return 1 + + _lg.info("Restoring snapshot %s to %s%s", + snapshot_entry.name, args.dest, + " (dry-run)" if args.dry_run else "") + + if not args.dry_run: + os.makedirs(dest_abs, exist_ok=True) + elif not os.path.isdir(dest_abs): + _lg.warning("Destination %s does not exist (dry-run, not " + "creating it)", args.dest) + + try: + skipped = restore.restore_snapshot( + snapshot_entry=snapshot_entry, + dest_dir=dest_abs, + dry_run=args.dry_run, + overwrite=args.overwrite, + rel_paths=args.paths or None, + ) + except restore.RestoreError as err: + _lg.error("Restore failed: %s", err) + return 1 + + if skipped: + _lg.warning("Skipped %d existing path(s) (use --overwrite " + "always to replace them): %s", + len(skipped), ", ".join(skipped)) + + if args.verify and not args.dry_run: + mismatches = restore.verify_restored( + snapshot_entry=snapshot_entry, + dest_dir=dest_abs, + rel_paths=args.paths or None, + ) + if mismatches: + _lg.error("Verification failed for %d path(s): %s", + len(mismatches), ", ".join(mismatches)) + exit_code = 1 + else: + _lg.info("Verified: restored content matches the manifest") + finally: + backup.release_backups_lock(backups_dir_abs) + + return exit_code + + +def main(): + formatter = logging.Formatter("{asctime}|{levelname}|{message}", style="{") + console_handler = logging.StreamHandler() + console_handler.setFormatter(formatter) + + parser = _build_parser() + args = parser.parse_args(_normalize_argv(sys.argv[1:])) + + loglevel = logging.DEBUG if args.verbose else logging.INFO + logging.basicConfig(level=loglevel, handlers=[console_handler]) + + _lg.info("Starting %s: %s", parser.prog, args) + + if sys.platform not in SUPPORTED_PLATFORMS: + _lg.error("Not supported platform: %s. Supported platforms: %s", + sys.platform, SUPPORTED_PLATFORMS) + return 1 + + if args.command == "restore": + return _run_restore(args) + return _run_backup(args) + + if __name__ == "__main__": sys.exit(main()) diff --git a/curateipsum/restore.py b/curateipsum/restore.py new file mode 100644 index 0000000..b8ce32f --- /dev/null +++ b/curateipsum/restore.py @@ -0,0 +1,190 @@ +""" +Module for restoring a backed-up snapshot (or a subset of its paths) to a +destination directory. +""" +import logging +import os +from typing import Iterable, List, Optional, Tuple, Union + +from curateipsum import backup, fs + +_lg = logging.getLogger(__name__) + +OVERWRITE_NEVER = "never" +OVERWRITE_ALWAYS = "always" +OVERWRITE_POLICIES = (OVERWRITE_NEVER, OVERWRITE_ALWAYS) + +# Snapshot-root entries that are backup bookkeeping, not backed-up data, +# and must never be copied out during a restore. +_SNAPSHOT_INTERNAL_PREFIXES = (backup.BACKUP_MARKER, backup.DELTA_DIR) + + +class RestoreError(Exception): + """Raised when a restore could not be completed safely or fully.""" + + +def _is_snapshot_internal(top_name: str) -> bool: + return any(top_name.startswith(p) for p in _SNAPSHOT_INTERNAL_PREFIXES) + + +def _under_any(relpath: str, rel_paths: List[str]) -> bool: + return any(relpath == p or relpath.startswith(p + os.sep) + for p in rel_paths) + + +def plan_restore( + snapshot_entry: Union[os.DirEntry, fs.PseudoDirEntry], + rel_paths: Optional[List[str]] = None, +) -> Iterable[Tuple[Union[os.DirEntry, fs.PseudoDirEntry], str]]: + """ + Yield (entry, relpath) pairs a restore would copy, directories before + their contents (fs.scantree's default order, which copy_direntry() + depends on so a directory exists before anything is written into it). + Skips snapshot-internal bookkeeping entries (backup marker, delta + dir). If rel_paths is given, only entries at or under one of those + snapshot-relative paths are included. + + A generator rather than a list: fs.scantree() already streams, and a + snapshot can hold far more entries than comfortably fit in memory at + once. + """ + for entry in fs.scantree(snapshot_entry.path): + relpath = os.path.relpath(entry.path, snapshot_entry.path) + top_name = relpath.split(os.sep, 1)[0] + if _is_snapshot_internal(top_name): + continue + if rel_paths is not None and not _under_any(relpath, rel_paths): + continue + yield entry, relpath + + +def restore_snapshot( + snapshot_entry: Union[os.DirEntry, fs.PseudoDirEntry], + dest_dir: str, + dry_run: bool = False, + overwrite: str = OVERWRITE_NEVER, + rel_paths: Optional[List[str]] = None, +) -> List[str]: + """ + Restore a completed snapshot (or a subset of its paths) into dest_dir. + + :param overwrite: "never" (default) leaves any destination path that + already exists untouched and reports it as skipped; "always" + removes and replaces it. + :param rel_paths: restrict the restore to these snapshot-relative + paths (and anything nested under them); default restores + everything in the snapshot. + :return: relative paths that already existed at dest_dir under + overwrite="never" - skipped for real, or (on dry_run) reported as + what would be skipped. + :raises RestoreError: if the snapshot has no manifest, dest_dir is + inside the snapshot being restored, a destination path would + resolve outside dest_dir (e.g. via a symlinked parent directory), + or a copy fails partway through (the destination is left as-is; + already-restored paths are not rolled back). + """ + if overwrite not in OVERWRITE_POLICIES: + raise ValueError("Unknown overwrite policy: %r" % overwrite) + + manifest = backup._read_manifest( + backup._get_backup_marker(snapshot_entry).path) + # A pre-migration (_LEGACY_MANIFEST) marker carries no checksums or + # excluded-entries record, but the snapshot data itself is still + # restorable - only verify_restored() needs real manifest content. + if manifest is None: + raise RestoreError( + "Cannot restore %s: missing or corrupt manifest" + % snapshot_entry.name + ) + + dest_canon = backup._canonical(dest_dir) + snapshot_canon = backup._canonical(snapshot_entry.path) + if backup._is_same_or_within(dest_canon, snapshot_canon): + raise RestoreError( + "Refusing to restore %s into itself (destination %s is inside " + "the snapshot)" % (snapshot_entry.name, dest_dir) + ) + + skipped = [] + for entry, relpath in plan_restore(snapshot_entry, rel_paths): + dst_path = os.path.join(dest_dir, relpath) + # A symlink planted under dest_dir (by another process, or left + # over from a previous run) could otherwise redirect lexists(), + # rm_direntry(), or copy_direntry() below to write, delete, or + # read through it, outside dest_dir entirely. + dst_canon = backup._canonical(dst_path) + if not backup._is_same_or_within(dst_canon, dest_canon): + raise RestoreError( + "Refusing to restore %s: resolves outside destination " + "%s (a parent path component may be a symlink)" + % (relpath, dest_dir) + ) + exists = os.path.lexists(dst_path) + if exists and overwrite == OVERWRITE_NEVER: + _lg.info("Skipping existing %s (overwrite=%s)", + relpath, OVERWRITE_NEVER) + skipped.append(relpath) + continue + if dry_run: + _lg.info("Would restore %s -> %s", relpath, dst_path) + continue + if exists: + fs.rm_direntry(fs.PseudoDirEntry(dst_path)) + parent = os.path.dirname(dst_path) + if parent and not os.path.isdir(parent): + os.makedirs(parent, exist_ok=True) + try: + fs.copy_direntry(entry, dst_path) + except (OSError, fs.BackupCreationError) as err: + raise RestoreError( + "Failed to restore %s: %s" % (relpath, err) + ) from err + + return skipped + + +def verify_restored( + snapshot_entry: Union[os.DirEntry, fs.PseudoDirEntry], + dest_dir: str, + rel_paths: Optional[List[str]] = None, +) -> List[str]: + """ + Verify restored file content at dest_dir against the checksums the + snapshot's manifest recorded at backup time. Mirrors + backup.verify_snapshot(), but against an arbitrary restore + destination rather than the snapshot directory itself. + + :return: relative paths whose restored content doesn't match the + recorded checksum, or is missing. Empty means verified OK. + :raises RestoreError: if the manifest itself is missing or corrupt. + """ + manifest = backup._read_manifest( + backup._get_backup_marker(snapshot_entry).path) + if manifest is None or manifest is backup._LEGACY_MANIFEST: + raise RestoreError( + "Cannot verify %s: missing, corrupt, or pre-migration manifest" + % snapshot_entry.name + ) + + dest_canon = backup._canonical(dest_dir) + mismatches = [] + for relpath, expected_digest in manifest.get("checksums", {}).items(): + if rel_paths is not None and not _under_any(relpath, rel_paths): + continue + full_path = os.path.join(dest_dir, relpath) + # Same symlink-escape concern as restore_snapshot(): don't let a + # planted symlink under dest_dir make this hash a file outside it. + full_canon = backup._canonical(full_path) + if not backup._is_same_or_within(full_canon, dest_canon): + _lg.error("Refusing to verify %s: resolves outside " + "destination %s", relpath, dest_dir) + mismatches.append(relpath) + continue + try: + actual_digest = backup._sha256_file(full_path) + except OSError: + mismatches.append(relpath) + continue + if actual_digest != expected_digest: + mismatches.append(relpath) + return mismatches diff --git a/tests/test_cli.py b/tests/test_cli.py index 6cf9587..f13637b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -586,3 +586,108 @@ class TestBackupExecution: assert any("Finished" in record.message for record in caplog.records) assert any("time spent" in record.message for record in caplog.records) + + +class TestRestoreSubcommand: + """End-to-end tests for `cura-te-ipsum restore`, against a real + snapshot on disk - restore's file operations are cheap enough that + mocking them out would just be testing the mocks.""" + + def _make_backup(self, backups_dir, source): + from curateipsum import backup as bk + (source / "file1.txt").write_text("content1") + entry = bk.initiate_backup(sources=[str(source)], + backups_dir=str(backups_dir)) + assert entry is not None + return entry + + def test_restore_copies_snapshot_to_dest(self, tmp_path): + backups_dir = tmp_path / "backups" + backups_dir.mkdir() + source = tmp_path / "src" + source.mkdir() + self._make_backup(backups_dir, source) + + dest = tmp_path / "dest" + with mock.patch('sys.argv', ['cura-te-ipsum', 'restore', '-b', + str(backups_dir), str(dest)]): + result = cli.main() + + assert result == 0 + restored = dest / "src" / "file1.txt" + assert restored.read_text() == "content1" + + def test_restore_unknown_backups_dir_fails(self, tmp_path): + dest = tmp_path / "dest" + with mock.patch('sys.argv', ['cura-te-ipsum', 'restore', '-b', + str(tmp_path / "nope"), str(dest)]): + result = cli.main() + + assert result == 1 + + def test_restore_unknown_snapshot_fails(self, tmp_path): + backups_dir = tmp_path / "backups" + backups_dir.mkdir() + source = tmp_path / "src" + source.mkdir() + self._make_backup(backups_dir, source) + + dest = tmp_path / "dest" + with mock.patch('sys.argv', [ + 'cura-te-ipsum', 'restore', '-b', str(backups_dir), + '--snapshot', '20000101_000000', str(dest)]): + result = cli.main() + + assert result == 1 + + def test_restore_dry_run_creates_nothing(self, tmp_path): + backups_dir = tmp_path / "backups" + backups_dir.mkdir() + source = tmp_path / "src" + source.mkdir() + self._make_backup(backups_dir, source) + + dest = tmp_path / "dest" + with mock.patch('sys.argv', ['cura-te-ipsum', 'restore', '-b', + str(backups_dir), '-n', str(dest)]): + result = cli.main() + + assert result == 0 + assert not dest.exists() + + def test_restore_verify_passes_after_clean_restore(self, tmp_path): + backups_dir = tmp_path / "backups" + backups_dir.mkdir() + source = tmp_path / "src" + source.mkdir() + self._make_backup(backups_dir, source) + + dest = tmp_path / "dest" + with mock.patch('sys.argv', [ + 'cura-te-ipsum', 'restore', '-b', str(backups_dir), + '--verify', str(dest)]): + result = cli.main() + + assert result == 0 + + def test_backup_subcommand_still_explicit(self, tmp_path): + """`cura-te-ipsum backup -b ... SRC` works the same as the + implicit form.""" + backups_dir = tmp_path / "backups" + backups_dir.mkdir() + source = tmp_path / "src" + source.mkdir() + + with mock.patch('sys.argv', ['cura-te-ipsum', 'backup', '-b', + str(backups_dir), str(source)]): + with mock.patch('curateipsum.backup.set_backups_lock', + return_value=True): + with mock.patch('curateipsum.backup.cleanup_old_backups'): + with mock.patch( + 'curateipsum.backup.initiate_backup') as m_init: + with mock.patch( + 'curateipsum.backup.release_backups_lock'): + result = cli.main() + + assert result == 0 + assert m_init.call_args.kwargs['sources'] == [str(source)] diff --git a/tests/test_restore.py b/tests/test_restore.py new file mode 100644 index 0000000..3e8898f --- /dev/null +++ b/tests/test_restore.py @@ -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))