diff --git a/patch/truecloud_nested.py b/patch/truecloud_nested.py index 0a432f2..df02c6a 100644 --- a/patch/truecloud_nested.py +++ b/patch/truecloud_nested.py @@ -56,6 +56,7 @@ from __future__ import annotations import contextlib import os +import stat import subprocess __all__ = [ @@ -155,8 +156,25 @@ def snapshot_tree_names(snapshot: str, all_names) -> list[str]: ] +def _probe_snapdir(path): + """Classify a snapshot directory: ``ok``, ``missing``, or why it is unusable. + + ``os.path.isdir()`` collapses "does not exist" and "cannot stat" into the + same ``False``, so an EACCES would report itself as "has no snapshot" and + send someone hunting for a snapshot that is sitting right there. Both cases + still abort the backup -- but it has to say which one. + """ + try: + st = os.stat(path) + except FileNotFoundError: + return "missing" + except OSError as e: + return f"cannot be read ({e.strerror})" + return "ok" if stat.S_ISDIR(st.st_mode) else "is not a directory" + + def plan_staging(base_dataset, base_mountpoint, path, snapshot_name, datasets, - staging_root, isdir=os.path.isdir): + staging_root, probe=_probe_snapdir): """Compute the bind-mount plan for staging a nested tree. Pure function. ``datasets`` is a list of dicts shaped like ``zfs.dataset.query`` results: @@ -216,11 +234,18 @@ def plan_staging(base_dataset, base_mountpoint, path, snapshot_name, datasets, continue src = snapdir(mp) - if not isdir(src): - # The recursive snapshot should have covered every descendant. If it - # did not, this dataset's data would be silently omitted. Refuse. + status = probe(src) + if status != "ok": + # Either the recursive snapshot missed this dataset, or we cannot read + # it. Either way its data would be silently omitted. Refuse -- but say + # WHICH, because "no snapshot" and "permission denied" send you to + # completely different places. + detail = ( + f"has no snapshot {snapshot_name!r}" if status == "missing" + else f"snapshot {snapshot_name!r} {status}" + ) raise StagingError( - f"dataset {name!r} has no snapshot {snapshot_name!r} at {src!r}; " + f"dataset {name!r} {detail} at {src!r}; " f"refusing to back up an incomplete tree" ) diff --git a/tests/test_truecloud_nested.py b/tests/test_truecloud_nested.py index aba764f..217aa6f 100644 --- a/tests/test_truecloud_nested.py +++ b/tests/test_truecloud_nested.py @@ -61,12 +61,12 @@ DATASETS = [ def yes(_path): - return True + return "ok" def plan(datasets=DATASETS, base_dataset="Tap", base_mp="/mnt/Tap", - path="/mnt/Tap", isdir=yes): - return plan_staging(base_dataset, base_mp, path, SNAP, datasets, ROOT, isdir=isdir) + path="/mnt/Tap", probe=yes): + return plan_staging(base_dataset, base_mp, path, SNAP, datasets, ROOT, probe=probe) class TestPlanStaging: @@ -142,15 +142,30 @@ class TestSilentOmissionGuard: @staticmethod def _missing_pgdata(path): - return "/mnt/Tap/apps/immich/pgdata/" not in path + return "missing" if "/mnt/Tap/apps/immich/pgdata/" in path else "ok" + + @staticmethod + def _denied_pgdata(path): + if "/mnt/Tap/apps/immich/pgdata/" in path: + return "cannot be read (Permission denied)" + return "ok" def test_missing_snapshot_on_descendant_raises(self): with pytest.raises(StagingError, match="incomplete tree"): - plan(isdir=self._missing_pgdata) + plan(probe=self._missing_pgdata) def test_error_names_the_offending_dataset(self): with pytest.raises(StagingError, match="Tap/apps/immich/pgdata"): - plan(isdir=self._missing_pgdata) + plan(probe=self._missing_pgdata) + + def test_missing_and_unreadable_are_reported_differently(self): + # os.path.isdir() collapses both into False, which would report a + # permission problem as "has no snapshot" and send you hunting for a + # snapshot that is sitting right there. Both abort -- but say which. + with pytest.raises(StagingError, match="has no snapshot"): + plan(probe=self._missing_pgdata) + with pytest.raises(StagingError, match="Permission denied"): + plan(probe=self._denied_pgdata) class TestSnapshotTreeNames: @@ -411,7 +426,7 @@ class TestApplyPlanRollback: runner = FakeRunner(fail_on="/src/b") with pytest.raises(StagingError, match="bind-mount"): - apply_plan(mounts, runner=runner, isdir=yes) + apply_plan(mounts, runner=runner, isdir=lambda _p: True) umounts = [c[-1] for c in runner.calls if c[0] == "umount"] assert umounts == [root + "/a", root]