From b8b7758826b56bae63d782120f052ad1f473b1f1 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 9 Jul 2026 22:45:12 -0400 Subject: [PATCH] =?UTF-8?q?feat(restore):=20Phase-2=20full=20restore=20cap?= =?UTF-8?q?stone=20=E2=80=94=20restore.py=20(slice=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Disjoint-union reassembly of a size-routed target from its _vault manifest (docs/RFC_incremental_v2.md §2.9): - completeness gate BEFORE any transfer: every selected object tar must exist on Fortress (reclaim.hsi_exists) or ABORT — never a partial restore - empty-dest guard: restore only into a new/empty directory - per-object extract by recorded transport: htar -xvf -L memberlist (random-access indexed) / hsi get + tar xf for htar_large (invariant #6) - per-file MD5 verify against each object's own run-log source_files (arcname-keyed, invariant #2) - MANDATORY final reconciliation on a full restore: set(walk(dest)) == U objects' arcnames, else ABORT - --slot solo: / --slot shard: / --member selective restore (reconciliation is full-restore-only) LOCAL only (hsi/htar at /opt/hsi/bin, prepended like ship_object). Driven by either the target's archive config (badge -> manifest) or --manifest directly. Tests: tests/test_restore.py (17) — tape mocked via reclaim.hsi_exists + an injected extract fn (mirrors ship_one). Full suite 238 -> 255 green. Co-Authored-By: Claude Opus 4.8 --- restore.py | 374 ++++++++++++++++++++++++++++++++++++++++++ tests/test_restore.py | 285 ++++++++++++++++++++++++++++++++ 2 files changed, 659 insertions(+) create mode 100644 restore.py create mode 100644 tests/test_restore.py diff --git a/restore.py b/restore.py new file mode 100644 index 0000000..24eb7b9 --- /dev/null +++ b/restore.py @@ -0,0 +1,374 @@ +#!/usr/bin/env python3 +""" +restore.py — full / selective restore of a Phase-2 size-routed target +(docs/RFC_incremental_v2.md §2.9, the Phase-2 capstone). + +A size-routed target lives on Fortress as MANY objects (solos + shards), indexed +by the per-target manifest at {log_dir}/_vault/{badge}.manifest.json. Because the +objects are a DISJOINT partition of the current member set, restore is a plain +union: extract every current object into an empty destination and reconcile — +no overlay ordering, no deletion replay (that is the deferred Phase-3 leveled +restore, RFC §3.4). + +Order of operations (each gate ABORTS the whole restore — never partial): + (a) completeness gate BEFORE any transfer — every selected object's tar must + still exist on Fortress (reclaim.hsi_exists), else "not fully restorable"; + (b) destination MUST be empty (created if absent) — never restore over data; + (c) per object, by recorded transport: + "htar" — `htar -xvf -L ` random-access indexed + extract straight into dest (reads only those members); + "htar_large" — `hsi get` the plain tar to scratch, then `tar xf` + (htar_large tars have no .idx; invariant #6 — never + stream htar_large/-xvf through stdout); + (d) every restored member's MD5 is verified against the object's own run-log + ({log_dir}/{run_timestamp}.json, key `source_files`, arcname-keyed — + invariant #2); + (e) on a FULL restore, a MANDATORY final reconciliation: + set(walk(dest)) == ⋃ objects' arcnames, else ABORT. + +Selective restore (`--slot solo:` / `--slot shard:` / `--member relpath`) +filters the object set — for `htar` objects this reads only the requested +members' bytes off tape. Reconciliation (e) runs only on a full restore. + +LOCAL only (needs `hsi`/`htar` at /opt/hsi/bin + the filesystem) — run on the +storage host (Negishi), like reclaim.py. Operator-invoked; NOT wired into +`reclaim --delete` (a full restore pulls every tar off tape, TB-scale). + +Usage: + python3 restore.py config.json --dest /scratch/.../restore_here + python3 restore.py --manifest /depot/.../logs/_vault/.manifest.json \ + --dest DIR [--log-dir DIR] [--tmp-dir DIR] + python3 restore.py config.json --dest DIR --slot shard:3 --slot solo:1a2b... + python3 restore.py config.json --dest DIR --member sub/path/file.bin +""" +import argparse +import hashlib +import json +import os +import subprocess +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import archive # noqa: E402 (badge_of, vault_dir_for, manifest_path, load_config) +import reclaim # noqa: E402 (hsi_exists — the same tape-presence check reclaim trusts) + + +class RestoreError(RuntimeError): + """Any condition that makes the restored tree untrustworthy — always aborts.""" + + +def prepend_hpss_path(): + """hsi/htar live in /opt/hsi/bin and are not on the non-interactive PATH; + htar_large is /usr/local/bin. Prepend both (mirrors archive.ship_object).""" + for _bin in ("/opt/hsi/bin", "/usr/local/bin"): + if os.path.isdir(_bin) and _bin not in os.environ.get("PATH", "").split(":"): + os.environ["PATH"] = _bin + ":" + os.environ.get("PATH", "") + + +def check_dest(dest): + """Gate (b): the destination must not exist, or exist EMPTY — restoring over + live data is forbidden (RFC §2.9: 'always into an empty scratch dir').""" + if os.path.exists(dest): + if not os.path.isdir(dest): + raise RestoreError(f"restore dest {dest} exists and is not a directory") + if os.listdir(dest): + raise RestoreError( + f"restore dest {dest} is NOT empty — refusing to restore over " + f"existing data. Point --dest at a new/empty directory.") + + +def select_objects(manifest, slots=None, members=None): + """ + Resolve the object set to restore. Returns [(object, arcnames_to_restore)] + in manifest order. No filter = full restore (every object, every member). + `slots` picks whole objects by slot id; `members` picks single files (each + mapped to the one current object that owns it). Unknown slot/member = abort + (a typo must never silently restore nothing). + """ + objects = manifest.get("objects") or [] + if not objects: + raise RestoreError("manifest has no objects — nothing to restore") + if not slots and not members: + return [(o, list(o.get("arcnames") or [])) for o in objects] + + by_slot = {o.get("slot"): o for o in objects} + picked = {} # slot -> set(arcnames) + for s in (slots or ()): + if s not in by_slot: + known = ", ".join(sorted(by_slot)[:8]) + raise RestoreError(f"unknown slot {s!r} — manifest has: {known}" + + (" ..." if len(by_slot) > 8 else "")) + picked.setdefault(s, set()).update(by_slot[s].get("arcnames") or []) + if members: + owner = {} + for o in objects: + for a in (o.get("arcnames") or ()): + owner[a] = o.get("slot") + for m in members: + if m not in owner: + raise RestoreError( + f"member {m!r} is not in any current object — check the relpath " + f"(arcnames are relative to the archived source_folder)") + picked.setdefault(owner[m], set()).add(m) + + return [(o, sorted(picked[o.get("slot")])) for o in objects + if o.get("slot") in picked] + + +def extract_object(obj, arcnames, dest, tmp_dir): + """ + Gate (c): pull ONE object's members off tape into dest, by the transport the + manifest recorded at ship time (archive.ship_object): + htar — indexed random-access extract of exactly `arcnames` via a + member-list file (only those members' bytes are read); + htar_large — plain tar (no .idx): hsi get to scratch, tar xf the members + (invariant #6: never htar_large -xvf through a pipe). + htar/tar resolve members relative to CWD, so both run with cwd=dest. + """ + prepend_hpss_path() + transport = obj.get("transport") or "htar" + tar = obj["fortress_tar"] + os.makedirs(tmp_dir, exist_ok=True) + + if transport == "htar": + listfile = os.path.join(tmp_dir, f"{obj['run_timestamp']}.restore.members") + with open(listfile, "w") as fh: + fh.write("\n".join(arcnames) + "\n") + try: + r = subprocess.run(["htar", "-xvf", tar, "-L", listfile], cwd=dest, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + finally: + try: + os.remove(listfile) + except OSError: + pass + if r.returncode != 0: + raise RestoreError( + f"htar extract failed (rc={r.returncode}) for {tar}:\n" + + r.stdout.decode("utf-8", "replace")[-4000:]) + else: + local_tar = os.path.join(tmp_dir, f"{obj['run_timestamp']}.restore.tar") + try: + g = subprocess.run(f'hsi -q "get {local_tar} : {tar}"', shell=True, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + if g.returncode != 0: + raise RestoreError( + f"hsi get failed (rc={g.returncode}) for {tar}:\n" + + g.stdout.decode("utf-8", "replace")[-4000:]) + x = subprocess.run(["tar", "xf", local_tar] + list(arcnames), cwd=dest, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + if x.returncode != 0: + raise RestoreError( + f"tar extract failed (rc={x.returncode}) for {local_tar}:\n" + + x.stdout.decode("utf-8", "replace")[-4000:]) + finally: + try: + os.remove(local_tar) + except OSError: + pass + + +def object_runlog_checksums(log_dir, run_timestamp): + """The object's own run-log ({run_timestamp}.json, written by ship_object) + carries the arcname-keyed per-file MD5s (`source_files`, invariant #2).""" + path = os.path.join(log_dir, f"{run_timestamp}.json") + if not os.path.exists(path): + raise RestoreError( + f"object run-log not found: {path} — cannot MD5-verify this object " + f"(pass --log-dir if logs live elsewhere)") + with open(path) as fh: + log = json.load(fh) + checks = log.get("source_files") + if not isinstance(checks, dict) or not checks: + raise RestoreError(f"object run-log {path} has no source_files checksums") + return checks + + +def md5_file(path): + h = hashlib.md5() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def verify_members(obj, arcnames, dest, log_dir): + """Gate (d): every restored member's MD5 must equal the MD5 recorded when the + object shipped. Any mismatch/missing file poisons the whole restore.""" + checks = object_runlog_checksums(log_dir, obj["run_timestamp"]) + for arc in arcnames: + expected = checks.get(arc) + if expected is None: + raise RestoreError( + f"no recorded MD5 for {arc!r} in run-log {obj['run_timestamp']}.json " + f"— checksum keys must match arcnames (invariant #2)") + full = os.path.join(dest, arc) + if not os.path.isfile(full): + raise RestoreError( + f"member {arc!r} missing from dest after extracting " + f"{obj.get('fortress_tar')}") + got = md5_file(full) + if got != expected: + raise RestoreError( + f"MD5 MISMATCH for {arc!r} from {obj.get('fortress_tar')} " + f"(expected {expected}, got {got})") + + +def walk_restored(dest): + """Set of relpaths of every file now under dest (reconciliation walk).""" + out = set() + for root, _dirs, files in os.walk(dest): + for f in files: + out.add(os.path.relpath(os.path.join(root, f), dest)) + return out + + +def restore_target(manifest, dest, log_dir, tmp_dir, slots=None, members=None, + extract=extract_object): + """ + Restore a routed target (or a --slot/--member selection of it) into an empty + dest. Implements gates (a)-(e) from the module docstring in order; raises + RestoreError on the first violated gate. `extract` is injected so tests can + mock the tape path (mirrors route_and_ship's ship_one). + + Returns a summary dict: badge, objects, files, restored (set of arcnames), + reconciled (True on a verified full restore, None when a selection skipped + the reconciliation). + """ + selected = select_objects(manifest, slots, members) + full_restore = not slots and not members + + check_dest(dest) # (b) — cheap, no side effects + + # (a) completeness gate BEFORE any transfer — a partially-restorable target is + # NOT restorable ("never 'restored minus one'", RFC §2.9). + missing = [o.get("fortress_tar") for o, _arcs in selected + if not reclaim.hsi_exists(o.get("fortress_tar"))] + if missing: + raise RestoreError( + f"target NOT fully restorable — {len(missing)} of {len(selected)} object " + f"tar(s) missing on Fortress (e.g. {missing[0]}); aborting before any " + f"extraction") + + os.makedirs(dest, exist_ok=True) + restored = set() + for obj, arcs in selected: + print(f" [RESTORE] {obj.get('slot')} ({obj.get('transport') or 'htar'}, " + f"{len(arcs)} member(s)) <- {obj.get('fortress_tar')}") + extract(obj, arcs, dest, tmp_dir) # (c) + verify_members(obj, arcs, dest, log_dir) # (d) + print(f" [OK] {obj.get('slot')}: {len(arcs)} member MD5(s) verified") + restored.update(arcs) + + reconciled = None + if full_restore: # (e) MANDATORY, full only + expected = set() + for o in manifest.get("objects") or []: + expected.update(o.get("arcnames") or ()) + actual = walk_restored(dest) + if actual != expected: + extra = sorted(actual - expected) + miss = sorted(expected - actual) + raise RestoreError( + "final reconciliation FAILED — restored tree != manifest aggregate " + f"({len(extra)} unexpected, e.g. {extra[:3]}; " + f"{len(miss)} missing, e.g. {miss[:3]}). " + "The restored tree must NOT be trusted.") + reconciled = True + print(f" [OK] final reconciliation: {len(actual)} restored file(s) == " + f"manifest aggregate") + + return {"badge": manifest.get("badge"), "objects": len(selected), + "files": len(restored), "restored": restored, "reconciled": reconciled} + + +def main(argv=None): + ap = argparse.ArgumentParser( + description="Restore a Phase-2 size-routed target from Fortress " + "(docs/RFC_incremental_v2.md §2.9). LOCAL only; empty dest only.") + ap.add_argument("config", nargs="?", + help="The target's archive config JSON (resolves badge -> " + "{log_dir}/_vault manifest). Alternative: --manifest.") + ap.add_argument("--manifest", + help="Path to a {badge}.manifest.json (bypasses the config)") + ap.add_argument("--dest", required=True, + help="Destination directory — MUST be empty (or absent)") + ap.add_argument("--log-dir", dest="log_dir", + help="Where the object run-logs (*.json) live. Default: the " + "config's log_dir, or the manifest's _vault parent.") + ap.add_argument("--tmp-dir", dest="tmp_dir", + help="Scratch for member lists + htar_large staging tars. " + "Default: the config's tmp_dir, else .restore-tmp") + ap.add_argument("--slot", action="append", dest="slots", metavar="SLOT", + help="Restore only this object (solo: or shard:); repeatable") + ap.add_argument("--member", action="append", dest="members", metavar="RELPATH", + help="Restore only this file (relpath under source_folder); repeatable") + args = ap.parse_args(argv) + + if bool(args.config) == bool(args.manifest): + ap.error("give exactly one of: a config JSON, or --manifest") + + if args.manifest: + mpath = os.path.abspath(args.manifest) + try: + with open(mpath) as fh: + manifest = json.load(fh) + except (OSError, ValueError) as e: + print(f" [ABORT] cannot read manifest {mpath}: {e}") + sys.exit(1) + log_dir = args.log_dir + if not log_dir: + vault = os.path.dirname(mpath) + if os.path.basename(vault) != "_vault": + ap.error("--manifest is not under a _vault/ dir; pass --log-dir") + log_dir = os.path.dirname(vault) + tmp_dir = args.tmp_dir + else: + try: + config = archive.load_config(args.config) + except (OSError, ValueError) as e: + print(f" [ABORT] bad config {args.config}: {e}") + sys.exit(1) + badge = archive.badge_of(config["project_name"], config["source_folder"], + config["file_pattern"]) + vault = archive.vault_dir_for(config["log_dir"]) + mpath = archive.manifest_path(vault, badge) + manifest = archive.load_manifest(vault, badge) + if manifest is None: + print(f" [ABORT] no size-routing manifest for " + f"{config['project_name']!r} at {mpath} — was this target ever " + f"archived with size_routing on?") + sys.exit(1) + log_dir = args.log_dir or config["log_dir"] + tmp_dir = args.tmp_dir or config["tmp_dir"] + + dest = os.path.abspath(args.dest) + if not tmp_dir: + tmp_dir = dest.rstrip(os.sep) + ".restore-tmp" + + print("#" * 64) + print("# FORTRESS RESTORE — Phase-2 routed target (RFC §2.9), LOCAL") + print(f"# badge: {manifest.get('badge')} project: {manifest.get('project')}") + print(f"# manifest: {mpath}") + print(f"# objects: {len(manifest.get('objects') or [])}" + + (f" (selection: {(args.slots or []) + (args.members or [])})" + if (args.slots or args.members) else " (FULL restore)")) + print(f"# dest: {dest}") + print("#" * 64) + + try: + summary = restore_target(manifest, dest, log_dir, tmp_dir, + slots=args.slots, members=args.members) + except RestoreError as e: + print(f"\n [ABORT] {e}") + print(" Nothing restored is trustworthy until every gate passes — if dest " + "was partially written, remove it before re-running.") + sys.exit(1) + + scope = "FULL restore, reconciled" if summary["reconciled"] else "selective restore" + print(f"\n [SUCCESS] {scope}: {summary['files']} file(s) from " + f"{summary['objects']} object(s) -> {dest}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_restore.py b/tests/test_restore.py new file mode 100644 index 0000000..762d5d3 --- /dev/null +++ b/tests/test_restore.py @@ -0,0 +1,285 @@ +""" +Unit tests for Phase-2 full restore (docs/RFC_incremental_v2.md §2.9, slice 5): +restore.restore_target() reassembles a size-routed target from its manifest as a +disjoint union of objects — completeness gate before any transfer, empty-dest +guard, per-file MD5 verify against each object's run-log, mandatory final +reconciliation on a full restore, --slot/--member selection. Tape is mocked +(reclaim.hsi_exists + an injected extract fn); no hsi/htar/Slurm needed. +""" +import hashlib +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import reclaim # noqa: E402 +import restore # noqa: E402 + + +STORE = { + "big.bin": b"B" * 4096, + "sub/s1.bin": b"one-one-one", + "sub/s2.bin": b"two-two", + "sub/s3.bin": b"three-3-three", +} + +# slot, kind, arcnames, run_timestamp, transport +OBJECTS = [ + ("solo:beef", "solo", ["big.bin"], + "proj__solo-beef_20260101_000000", "htar"), + ("shard:0", "shard", ["sub/s1.bin", "sub/s3.bin"], + "proj__shard-0of2_20260101_000000", "htar"), + ("shard:1", "shard", ["sub/s2.bin"], + "proj__shard-1of2_20260101_000000", "htar_large"), +] + + +def build_fixture(d): + """log_dir with per-object run-logs (source_files MD5s) + a vault manifest. + Returns (manifest, log_dir).""" + log_dir = os.path.join(d, "logs") + os.makedirs(os.path.join(log_dir, "_vault")) + entries = [] + for slot, kind, arcs, rts, transport in OBJECTS: + checks = {a: hashlib.md5(STORE[a]).hexdigest() for a in arcs} + tar = "/tape/%s.tar" % rts + idx = tar + ".idx" if transport == "htar" else None + with open(os.path.join(log_dir, rts + ".json"), "w") as fh: + json.dump({"project": rts, "run_timestamp": rts, "status": "success", + "routed_object": True, "fortress_tar": tar, + "transport": transport, "idx": idx, + "source_files": checks, "n_files": len(arcs), + "source_bytes": sum(len(STORE[a]) for a in arcs)}, fh) + entries.append({"slot": slot, "kind": kind, "arcnames": list(arcs), + "fortress_tar": tar, "run_timestamp": rts, + "transport": transport, "idx": idx, + "n_files": len(arcs), + "source_bytes": sum(len(STORE[a]) for a in arcs), + "fingerprint": {}}) + manifest = {"schema_version": 1, "badge": "tstbadge", "project": "proj", + "source_folder": "/depot/anywhere/src", "file_pattern": ".*", + "exclusion": {"status": "OFF"}, + "routing": {"t_small": 256, "shard_count": 2}, + "objects": entries} + with open(os.path.join(log_dir, "_vault", "tstbadge.manifest.json"), "w") as fh: + json.dump(manifest, fh) + return manifest, log_dir + + +def fake_extract(store=STORE, calls=None, extra=None): + """An extract fn that materializes members from an in-memory 'tape'. + `calls` collects (slot, arcnames); `extra` = stray relpaths to also write + (reconciliation-failure fixture).""" + def _extract(obj, arcnames, dest, tmp_dir): + if calls is not None: + calls.append((obj["slot"], list(arcnames))) + for a in list(arcnames) + list(extra or ()): + p = os.path.join(dest, a) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "wb") as fh: + fh.write(store.get(a, b"stray")) + return _extract + + +class RestoreBase(unittest.TestCase): + def setUp(self): + self._hsi = reclaim.hsi_exists + reclaim.hsi_exists = lambda tar: True # all tars present by default + + def tearDown(self): + reclaim.hsi_exists = self._hsi + + def run_restore(self, d, manifest, log_dir, **kw): + kw.setdefault("extract", fake_extract()) + return restore.restore_target( + manifest, os.path.join(d, "dest"), log_dir, + os.path.join(d, "tmp"), **kw) + + +class TestFullRestore(RestoreBase): + def test_reconstructs_exact_aggregate(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + r = self.run_restore(d, manifest, log_dir) + dest = os.path.join(d, "dest") + expected = set(STORE) + self.assertEqual(restore.walk_restored(dest), expected) + self.assertTrue(r["reconciled"]) + self.assertEqual(r["files"], len(expected)) + self.assertEqual(r["objects"], 3) + for a, body in STORE.items(): # bytes, not just names + with open(os.path.join(dest, a), "rb") as fh: + self.assertEqual(fh.read(), body) + + def test_reconciliation_stray_file_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + with self.assertRaisesRegex(restore.RestoreError, "reconciliation FAILED"): + self.run_restore(d, manifest, log_dir, + extract=fake_extract(extra=["stray/x.bin"])) + + def test_reconciliation_missing_member_aborts(self): + # An arcname in the manifest aggregate that no selected object delivered + # (e.g. a manifest entry drifted) must fail the union check, not pass. + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + ghost = dict(manifest["objects"][2]) + ghost = json.loads(json.dumps(ghost)) + ghost["slot"] = "shard:9" + ghost["arcnames"] = ["sub/ghost.bin"] + manifest["objects"].append(ghost) + + def skip_ghost(obj, arcnames, dest, tmp_dir): + if obj["slot"] != "shard:9": + fake_extract()(obj, arcnames, dest, tmp_dir) + # ghost's members never verify -> that aborts first; make verify pass + # by pointing its run-log checks at nothing it claims... simpler: the + # ghost object also fails MD5-verify, which is itself an abort — so + # assert restore aborts either way (never a silent partial tree). + with self.assertRaises(restore.RestoreError): + self.run_restore(d, manifest, log_dir, extract=skip_ghost) + + +class TestCompletenessGate(RestoreBase): + def test_partial_gone_aborts_before_any_extraction(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + gone = manifest["objects"][1]["fortress_tar"] + reclaim.hsi_exists = lambda tar: tar != gone + calls = [] + with self.assertRaisesRegex(restore.RestoreError, + "NOT fully restorable"): + self.run_restore(d, manifest, log_dir, + extract=fake_extract(calls=calls)) + self.assertEqual(calls, []) # nothing extracted + self.assertFalse(os.path.exists(os.path.join(d, "dest"))) # dest untouched + + def test_all_gone_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + reclaim.hsi_exists = lambda tar: False + with self.assertRaises(restore.RestoreError): + self.run_restore(d, manifest, log_dir) + + +class TestSelectiveRestore(RestoreBase): + def test_slot_restores_one_object_only(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + calls = [] + r = self.run_restore(d, manifest, log_dir, slots=["shard:0"], + extract=fake_extract(calls=calls)) + self.assertEqual([c[0] for c in calls], ["shard:0"]) + self.assertEqual(restore.walk_restored(os.path.join(d, "dest")), + {"sub/s1.bin", "sub/s3.bin"}) + self.assertIsNone(r["reconciled"]) # reconciliation is full-only + + def test_slot_gate_checks_only_selected_tars(self): + # A GONE tar OUTSIDE the selection must not block a selective restore. + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + gone = manifest["objects"][0]["fortress_tar"] # solo — not selected + reclaim.hsi_exists = lambda tar: tar != gone + r = self.run_restore(d, manifest, log_dir, slots=["shard:1"]) + self.assertEqual(r["files"], 1) + + def test_member_restores_single_file(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + calls = [] + r = self.run_restore(d, manifest, log_dir, members=["sub/s3.bin"], + extract=fake_extract(calls=calls)) + self.assertEqual(calls, [("shard:0", ["sub/s3.bin"])]) + self.assertEqual(restore.walk_restored(os.path.join(d, "dest")), + {"sub/s3.bin"}) + self.assertIsNone(r["reconciled"]) + + def test_unknown_slot_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + with self.assertRaisesRegex(restore.RestoreError, "unknown slot"): + self.run_restore(d, manifest, log_dir, slots=["shard:99"]) + + def test_unknown_member_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + with self.assertRaisesRegex(restore.RestoreError, + "not in any current object"): + self.run_restore(d, manifest, log_dir, members=["nope/missing.bin"]) + + +class TestVerification(RestoreBase): + def test_md5_mismatch_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + corrupted = dict(STORE) + corrupted["sub/s1.bin"] = b"BITROT" + with self.assertRaisesRegex(restore.RestoreError, "MD5 MISMATCH"): + self.run_restore(d, manifest, log_dir, + extract=fake_extract(store=corrupted)) + + def test_missing_runlog_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + os.remove(os.path.join( + log_dir, manifest["objects"][0]["run_timestamp"] + ".json")) + with self.assertRaisesRegex(restore.RestoreError, "run-log not found"): + self.run_restore(d, manifest, log_dir) + + def test_member_missing_after_extract_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + + def forgetful(obj, arcnames, dest, tmp_dir): + fake_extract()(obj, [a for a in arcnames if a != "big.bin"], + dest, tmp_dir) + with self.assertRaisesRegex(restore.RestoreError, "missing from dest"): + self.run_restore(d, manifest, log_dir, extract=forgetful) + + +class TestDestGuard(RestoreBase): + def test_nonempty_dest_refused(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + dest = os.path.join(d, "dest") + os.makedirs(dest) + with open(os.path.join(dest, "live.txt"), "w") as fh: + fh.write("precious") + calls = [] + with self.assertRaisesRegex(restore.RestoreError, "NOT empty"): + self.run_restore(d, manifest, log_dir, + extract=fake_extract(calls=calls)) + self.assertEqual(calls, []) + with open(os.path.join(dest, "live.txt")) as fh: # untouched + self.assertEqual(fh.read(), "precious") + + def test_dest_is_a_file_refused(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + dest = os.path.join(d, "dest") + with open(dest, "w") as fh: + fh.write("x") + with self.assertRaisesRegex(restore.RestoreError, "not a directory"): + self.run_restore(d, manifest, log_dir) + + def test_empty_existing_dest_ok(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + os.makedirs(os.path.join(d, "dest")) + r = self.run_restore(d, manifest, log_dir) + self.assertTrue(r["reconciled"]) + + +class TestEmptyManifest(RestoreBase): + def test_no_objects_aborts(self): + with tempfile.TemporaryDirectory() as d: + manifest, log_dir = build_fixture(d) + manifest["objects"] = [] + with self.assertRaisesRegex(restore.RestoreError, "no objects"): + self.run_restore(d, manifest, log_dir) + + +if __name__ == "__main__": + unittest.main()