From c8db396b4681e4dbd93347fe88ff31909c0e308c Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Fri, 26 Jun 2026 14:16:44 -0400 Subject: [PATCH] feat(incremental): level-decision + manifest engine (PR 2, opt-in, unwired) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine core for leveled incremental backup (docs/RFC_incremental_v2.md §2-3). Gated behind a new config key 'incremental' (default false), so existing configs and the whole non-incremental path are byte-for-byte unchanged. The __main__ wiring that calls this lands in PR 2b (it touches the live archive path and needs the review workflow + a Negishi staging run before 'incremental' is ever enabled). Adds to archive.py (all LOCAL, not dill-safe -> never passed to run()): - classify_delta(): pure mtime+size diff -> delta / deletions_only / noop. - decide_level(): full L0 (first run / force / re-anchor past _MAX_DELTAS / when the prior chain is unreadable) vs a delta off the current L0. - effective_catalog(): reconstruct L0 + deltas (members then deletions, in run_timestamp order) from sibling catalogs; superseded-L0 chains ignored. - load/update_manifest(): atomic (tmp+fsync+os.replace); a corrupt manifest is preserved aside and the write refuses (never silently truncates the chain). - write/read_level_catalog(): per-level (mtime,size) catalogs, kept INSIDE _vault/ so reclaim's non-recursive *.json glob never sees them. - make_zip_files(only_arcnames=...): archive only a delta's member set. - load_config(): validate the 'incremental' gate. Hardened per an adversarial review workflow: catalogs moved into _vault/ (were scan-visible); corrupt-manifest protection + fsync; decide_level re-baselines instead of crashing on an unreadable prior catalog; removed the dead 'rebaseline' branch (byte-sum is redundant at per-file granularity) and documented the real same-size+same-mtime blind spot honestly. Tests: tests/test_incremental_engine.py (classify_delta incl. the blind spot, decide_level incl. deletions-only / re-anchor / 'z'-boundary / unreadable-prior fallback, effective_catalog folding incl. delete-then-readd, manifest atomicity + corrupt protection, _vault invisibility, only_arcnames, config gate). Full suite: 87 passing (was 54). Co-Authored-By: Claude Opus 4.8 --- archive.py | 253 ++++++++++++++++++++- tests/test_incremental_engine.py | 365 +++++++++++++++++++++++++++++++ 2 files changed, 617 insertions(+), 1 deletion(-) create mode 100644 tests/test_incremental_engine.py diff --git a/archive.py b/archive.py index bddec65..2dbeb87 100644 --- a/archive.py +++ b/archive.py @@ -109,6 +109,19 @@ def load_config(config_path): f"got: {config['cleanup_zip_on_success']!r}" ) + # incremental: opt-in leveled backup (L0 full + L1/L2… deltas). Default false + # keeps the historical whole-snapshot behaviour byte-for-byte. When true, a + # run consults the target's manifest and ships only changed/added files as a + # delta. The decision/manifest engine lives in this module (decide_level, + # update_manifest, …); the __main__ wiring that acts on it lands in a + # follow-up. See docs/RFC_incremental_v2.md. Leveling is LOCAL-mode only. + config.setdefault("incremental", False) + if not isinstance(config["incremental"], bool): + raise ValueError( + f"config 'incremental' must be true or false, " + f"got: {config['incremental']!r}" + ) + return config @@ -423,8 +436,235 @@ def decode_blonde(blonde): } +# --------------------------------------------------------------------------- +# Level decision + manifest engine (see docs/RFC_incremental_v2.md §2–3, PR 2). +# +# All LOCAL: these read/write log_dir, which exists only on the launching host +# (invariant #1). Leveling is local-mode only; the __main__ wiring (PR 2b) +# refuses it under --globus. classify_delta is split out pure for testing. +# The manifest lives at {log_dir}/_vault/{badge}.manifest.json — the _vault/ +# subdir is invisible to reclaim's non-recursive "*.json" glob (no reclaim +# edit forced). Per-level (mtime,size) catalogs live in _vault/ too, as +# {stem}.catalog.json — also hidden from reclaim's glob; effective_catalog folds +# them back. NOTE: the pure helpers here reference module-level names and are +# NOT dill-safe — LOCAL-only, never pass them to run()/gce.submit() (invariant #1). +# --------------------------------------------------------------------------- + +_VAULT_SUBDIR = "_vault" +_MAX_DELTAS = 35 # detail is one base36 char ('1'..'z'); past this, re-anchor a new L0 + + +def _vault_dir(log_dir): + import os + return os.path.join(log_dir, _VAULT_SUBDIR) + + +def manifest_path(log_dir, badge): + import os + return os.path.join(_vault_dir(log_dir), f"{badge}.manifest.json") + + +def load_manifest(log_dir, badge): + """Return the target's manifest dict, or None if absent/unreadable.""" + import json + try: + with open(manifest_path(log_dir, badge)) as fh: + return json.load(fh) + except (OSError, ValueError): + return None + + +def write_level_catalog(log_dir, log_stem, catalog): + """Persist a level's {arcname:(mtime_ns,size)} catalog as {stem}.catalog.json + INSIDE _vault/ (so it's hidden from reclaim's non-recursive "*.json" glob, + exactly like the manifest). Atomic (tmp + fsync + os.replace). Returns the + catalog_ref (basename) stored in the manifest node.""" + import os + import json + vd = _vault_dir(log_dir) + os.makedirs(vd, exist_ok=True) + ref = f"{log_stem}.catalog.json" + path = os.path.join(vd, ref) + tmp = path + ".tmp" + with open(tmp, "w") as fh: + json.dump({a: [mt, sz] for a, (mt, sz) in catalog.items()}, fh) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + return ref + + +def read_level_catalog(log_dir, catalog_ref): + """Load a _vault/{catalog_ref} catalog JSON back to {arcname:(mtime_ns,size)}.""" + import os + import json + with open(os.path.join(_vault_dir(log_dir), catalog_ref)) as fh: + raw = json.load(fh) + return {a: (v[0], v[1]) for a, v in raw.items()} + + +def _ordered_levels(manifest): + """The current L0 followed by its deltas in apply order (run_timestamp). + + Star topology: deltas reference their L0 (parent_blonde), not each other, so + run_timestamp is the only sequencer. Levels belonging to a superseded L0 are + ignored — only the current chain is reconstructed.""" + l0 = manifest.get("current_l0_blonde") + levels = [n for n in manifest.get("levels", []) + if n.get("level") == 0 and n.get("blonde") == l0] + deltas = [n for n in manifest.get("levels", []) + if n.get("level", 0) > 0 and n.get("parent_blonde") == l0] + deltas.sort(key=lambda n: n.get("run_timestamp", "")) + return levels + deltas + + +def effective_catalog(log_dir, manifest): + """Reconstruct the current effective {arcname:(mtime_ns,size)} from the L0 + and its deltas — apply each level's members, then its deletions, in order. + Returns (catalog, total_bytes).""" + catalog = {} + for node in _ordered_levels(manifest): + ref = node.get("catalog_ref") + if ref: + catalog.update(read_level_catalog(log_dir, ref)) + for arcname in node.get("deleted", []): + catalog.pop(arcname, None) + total = sum(sz for (_mt, sz) in catalog.values()) + return catalog, total + + +def _n_deltas(manifest): + """Number of deltas hanging off the current L0.""" + l0 = manifest.get("current_l0_blonde") + return sum(1 for n in manifest.get("levels", []) + if n.get("level", 0) > 0 and n.get("parent_blonde") == l0) + + +def next_detail(manifest): + """Single base36 detail char for the next delta off the current L0 + (L0 is '0'; first delta '1', then '2'…). Caller re-anchors past _MAX_DELTAS.""" + return _B36_DIGITS[_n_deltas(manifest) + 1] + + +def classify_delta(live_catalog, prior_catalog): + """PURE. Compare the live catalog against the reconstructed prior catalog + (both {arcname:(mtime_ns,size)}). Returns dict(kind, members, deleted): + + 'delta' — members (added ∪ modified) to ship, plus any deletions + 'deletions_only' — only removals; ships no tar + 'noop' — nothing detectably changed + + A file is 'modified' when its size differs OR its mtime (floored to whole + seconds via floor_s — sub-second jitter is not a change) differs. ACCEPTED + BLIND SPOT (docs/RFC_incremental_v2.md §1): an in-place rewrite preserving + BOTH size AND mtime is invisible here. A total-byte-sum check would add + nothing at this per-file granularity — a size change is already caught as a + per-file 'modified'. The operational mitigation is a periodic --full + re-baseline.""" + added = [a for a in live_catalog if a not in prior_catalog] + removed = [a for a in prior_catalog if a not in live_catalog] + modified = [a for a in live_catalog if a in prior_catalog and + (floor_s(live_catalog[a][0]) != floor_s(prior_catalog[a][0]) + or live_catalog[a][1] != prior_catalog[a][1])] + if not (added or modified or removed): + return {"kind": "noop", "members": [], "deleted": []} + members = sorted(set(added) | set(modified)) + if not members: + return {"kind": "deletions_only", "members": [], "deleted": sorted(removed)} + return {"kind": "delta", "members": members, "deleted": sorted(removed)} + + +def decide_level(log_dir, badge, live_catalog, manifest, force_full=False): + """Decide this run's level. Returns dict(kind, detail, parent, members, deleted): + + kind 'full' | 'delta' | 'deletions_only' | 'noop' + detail level char ('0' full; '1'..'z' delta; None for noop) + parent parent L0 blonde (None for a full) + members arcnames to archive (all matched for full; the change set for delta) + deleted arcnames removed since the prior level (deltas only) + + A full L0 is chosen when force_full, when no current L0 exists yet, or when + the chain has hit _MAX_DELTAS (re-anchor). The manifest is NOT mutated here — + the caller assembles + writes the level node after the transfer verifies.""" + has_l0 = bool( + manifest and manifest.get("current_l0_blonde") + and any(n.get("level") == 0 and n.get("blonde") == manifest["current_l0_blonde"] + for n in manifest.get("levels", [])) + ) + if force_full or not has_l0: + return {"kind": "full", "detail": "0", "parent": None, + "members": sorted(live_catalog), "deleted": []} + + try: + prior_catalog, _prior_bytes = effective_catalog(log_dir, manifest) + except (OSError, ValueError): + # Prior chain unreadable (a missing/corrupt sibling catalog) — re-baseline + # to a safe full L0 rather than refuse to back up. A backup engine must + # never decline to back up because old state is unreadable. + return {"kind": "full", "detail": "0", "parent": None, + "members": sorted(live_catalog), "deleted": []} + cls = classify_delta(live_catalog, prior_catalog) + if cls["kind"] == "noop": + return {"kind": "noop", "detail": None, + "parent": manifest["current_l0_blonde"], "members": [], "deleted": []} + + # Re-anchor once the chain is too deep to encode in one detail char. + if _n_deltas(manifest) + 1 > _MAX_DELTAS: + return {"kind": "full", "detail": "0", "parent": None, + "members": sorted(live_catalog), "deleted": []} + + return {"kind": cls["kind"], "detail": next_detail(manifest), + "parent": manifest["current_l0_blonde"], + "members": cls["members"], "deleted": cls["deleted"]} + + +def update_manifest(log_dir, badge, node): + """Append a level node to the target's manifest, atomically (tmp + os.replace). + Sets current_l0_blonde when node is an L0. Creates the manifest if absent; + replaces any existing node with the same blonde (idempotent re-runs). Returns + the manifest path. Best-effort by contract — the caller must treat a failure + as a warning, never failing an archive already verified on tape.""" + import os + import json + vd = _vault_dir(log_dir) + os.makedirs(vd, exist_ok=True) + path = manifest_path(log_dir, badge) + if os.path.exists(path): + try: + with open(path) as fh: + m = json.load(fh) + except (OSError, ValueError): + # Present but unreadable: NEVER silently overwrite the only on-disk + # record of the level chain with a fresh one (that would truncate the + # chain and lie). Preserve the bad file aside and refuse — the + # best-effort caller logs the warning; the archive is already on tape. + aside = path + ".corrupt" + try: + os.replace(path, aside) + except OSError: + pass + raise RuntimeError( + f"manifest {path} is corrupt (moved to {aside}); refusing to " + f"overwrite the level chain. Rebuild it from the run logs." + ) + else: + m = {"schema_version": 1, "badge": badge, "levels": []} + m["levels"] = [n for n in m.get("levels", []) if n.get("blonde") != node.get("blonde")] + m["levels"].append(node) + if node.get("level") == 0: + m["current_l0_blonde"] = node["blonde"] + tmp = path + ".tmp" + with open(tmp, "w") as fh: + json.dump(m, fh, indent=2) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + return path + + def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, - compression="deflate", allow_empty_files=False): + compression="deflate", allow_empty_files=False, + only_arcnames=None): """ On the RCAC compute system: find files in source_folder matching file_pattern (regex), zip them preserving relative paths, verify @@ -443,6 +683,14 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, 0-byte member. Set True to archive legitimately-empty files as-is (e.g. .gitkeep placeholders, empty __init__.py markers); they still get a valid (empty-string) MD5 and are counted, so reclaim's drift check stays exact. + + only_arcnames: when not None, restrict the archive to this set/list of + arcnames AFTER pattern matching — used to build an incremental delta zip + containing only the changed/added members chosen by decide_level. The + arcnames must be the same os.path.relpath keys make_zip_files itself + produces (invariant #2). None (default) = archive every matched file (the + full / L0 behaviour). Only the selected files are zipped and MD5'd, so + unchanged bulk is never re-read. """ import os import re @@ -454,12 +702,15 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, pattern = re.compile(file_pattern) source_folder = os.path.normpath(source_folder) + only = set(only_arcnames) if only_arcnames is not None else None for dirpath, dirnames, filenames in os.walk(source_folder): dirnames.sort() for fname in sorted(filenames): full_path = os.path.join(dirpath, fname) arcname = os.path.relpath(full_path, source_folder) if pattern.search(arcname): + if only is not None and arcname not in only: + continue # delta: skip members not in the change set matched_files.append((full_path, arcname)) if not matched_files: diff --git a/tests/test_incremental_engine.py b/tests/test_incremental_engine.py new file mode 100644 index 0000000..6569ec7 --- /dev/null +++ b/tests/test_incremental_engine.py @@ -0,0 +1,365 @@ +""" +Unit tests for the leveled-incremental decision + manifest engine (PR 2, +docs/RFC_incremental_v2.md §2-3): classify_delta, decide_level, effective_catalog, +the manifest read/append/atomic write, the sibling catalog round-trip, and the +make_zip_files only_arcnames filter + the incremental config gate. + +Pure logic + local-file I/O only — no tape, Slurm, or Globus. The __main__ +orchestration that calls these lands in PR 2b. +""" +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 archive # noqa: E402 + + +def cat(*items): + """Build a catalog {arcname: (mtime_ns, size)} from (arcname, mtime_s, size).""" + return {a: (mt * 1_000_000_000, sz) for (a, mt, sz) in items} + + +class TestClassifyDelta(unittest.TestCase): + def test_noop_identical(self): + c = cat(("a", 100, 10), ("b", 100, 20)) + self.assertEqual(archive.classify_delta(c, c)["kind"], "noop") + + def test_added(self): + prior = cat(("a", 100, 10)) + live = cat(("a", 100, 10), ("new", 200, 5)) + r = archive.classify_delta(live, prior) + self.assertEqual(r["kind"], "delta") + self.assertEqual(r["members"], ["new"]) + self.assertEqual(r["deleted"], []) + + def test_modified_by_size(self): + # The size-changing in-place rewrite: caught as a per-file 'modified' + # (this is why a separate byte-sum check is redundant here). + prior = cat(("a", 100, 10)) + live = cat(("a", 100, 12)) # same mtime, bigger + r = archive.classify_delta(live, prior) + self.assertEqual(r["kind"], "delta") + self.assertEqual(r["members"], ["a"]) + + def test_modified_by_mtime(self): + prior = cat(("a", 100, 10)) + live = cat(("a", 200, 10)) # newer mtime, same size + self.assertEqual(archive.classify_delta(live, prior)["members"], ["a"]) + + def test_subsecond_jitter_not_modified(self): + prior = {"a": (100_000_000_001, 10)} + live = {"a": (100_999_999_999, 10)} # same whole second, same size + self.assertEqual(archive.classify_delta(live, prior)["kind"], "noop") + + def test_same_size_same_mtime_is_blind_spot(self): + # Accepted blind spot (RFC §1): a content rewrite preserving BOTH size + # AND mtime is invisible — classified noop. Documented, not a bug. + prior = {"a": (100_000_000_000, 10)} + live = {"a": (100_000_000_000, 10)} + self.assertEqual(archive.classify_delta(live, prior)["kind"], "noop") + + def test_deletions_only(self): + prior = cat(("a", 100, 10), ("gone", 100, 7)) + live = cat(("a", 100, 10)) + r = archive.classify_delta(live, prior) + self.assertEqual(r["kind"], "deletions_only") + self.assertEqual(r["members"], []) + self.assertEqual(r["deleted"], ["gone"]) + + def test_added_and_deleted_together(self): + prior = cat(("a", 100, 10), ("gone", 100, 7)) + live = cat(("a", 100, 10), ("new", 200, 3)) + r = archive.classify_delta(live, prior) + self.assertEqual(r["kind"], "delta") + self.assertEqual(r["members"], ["new"]) + self.assertEqual(r["deleted"], ["gone"]) + + +class TestEffectiveCatalog(unittest.TestCase): + def _manifest_with(self, d, nodes, current_l0): + return {"schema_version": 1, "badge": "bdg", + "current_l0_blonde": current_l0, "levels": nodes} + + def test_l0_only(self): + with tempfile.TemporaryDirectory() as d: + ref = archive.write_level_catalog(d, "stem0", cat(("a", 100, 10), ("b", 100, 20))) + m = self._manifest_with(d, [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": ref, "deleted": [], + "run_timestamp": "p_20260101_000000"}], "L0") + eff, total = archive.effective_catalog(d, m) + self.assertEqual(set(eff), {"a", "b"}) + self.assertEqual(total, 30) + + def test_l0_plus_delta_add_and_modify(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + r1 = archive.write_level_catalog(d, "s1", cat(("a", 200, 15), ("new", 200, 5))) + m = self._manifest_with(d, [ + {"blonde": "L0", "level": 0, "parent_blonde": None, "catalog_ref": r0, + "deleted": [], "run_timestamp": "p_20260101_000000"}, + {"blonde": "L1", "level": 1, "parent_blonde": "L0", "catalog_ref": r1, + "deleted": [], "run_timestamp": "p_20260102_000000"}, + ], "L0") + eff, total = archive.effective_catalog(d, m) + self.assertEqual(eff["a"], (200 * 1_000_000_000, 15)) # delta overrode L0 + self.assertEqual(set(eff), {"a", "new"}) + self.assertEqual(total, 20) + + def test_delete_then_readd_ordering(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + r1 = archive.write_level_catalog(d, "s1", {}) # deletions-only + r2 = archive.write_level_catalog(d, "s2", cat(("a", 300, 99))) + m = self._manifest_with(d, [ + {"blonde": "L0", "level": 0, "parent_blonde": None, "catalog_ref": r0, + "deleted": [], "run_timestamp": "p_20260101_000000"}, + {"blonde": "L1", "level": 1, "parent_blonde": "L0", "catalog_ref": r1, + "deleted": ["a"], "run_timestamp": "p_20260102_000000"}, + {"blonde": "L2", "level": 2, "parent_blonde": "L0", "catalog_ref": r2, + "deleted": [], "run_timestamp": "p_20260103_000000"}, + ], "L0") + eff, total = archive.effective_catalog(d, m) + self.assertEqual(eff["a"], (300 * 1_000_000_000, 99)) # re-added wins + self.assertEqual(total, 99) + + def test_superseded_l0_ignored(self): + with tempfile.TemporaryDirectory() as d: + old = archive.write_level_catalog(d, "old", cat(("x", 1, 1))) + new = archive.write_level_catalog(d, "new", cat(("a", 100, 10))) + m = self._manifest_with(d, [ + {"blonde": "OLD", "level": 0, "parent_blonde": None, "catalog_ref": old, + "deleted": [], "run_timestamp": "p_20260101_000000"}, + {"blonde": "NEW", "level": 0, "parent_blonde": None, "catalog_ref": new, + "deleted": [], "run_timestamp": "p_20260201_000000"}, + ], "NEW") + eff, _ = archive.effective_catalog(d, m) + self.assertEqual(set(eff), {"a"}) # only the current L0 chain + + +class TestDecideLevel(unittest.TestCase): + def test_first_run_is_full(self): + with tempfile.TemporaryDirectory() as d: + live = cat(("a", 100, 10), ("b", 100, 20)) + r = archive.decide_level(d, "bdg", live, None, force_full=False) + self.assertEqual(r["kind"], "full") + self.assertEqual(r["detail"], "0") + self.assertIsNone(r["parent"]) + self.assertEqual(r["members"], ["a", "b"]) + + def test_force_full_when_l0_exists(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", + "levels": [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": r0, "deleted": [], "run_timestamp": "p_1"}]} + r = archive.decide_level(d, "bdg", cat(("a", 100, 10)), m, force_full=True) + self.assertEqual(r["kind"], "full") + self.assertEqual(r["detail"], "0") + + def test_change_yields_delta(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", + "levels": [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": r0, "deleted": [], "run_timestamp": "p_1"}]} + live = cat(("a", 100, 10), ("b", 200, 5)) + r = archive.decide_level(d, "bdg", live, m, force_full=False) + self.assertEqual(r["kind"], "delta") + self.assertEqual(r["detail"], "1") + self.assertEqual(r["parent"], "L0") + self.assertEqual(r["members"], ["b"]) + + def test_noop_when_unchanged(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", + "levels": [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": r0, "deleted": [], "run_timestamp": "p_1"}]} + r = archive.decide_level(d, "bdg", cat(("a", 100, 10)), m, force_full=False) + self.assertEqual(r["kind"], "noop") + self.assertIsNone(r["detail"]) + + def test_reanchor_past_max_deltas(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + levels = [{"blonde": "L0", "level": 0, "parent_blonde": None, "catalog_ref": r0, + "deleted": [], "run_timestamp": "p_0"}] + for i in range(1, archive._MAX_DELTAS + 1): # fill the chain to the cap + levels.append({"blonde": f"D{i}", "level": i, "parent_blonde": "L0", + "catalog_ref": r0, "deleted": [], "run_timestamp": f"p_{i:03d}"}) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", "levels": levels} + live = cat(("a", 100, 10), ("b", 999, 5)) # a change is present + r = archive.decide_level(d, "bdg", live, m, force_full=False) + self.assertEqual(r["kind"], "full") # re-anchored, not a 36th delta + + def test_deletions_only_through_decide_level(self): + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10), ("gone", 100, 7))) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", + "levels": [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": r0, "deleted": [], "run_timestamp": "p_1"}]} + live = cat(("a", 100, 10)) # 'gone' removed + r = archive.decide_level(d, "bdg", live, m, force_full=False) + self.assertEqual(r["kind"], "deletions_only") + self.assertEqual(r["detail"], "1") + self.assertEqual(r["parent"], "L0") + self.assertEqual(r["members"], []) + self.assertEqual(r["deleted"], ["gone"]) + + def test_last_delta_detail_char_is_z_at_cap(self): + # Pins the encoding contract: with _MAX_DELTAS-1 deltas present the next + # delta is the last representable char ('z'); the re-anchor guard (tested + # above) is what prevents indexing past it. + with tempfile.TemporaryDirectory() as d: + r0 = archive.write_level_catalog(d, "s0", cat(("a", 100, 10))) + levels = [{"blonde": "L0", "level": 0, "parent_blonde": None, "catalog_ref": r0, + "deleted": [], "run_timestamp": "p_000"}] + for i in range(1, archive._MAX_DELTAS): # 34 deltas -> _n_deltas == 34 + levels.append({"blonde": f"D{i}", "level": i, "parent_blonde": "L0", + "catalog_ref": r0, "deleted": [], "run_timestamp": f"p_{i:03d}"}) + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", "levels": levels} + live = cat(("a", 200, 10)) # a change is present + r = archive.decide_level(d, "bdg", live, m, force_full=False) + self.assertEqual(r["kind"], "delta") + self.assertEqual(r["detail"], "z") # _B36_DIGITS[35] + + def test_unreadable_prior_catalog_falls_back_to_full(self): + with tempfile.TemporaryDirectory() as d: + m = {"schema_version": 1, "badge": "bdg", "current_l0_blonde": "L0", + "levels": [{"blonde": "L0", "level": 0, "parent_blonde": None, + "catalog_ref": "missing.catalog.json", # never written + "deleted": [], "run_timestamp": "p_1"}]} + r = archive.decide_level(d, "bdg", cat(("a", 100, 10)), m, force_full=False) + self.assertEqual(r["kind"], "full") # safe re-baseline, not a crash + + +class TestVaultHiddenFromReclaim(unittest.TestCase): + def test_manifest_and_catalog_invisible_to_reclaim_glob(self): + # reclaim's find_logs is a NON-recursive glob of log_dir/*.json. Both the + # manifest and the per-level catalogs must live under _vault/ so a --scan + # never picks them up as spurious archives. + import glob + with tempfile.TemporaryDirectory() as d: + archive.write_level_catalog(d, "p_20260101_000000", cat(("a", 100, 10))) + archive.update_manifest(d, "bdg", {"blonde": "L0", "level": 0, + "parent_blonde": None, "deleted": []}) + self.assertEqual(glob.glob(os.path.join(d, "*.json")), []) # nothing at top level + self.assertTrue(glob.glob(os.path.join(d, "_vault", "*.json"))) # both hidden inside + + +class TestManifestIO(unittest.TestCase): + def test_append_and_roundtrip_sets_current_l0(self): + with tempfile.TemporaryDirectory() as d: + l0 = {"blonde": "B.x.0.B", "level": 0, "parent_blonde": None, "deleted": []} + archive.update_manifest(d, "bdg", l0) + m = archive.load_manifest(d, "bdg") + self.assertEqual(m["current_l0_blonde"], "B.x.0.B") + self.assertEqual(len(m["levels"]), 1) + # manifest lives in the _vault subdir (invisible to reclaim's glob) + self.assertTrue(os.path.exists( + os.path.join(d, "_vault", "bdg.manifest.json"))) + + def test_append_delta_keeps_l0(self): + with tempfile.TemporaryDirectory() as d: + archive.update_manifest(d, "bdg", {"blonde": "L0", "level": 0, + "parent_blonde": None, "deleted": []}) + archive.update_manifest(d, "bdg", {"blonde": "L1", "level": 1, + "parent_blonde": "L0", "deleted": []}) + m = archive.load_manifest(d, "bdg") + self.assertEqual(m["current_l0_blonde"], "L0") # unchanged by the delta + self.assertEqual(len(m["levels"]), 2) + + def test_idempotent_same_blonde(self): + with tempfile.TemporaryDirectory() as d: + node = {"blonde": "L0", "level": 0, "parent_blonde": None, "deleted": []} + archive.update_manifest(d, "bdg", node) + archive.update_manifest(d, "bdg", node) # re-run, same blonde + m = archive.load_manifest(d, "bdg") + self.assertEqual(len(m["levels"]), 1) + + def test_load_missing_is_none(self): + with tempfile.TemporaryDirectory() as d: + self.assertIsNone(archive.load_manifest(d, "nope")) + + def test_catalog_roundtrip(self): + with tempfile.TemporaryDirectory() as d: + c = cat(("a/b.bin", 123, 7)) + ref = archive.write_level_catalog(d, "stem", c) + self.assertEqual(archive.read_level_catalog(d, ref), c) + + def test_corrupt_manifest_preserved_not_overwritten(self): + # A corrupt existing manifest must NOT be silently reset to a fresh one + # (that would truncate the level chain and lie). It's preserved aside and + # the write refuses. + with tempfile.TemporaryDirectory() as d: + archive.update_manifest(d, "bdg", {"blonde": "L0", "level": 0, + "parent_blonde": None, "deleted": []}) + path = archive.manifest_path(d, "bdg") + with open(path, "w") as fh: + fh.write("{ this is not valid json") + with self.assertRaises(RuntimeError): + archive.update_manifest(d, "bdg", {"blonde": "L1", "level": 1, + "parent_blonde": "L0", "deleted": []}) + self.assertTrue(os.path.exists(path + ".corrupt")) # bad file kept for recovery + self.assertFalse(os.path.exists(path)) # not replaced with a lie + + +class TestOnlyArcnames(unittest.TestCase): + FILES = [("raw/a.bin", b"aaa"), ("raw/b.bin", b"bbbb"), ("top.txt", b"x")] + + def _tree(self, d): + os.makedirs(os.path.join(d, "raw")) + for rel, data in self.FILES: + with open(os.path.join(d, rel), "wb") as fh: + fh.write(data) + + def test_only_subset_zipped(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._tree(src) + _, _, file_checksums, members, _ = archive.make_zip_files( + src, ".*", tmp, "p", only_arcnames=["raw/a.bin"]) + self.assertEqual(set(members), {"raw/a.bin"}) + self.assertEqual(set(file_checksums), {"raw/a.bin"}) + + def test_none_zips_everything(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._tree(src) + _, _, _, members, _ = archive.make_zip_files(src, ".*", tmp, "p") + self.assertEqual(set(members), {"raw/a.bin", "raw/b.bin", "top.txt"}) + + def test_empty_subset_raises(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._tree(src) + with self.assertRaises(RuntimeError): + archive.make_zip_files(src, ".*", tmp, "p", only_arcnames=["nope"]) + + +class TestIncrementalConfigGate(unittest.TestCase): + def _write(self, d, **extra): + cfg = {"source_folder": "/s", "file_pattern": ".*", "fortress_base_dir": "/f", + "emails": ["a@b.c"], "project_name": "p"} + cfg.update(extra) + p = os.path.join(d, "c.json") + with open(p, "w") as fh: + json.dump(cfg, fh) + return p + + def test_default_false(self): + with tempfile.TemporaryDirectory() as d: + self.assertFalse(archive.load_config(self._write(d))["incremental"]) + + def test_true_accepted(self): + with tempfile.TemporaryDirectory() as d: + self.assertTrue(archive.load_config(self._write(d, incremental=True))["incremental"]) + + def test_non_bool_rejected(self): + with tempfile.TemporaryDirectory() as d: + with self.assertRaises(ValueError): + archive.load_config(self._write(d, incremental="yes")) + + +if __name__ == "__main__": + unittest.main()