List-first pruning of a full Storage Share CNPG dump folder (founder-run)

The CNPG Option A dumps prune only locally, so r01-* folders grow without bound;
the Backup account filled and the identity backup has refused since 2026-10-20.
plan is read-only and hashes the plan; delete needs that hash, re-lists and
refuses on any change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56

tests/test_prune_storage_share_dumps.py

@@ -0,0 +1,195 @@
+"""Synthetic tests for the list-first Storage Share dump pruning (no network, no credentials)."""
+
+import hashlib
+import importlib.util
+import io
+import json
+import urllib.error
+import urllib.parse
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location(
+    "prune", ROOT / "tools/bootstrap-recovery/prune-storage-share-dumps.py"
+)
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+FOLDER = "r01-net-kingdom-pg"
+
+
+def names(days, prefix="net-kingdom-pg-privacyidea_db"):
+    return {f"{prefix}-202610{d:02d}T023000Z.dump.age": 1000 + d for d in days}
+
+
+class Response(io.BytesIO):
+    def __init__(self, status, url, body=b""):
+        super().__init__(body)
+        self.status, self.url = status, url
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+
+class Share:
+    def __init__(self, files, extra=()):
+        self.files, self.extra, self.deleted = dict(files), list(extra), []
+
+    def open(self, request, timeout):
+        url, method = request.full_url, request.get_method()
+        assert request.headers["Authorization"].startswith("Basic ")
+        if method == "PROPFIND":
+            rows = "".join(
+                f"<d:response><d:href>{m.BASE}{FOLDER}/{urllib.parse.quote(n)}</d:href><d:propstat><d:prop>"
+                f"<d:resourcetype/><d:getcontentlength>{s}</d:getcontentlength></d:prop>"
+                f"<d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response>"
+                for n, s in [*self.files.items(), *self.extra]
+            )
+            body = (
+                f'<d:multistatus xmlns:d="DAV:"><d:response><d:href>{m.BASE}{FOLDER}/</d:href></d:response>'
+                f"{rows}</d:multistatus>"
+            ).encode()
+            return Response(207, url, body)
+        name = urllib.parse.unquote(url.rsplit("/", 1)[1])
+        if method == "DELETE":
+            del self.files[name]
+            self.deleted.append(name)
+            return Response(204, url)
+        if method == "HEAD":
+            if name in self.files:
+                return Response(200, url)
+            raise urllib.error.HTTPError(url, 404, "Not Found", None, None)
+        raise AssertionError(method)
+
+
+def make_plan(tmp_path, share, *extra):
+    path = tmp_path / "plan.json"
+    assert (
+        m.main(
+            ["plan", "--folder", FOLDER, "--plan", str(path), *extra],
+            client=share,
+            secret="pw",
+        )
+        == 0
+    )
+    return path, hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def test_plan_keeps_newest_daily_and_weekly_per_database(tmp_path, capsys):
+    share = Share(
+        {**names(range(1, 29)), **names(range(1, 29), "net-kingdom-pg-other_db")}
+    )
+    path, _ = make_plan(tmp_path, share)
+    plan = json.loads(path.read_text())
+    for prefix in ("net-kingdom-pg-privacyidea_db", "net-kingdom-pg-other_db"):
+        kept = sorted(n for n in plan["keep"] if n.startswith(prefix + "-"))
+        assert (
+            f"{prefix}-20261028T023000Z.dump.age" in kept
+            and f"{prefix}-20261022T023000Z.dump.age" in kept
+        )
+        assert 7 <= len(kept) <= 11
+    assert not share.deleted and set(plan["keep"]) | set(plan["delete"]) == set(
+        share.files
+    )
+    assert "plan sha256" in capsys.readouterr().out
+
+
+def test_delete_needs_the_plan_hash_and_removes_only_planned_files(tmp_path):
+    share = Share(names(range(1, 29)))
+    path, digest = make_plan(tmp_path, share)
+    plan = json.loads(path.read_text())
+    assert (
+        m.main(
+            [
+                "delete",
+                "--plan",
+                str(path),
+                "--confirm",
+                "0" * 64,
+                "--receipt",
+                str(tmp_path / "r0.json"),
+            ],
+            client=share,
+            secret="pw",
+        )
+        == 1
+    )
+    assert not share.deleted
+    assert (
+        m.main(
+            [
+                "delete",
+                "--plan",
+                str(path),
+                "--confirm",
+                digest,
+                "--receipt",
+                str(tmp_path / "r.json"),
+            ],
+            client=share,
+            secret="pw",
+        )
+        == 0
+    )
+    assert sorted(share.deleted) == sorted(plan["delete"]) and set(share.files) == set(
+        plan["keep"]
+    )
+    receipt = json.loads((tmp_path / "r.json").read_text())
+    assert (
+        receipt["kept_all_present"]
+        and receipt["deleted"] == len(plan["delete"])
+        and "pw" not in json.dumps(receipt)
+    )
+
+
+def test_delete_refuses_when_the_folder_changed_since_the_plan(tmp_path):
+    share = Share(names(range(1, 29)))
+    path, digest = make_plan(tmp_path, share)
+    victim = min(json.loads(path.read_text())["delete"])
+    share.files[victim] += 1
+    assert (
+        m.main(
+            [
+                "delete",
+                "--plan",
+                str(path),
+                "--confirm",
+                digest,
+                "--receipt",
+                str(tmp_path / "r.json"),
+            ],
+            client=share,
+            secret="pw",
+        )
+        == 1
+    )
+    assert not share.deleted
+
+
+@pytest.mark.parametrize("folder", ["net-kingdom-identity", "openbao", "r01-x/..", ""])
+def test_only_r01_dump_folders_are_accepted(tmp_path, folder):
+    assert (
+        m.main(
+            ["plan", "--folder", folder, "--plan", str(tmp_path / "p.json")],
+            client=Share({}),
+            secret="pw",
+        )
+        == 1
+    )
+
+
+def test_unexpected_entries_stop_the_plan(tmp_path):
+    share = Share(names(range(1, 9)), extra=[("notes.txt", 5)])
+    assert (
+        m.main(
+            ["plan", "--folder", FOLDER, "--plan", str(tmp_path / "p.json")],
+            client=share,
+            secret="pw",
+        )
+        == 1
+    )
+    assert not (tmp_path / "p.json").exists()