From 0f0cb99f5c6b1202c8e25f0e3d97c5df181ecc63 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Wed, 8 Jul 2026 09:40:39 -0400 Subject: [PATCH] feat(routing): Phase-2 size-routing primitives (slice 1, behavior-neutral) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the pure, additive size-routing helpers to archive.py — nothing wires them into the archive or reclaim flow yet (behavior-neutral; full suite 136 -> 172 green): - route(catalog, t_small, shard_count) -> [{slot, kind, arcnames}] partition: files >= T_small become solo objects; smaller files bucket into content-addressed shards k = shake_128(arcname) mod K (no repack cascade -> cheap APPEND). - shard_of / solo_slot / shard_slot / derive_shard_count (K = ceil(small_bytes / shard_target), frozen per badge at baseline). - object_fingerprint (solo {mtime_s,size}; shard {members,byte_sum,digest}) + object_unchanged (the skip/append signal, also the resume mechanism). - clamp_t_small (100 MB floor), budget_warnings (warn-only max_solo/max_objects/ oversize-shard), vault_dir_for/manifest_path/load_manifest/write_manifest (atomic tmp+os.replace, best-effort; _vault child dodges reclaim's glob). - make_zip_files gains an optional only_arcnames filter (no caller yet) so a later slice can ship exactly one routed object; keys stay arcname-relpath (invariant #2). Tests: tests/test_routing_primitives.py (36 cases) — boundary partition, shard stability, K frozen, fingerprint change detection incl. reshuffle, skip vs shard-append, only_arcnames subset with matching checksum keys, manifest round-trip. See docs/RFC_incremental_v2.md §2 (slice 1). Co-Authored-By: Claude Opus 4.8 --- archive.py | 267 ++++++++++++++++++++++++++++- tests/test_routing_primitives.py | 283 +++++++++++++++++++++++++++++++ 2 files changed, 549 insertions(+), 1 deletion(-) create mode 100644 tests/test_routing_primitives.py diff --git a/archive.py b/archive.py index c12eb9c..c168d6c 100644 --- a/archive.py +++ b/archive.py @@ -548,6 +548,254 @@ def decode_blonde(blonde): } +# --------------------------------------------------------------------------- +# Size-routing primitives (Phase 2 — see docs/RFC_incremental_v2.md §2). +# +# Partition a target's survivors into independent Fortress OBJECTS: +# - solo: one file >= T_small, its own object, skip-if-unchanged. +# - shard: files < T_small bucketed by shard = hash(relpath) mod K +# (K frozen per badge at baseline), whole-shard re-ship on change. +# Content-addressed => a file's bucket is a pure function of its relpath, stable +# every run, so adding/changing one file touches exactly one object (no repack +# cascade) — this is what makes APPEND cheap. +# +# Pure, additive, deterministic helpers. NOTHING wires them into the archive or +# reclaim flow yet — this slice is BEHAVIOR-NEUTRAL. Later slices add +# route_and_ship (archive), verify_target (reclaim aggregate), the reship-monitor, +# and restore.py. Routing is LOCAL-only (a later slice refuses --globus). Imports +# stay inside the bodies to match the file's convention. +# --------------------------------------------------------------------------- + +# Defaults the RFC (§2.2) cites; floor/budgets are enforced/emitted by callers. +T_SMALL_DEFAULT = 256 * 1024 * 1024 # 256 MiB — solo/shard cutoff +T_SMALL_FLOOR = 100 * 1000 * 1000 # 100 MB — COS disk->tape line; hard floor +SHARD_TARGET_DEFAULT = 256 * 1024 ** 3 # 256 GiB — target uncompressed shard size +MAX_SOLO_DEFAULT = 500 +MAX_OBJECTS_DEFAULT = 1000 + + +def clamp_t_small(t_small): + """ + Raise a configured T_small to the 100 MB floor if needed (RFC §2.2): below it a + solo lands in the Fortress COS-10 disk cache — tape-fronted, so not unsafe, just + inefficient. Returns an int >= T_SMALL_FLOOR. + """ + return max(int(t_small), T_SMALL_FLOOR) + + +def shard_of(arcname, shard_count): + """ + Content-addressed bucket for a relpath: k = int(shake_128(arcname)) mod K, in + [0, shard_count). PURE + STABLE — same arcname and same K always give the same k, + every run, so adding/changing one file touches exactly one shard (no repack + cascade). K is frozen per badge at baseline (recorded in the manifest); never + recompute it from a changed shard_count without a --fresh re-baseline. + """ + import hashlib + + if shard_count < 1: + raise ValueError("shard_count must be >= 1") + h = hashlib.shake_128(arcname.encode("utf-8")).digest(8) + return int.from_bytes(h, "big") % shard_count + + +def solo_slot(arcname): + """ + Stable logical slot key for a solo object: 'solo:', hex = 64-bit shake_128 + of the relpath. Stable per file path across runs (a solo whose bytes change + re-ships to a new tar under the SAME slot). 64 bits so hundreds of solos never + collide in practice (birthday ~ n^2 / 2^65). + """ + import hashlib + + return "solo:" + hashlib.shake_128(arcname.encode("utf-8")).hexdigest(8) + + +def shard_slot(k): + """Slot key for shard k: 'shard:'.""" + return "shard:{}".format(int(k)) + + +def derive_shard_count(catalog, t_small=T_SMALL_DEFAULT, + shard_target=SHARD_TARGET_DEFAULT): + """ + Baseline K for a target: ceil(small-tier bytes / shard_target), floored at 1. + Computed ONCE at baseline from the live catalog, then FROZEN in the manifest + (immutable for the badge's life — a content-addressed map must be stable). The + catalog is {arcname: (mtime_ns, size)} from enumerate_source_catalog. + """ + small_bytes = sum(size for (_m, size) in catalog.values() if size < t_small) + if small_bytes <= 0: + return 1 + return max(1, -(-small_bytes // shard_target)) # ceil division, integer-only + + +def route(catalog, t_small=T_SMALL_DEFAULT, shard_count=None, + shard_target=SHARD_TARGET_DEFAULT): + """ + Partition a catalog into objects. Returns a list of dicts, deterministic order + (solos by slot, then shards by k): + + {"slot": str, "kind": "solo"|"shard", "arcnames": [relpath, ...]} + + Files >= t_small each become a solo object; smaller files bucket into shards by + shard_of(arcname, K). Empty buckets are omitted (nothing to ship); K itself is + fixed, so an empty bucket that later gains a file simply reappears. If shard_count + is None it is derived (derive_shard_count) — callers WITH an existing manifest MUST + pass the frozen K from it instead of re-deriving. + """ + if shard_count is None: + shard_count = derive_shard_count(catalog, t_small, shard_target) + + solos, buckets = [], {} + for arcname in sorted(catalog): + size = catalog[arcname][1] + if size >= t_small: + solos.append({"slot": solo_slot(arcname), "kind": "solo", + "arcnames": [arcname]}) + else: + buckets.setdefault(shard_of(arcname, shard_count), []).append(arcname) + + # Guard the (astronomically unlikely) solo-slot collision rather than silently + # merge two distinct files into one slot. + slots = [o["slot"] for o in solos] + if len(set(slots)) != len(slots): + raise RuntimeError("solo slot collision — two relpaths hashed to one solo " + "slot; should be impossible at 64 bits") + + solos.sort(key=lambda o: o["slot"]) + shards = [{"slot": shard_slot(k), "kind": "shard", + "arcnames": sorted(buckets[k])} for k in sorted(buckets)] + return solos + shards + + +def object_fingerprint(kind, arcnames, catalog): + """ + Compact fingerprint of an object's CURRENT content, from the live catalog. + Change detection = recompute this and compare against the manifest's stored + fingerprint for the slot; any difference re-ships the object. + + - solo: {"mtime_s", "size"} of the single file. + - shard: {"members", "byte_sum", "digest"} where digest = shake_128 over the + sorted (arcname, floor_s(mtime), size) triples — the authoritative + change signal (catches add/remove/modify/size, and a same-count/ + same-bytes reshuffle a pure sum would miss). members/byte_sum are + readable cross-checks. + + BLIND SPOT (documented, shared with reclaim): a same-size + same-mtime in-place + rewrite defeats both fingerprints — mitigated by periodic re-hash + --fresh + (RFC §2.3). Floors mtimes to whole seconds (floor_s) so sub-second jitter never + reads as a change. + """ + import hashlib + + if kind == "solo": + mtime_ns, size = catalog[arcnames[0]] + return {"mtime_s": floor_s(mtime_ns), "size": size} + + items = sorted(arcnames) + parts, byte_sum = [], 0 + for a in items: + mtime_ns, size = catalog[a] + parts.append("{}\0{}\0{}".format(a, floor_s(mtime_ns), size)) + byte_sum += size + digest = hashlib.shake_128("\n".join(parts).encode("utf-8")).hexdigest(16) + return {"members": len(items), "byte_sum": byte_sum, "digest": digest} + + +def object_unchanged(current_object, stored_object, catalog): + """ + True iff a routed object can be SKIPPED this run — its live content matches what + the manifest recorded for that slot (the append win + resume, RFC §2.3/§2.8). + + current_object is a fresh route() entry (so its arcnames reflect live membership, + including a small file just appended to a shard); stored_object is the manifest + entry for the same slot (or None at baseline). Compares recomputed vs stored + fingerprint. A brand-new slot (stored_object is None) is never unchanged. + """ + if not stored_object: + return False + fp = object_fingerprint(current_object["kind"], + current_object["arcnames"], catalog) + return fp == stored_object.get("fingerprint") + + +def budget_warnings(objects, catalog, shard_target=SHARD_TARGET_DEFAULT, + max_solo=MAX_SOLO_DEFAULT, max_objects=MAX_OBJECTS_DEFAULT): + """ + Warn-only object-count / shard-size guards (RFC §2.2). Returns a list of + human-readable strings; NEVER raises and never blocks a run — routing proceeds + regardless (the operator raises shard_target / re-baselines in response). + """ + warnings = [] + n_solo = sum(1 for o in objects if o["kind"] == "solo") + n_total = len(objects) + if n_solo > max_solo: + warnings.append("solo objects {} exceed max_solo {}".format(n_solo, max_solo)) + if n_total > max_objects: + warnings.append("total objects {} exceed max_objects_per_target {}" + .format(n_total, max_objects)) + limit = 2 * shard_target + for o in objects: + if o["kind"] != "shard": + continue + b = sum(catalog[a][1] for a in o["arcnames"] if a in catalog) + if b > limit: + warnings.append("shard {} is {} bytes (> 2x shard_target {}) — raise " + "shard_target and re-baseline (--fresh)" + .format(o["slot"], b, shard_target)) + return warnings + + +def vault_dir_for(log_dir): + """ + The per-deployment manifest vault: {log_dir}/_vault. The '_vault' child is + invisible to reclaim.find_logs' non-recursive *.json glob, so manifests never + collide with run-log scanning (RFC §2.4). + """ + import os + + return os.path.join(log_dir, "_vault") + + +def manifest_path(vault_dir, badge): + """Path to a target's manifest file inside the vault.""" + import os + + return os.path.join(vault_dir, "{}.manifest.json".format(badge)) + + +def load_manifest(vault_dir, badge): + """Load a target's manifest dict, or None if it has none yet (baseline).""" + import os + import json + + path = manifest_path(vault_dir, badge) + if not os.path.exists(path): + return None + with open(path) as fh: + return json.load(fh) + + +def write_manifest(vault_dir, badge, manifest): + """ + Atomically write a target's manifest (tmp + os.replace). Callers treat it as + best-effort (a manifest write must never fail an archive already verified on + tape); the atomic replace guarantees a reader never sees a half-written file. + Returns the written path. + """ + import os + import json + + os.makedirs(vault_dir, exist_ok=True) + path = manifest_path(vault_dir, badge) + tmp = path + ".tmp" + with open(tmp, "w") as fh: + json.dump(manifest, fh, indent=2, sort_keys=True) + os.replace(tmp, path) + return path + + # --------------------------------------------------------------------------- # Exclusion primitives (see docs/EXCLUSION_SPEC.md). # @@ -1186,7 +1434,7 @@ def evaluate_exclusions(matched_members, source_folder, spec, def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, compression="deflate", allow_empty_files=False, - exclude_spec=None, exclude_optional=()): + exclude_spec=None, exclude_optional=(), only_arcnames=None): """ On the RCAC compute system: find files in source_folder matching file_pattern (regex), zip them preserving relative paths, verify @@ -1213,6 +1461,12 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, exclude_optional is the per-asset Tier-2 opt-out list. evaluate_exclusions NEVER raises: any problem yields "back up everything" with a recorded reason. + only_arcnames: optional iterable of relpaths (Phase 2 size-routing, RFC §2.5). + When given, the matched set is narrowed to exactly these members — one routed + object (a solo's single file, or a shard's bucket). The router applies exclusion + LOCALLY before routing, so a routed call passes exclude_spec=None. Default None = + unchanged whole-target behavior; no caller supplies it yet (behavior-neutral). + Returns (zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion) — the 6th element is the exclusion report (status OFF/DISABLED/ON + excluded/demoted/counts) for the run log, banner, and completion email. @@ -1236,11 +1490,22 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, if pattern.search(arcname): matched_files.append((full_path, arcname)) + # Size-routing (Phase 2, RFC §2.5): when only_arcnames is given, keep just that + # object's members (a solo's one file, or a shard's bucket). The set is already + # exclusion-applied by the LOCAL router, so the caller passes exclude_spec=None + # and the git gate is not re-run per object. No caller passes this yet — routing + # wiring lands in a later slice; here it is inert unless explicitly supplied. + if only_arcnames is not None: + keep = set(only_arcnames) + matched_files = [(fp, arc) for (fp, arc) in matched_files if arc in keep] + if not matched_files: raise RuntimeError( f"No files matched pattern '{file_pattern}' in '{source_folder}' " f"(searched recursively — pattern is tested against relative paths " f"such as 'raw/IMG_001.raw')" + + ("" if only_arcnames is None else + " — after the only_arcnames routing filter (empty object)") ) # Second, subtractive filter (docs/EXCLUSION_SPEC.md): drop git-COVERED diff --git a/tests/test_routing_primitives.py b/tests/test_routing_primitives.py new file mode 100644 index 0000000..6cb5d4e --- /dev/null +++ b/tests/test_routing_primitives.py @@ -0,0 +1,283 @@ +""" +Unit tests for the Phase-2 size-routing primitives (docs/RFC_incremental_v2.md §2, +build-order slice 1). These are pure, additive, deterministic helpers — nothing in +the archive or reclaim flow calls them yet (slice 1 is behavior-neutral), so this +suite is fully self-contained (no tape, no Slurm, no Globus). + +Covers: shard_of stability, slot encoding, derive_shard_count, route() partition at +the T_small boundary, object_fingerprint change detection, object_unchanged (the +skip/append signal), budget warnings, clamp_t_small, manifest round-trip, and the +new (unused) only_arcnames filter on make_zip_files. +""" +import hashlib +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 + + +# A small live-style catalog: {arcname: (mtime_ns, size)}. +MiB = 1024 * 1024 +BIG = 300 * MiB # >= 256 MiB default T_small -> solo +SMALL = 4 * MiB # < T_small -> shard + + +def cat(**entries): + """Build a catalog from name=(mtime_s, size) kwargs (mtime given in seconds).""" + return {a: (mt * 1_000_000_000, sz) for a, (mt, sz) in entries.items()} + + +class TestShardOf(unittest.TestCase): + def test_stable_across_calls(self): + self.assertEqual(archive.shard_of("a/b/c.txt", 8), + archive.shard_of("a/b/c.txt", 8)) + + def test_in_range(self): + for i in range(200): + k = archive.shard_of("file_%d.dat" % i, 8) + self.assertTrue(0 <= k < 8) + + def test_distributes(self): + seen = {archive.shard_of("file_%d.dat" % i, 8) for i in range(200)} + self.assertGreater(len(seen), 1) # not all in one bucket + + def test_k1_all_zero(self): + self.assertEqual(archive.shard_of("anything", 1), 0) + + def test_rejects_bad_k(self): + with self.assertRaises(ValueError): + archive.shard_of("x", 0) + + +class TestSlots(unittest.TestCase): + def test_solo_slot_format_and_stable(self): + s = archive.solo_slot("aligned/sample.bam") + self.assertTrue(s.startswith("solo:")) + self.assertEqual(len(s), len("solo:") + 16) # 8-byte hexdigest = 16 chars + self.assertEqual(s, archive.solo_slot("aligned/sample.bam")) + + def test_solo_slot_distinct(self): + self.assertNotEqual(archive.solo_slot("a"), archive.solo_slot("b")) + + def test_shard_slot_format(self): + self.assertEqual(archive.shard_slot(3), "shard:3") + + +class TestDeriveShardCount(unittest.TestCase): + def test_ceil_of_small_bytes(self): + # 3 small files of 100 bytes, shard_target 250 -> ceil(300/250) = 2 + c = cat(a=(1, 100), b=(1, 100), d=(1, 100)) + self.assertEqual(archive.derive_shard_count(c, t_small=BIG, shard_target=250), 2) + + def test_big_files_excluded_from_small_bytes(self): + # one big (solo) + one tiny: only the tiny counts toward K + c = cat(big=(1, BIG), tiny=(1, 10)) + self.assertEqual(archive.derive_shard_count(c, shard_target=archive.SHARD_TARGET_DEFAULT), 1) + + def test_floor_at_one_when_no_small(self): + c = cat(big=(1, BIG)) + self.assertEqual(archive.derive_shard_count(c), 1) + + +class TestRoute(unittest.TestCase): + def test_boundary_solo_vs_shard(self): + # exactly T_small default is a solo (>=); one byte under is a shard + c = cat(at=(1, archive.T_SMALL_DEFAULT), + under=(1, archive.T_SMALL_DEFAULT - 1)) + objs = archive.route(c, shard_count=4) + solos = [o for o in objs if o["kind"] == "solo"] + shards = [o for o in objs if o["kind"] == "shard"] + self.assertEqual([o["arcnames"] for o in solos], [["at"]]) + self.assertEqual(sum(len(o["arcnames"]) for o in shards), 1) + self.assertIn("under", shards[0]["arcnames"]) + + def test_empty_shards_omitted(self): + c = cat(x=(1, SMALL)) # one small file, K=8 -> exactly one shard object + objs = archive.route(c, shard_count=8) + shards = [o for o in objs if o["kind"] == "shard"] + self.assertEqual(len(shards), 1) + + def test_membership_matches_shard_of(self): + c = cat(a=(1, SMALL), b=(1, SMALL), d=(1, SMALL)) + objs = archive.route(c, shard_count=4) + for o in objs: + if o["kind"] == "shard": + k = int(o["slot"].split(":")[1]) + for arc in o["arcnames"]: + self.assertEqual(archive.shard_of(arc, 4), k) + + def test_deterministic_and_frozen_k(self): + c = cat(a=(1, SMALL), b=(1, SMALL), big=(1, BIG)) + self.assertEqual(archive.route(c, shard_count=4), + archive.route(c, shard_count=4)) + + def test_all_members_present_exactly_once(self): + c = cat(a=(1, SMALL), b=(1, BIG), d=(1, SMALL), e=(1, BIG)) + objs = archive.route(c, shard_count=4) + flat = [arc for o in objs for arc in o["arcnames"]] + self.assertEqual(sorted(flat), sorted(c)) + self.assertEqual(len(flat), len(set(flat))) # disjoint + + +class TestFingerprint(unittest.TestCase): + def test_solo_floors_mtime(self): + c = {"f": (1_500_000_000_999_999_999, 10)} # sub-second ns tail + fp = archive.object_fingerprint("solo", ["f"], c) + self.assertEqual(fp, {"mtime_s": 1_500_000_000, "size": 10}) + + def test_shard_shape(self): + c = cat(a=(1, 10), b=(2, 20)) + fp = archive.object_fingerprint("shard", ["a", "b"], c) + self.assertEqual(fp["members"], 2) + self.assertEqual(fp["byte_sum"], 30) + self.assertIn("digest", fp) + + def test_shard_change_signals(self): + base = cat(a=(1, 10), b=(2, 20)) + fp0 = archive.object_fingerprint("shard", ["a", "b"], base) + # size change + self.assertNotEqual(fp0, archive.object_fingerprint( + "shard", ["a", "b"], cat(a=(1, 11), b=(2, 20)))) + # mtime change (whole second) + self.assertNotEqual(fp0, archive.object_fingerprint( + "shard", ["a", "b"], cat(a=(9, 10), b=(2, 20)))) + # membership add + self.assertNotEqual(fp0, archive.object_fingerprint( + "shard", ["a", "b", "c"], cat(a=(1, 10), b=(2, 20), c=(3, 5)))) + # identical -> equal + self.assertEqual(fp0, archive.object_fingerprint("shard", ["a", "b"], base)) + + def test_shard_digest_catches_reshuffle_that_sums_match(self): + # remove b(20) add c(20): same members count + same byte_sum, but the + # digest (over arcnames) must still differ. + fp0 = archive.object_fingerprint("shard", ["a", "b"], cat(a=(1, 10), b=(1, 20))) + fp1 = archive.object_fingerprint("shard", ["a", "c"], cat(a=(1, 10), c=(1, 20))) + self.assertEqual(fp0["members"], fp1["members"]) + self.assertEqual(fp0["byte_sum"], fp1["byte_sum"]) + self.assertNotEqual(fp0["digest"], fp1["digest"]) + + +class TestObjectUnchanged(unittest.TestCase): + def _stored(self, kind, arcnames, catalog): + return {"slot": "s", "kind": kind, "arcnames": arcnames, + "fingerprint": archive.object_fingerprint(kind, arcnames, catalog)} + + def test_baseline_none_is_changed(self): + c = cat(f=(1, BIG)) + cur = {"kind": "solo", "arcnames": ["f"]} + self.assertFalse(archive.object_unchanged(cur, None, c)) + + def test_unchanged_solo_skips(self): + c = cat(f=(1, BIG)) + stored = self._stored("solo", ["f"], c) + cur = {"kind": "solo", "arcnames": ["f"]} + self.assertTrue(archive.object_unchanged(cur, stored, c)) + + def test_solo_mtime_bump_is_changed(self): + stored = self._stored("solo", ["f"], cat(f=(1, BIG))) + cur = {"kind": "solo", "arcnames": ["f"]} + self.assertFalse(archive.object_unchanged(cur, stored, cat(f=(2, BIG)))) + + def test_shard_append_is_changed(self): + # stored shard had {a}; a new small file b was appended to the same shard. + stored = self._stored("shard", ["a"], cat(a=(1, SMALL))) + cur = {"kind": "shard", "arcnames": ["a", "b"]} + self.assertFalse(archive.object_unchanged(cur, stored, cat(a=(1, SMALL), b=(1, SMALL)))) + + def test_shard_unchanged_skips(self): + c = cat(a=(1, SMALL), b=(2, SMALL)) + stored = self._stored("shard", ["a", "b"], c) + cur = {"kind": "shard", "arcnames": ["a", "b"]} + self.assertTrue(archive.object_unchanged(cur, stored, c)) + + +class TestBudgetWarnings(unittest.TestCase): + def test_clean_is_silent(self): + c = cat(a=(1, SMALL), big=(1, BIG)) + objs = archive.route(c, shard_count=2) + self.assertEqual(archive.budget_warnings(objs, c), []) + + def test_max_solo_exceeded(self): + c = cat(**{"f%d" % i: (1, BIG) for i in range(5)}) + objs = archive.route(c, shard_count=1) + w = archive.budget_warnings(objs, c, max_solo=3) + self.assertTrue(any("max_solo" in s for s in w)) + + def test_oversize_shard(self): + c = cat(a=(1, SMALL), b=(1, SMALL)) + objs = archive.route(c, shard_count=1) # both small files into one shard + w = archive.budget_warnings(objs, c, shard_target=1) # 2x = 2 bytes; way over + self.assertTrue(any("shard" in s and "shard_target" in s for s in w)) + + +class TestClampTSmall(unittest.TestCase): + def test_raises_to_floor(self): + self.assertEqual(archive.clamp_t_small(1), archive.T_SMALL_FLOOR) + + def test_keeps_above_floor(self): + self.assertEqual(archive.clamp_t_small(archive.T_SMALL_DEFAULT), + archive.T_SMALL_DEFAULT) + + +class TestManifest(unittest.TestCase): + def test_round_trip_and_missing(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + self.assertIsNone(archive.load_manifest(vault, "abc123")) + m = {"schema_version": 1, "badge": "abc123", "objects": []} + path = archive.write_manifest(vault, "abc123", m) + self.assertTrue(path.endswith("abc123.manifest.json")) + self.assertEqual(archive.load_manifest(vault, "abc123"), m) + + def test_write_leaves_no_tmp(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + archive.write_manifest(vault, "b", {"x": 1}) + self.assertEqual([f for f in os.listdir(vault) if f.endswith(".tmp")], []) + + def test_vault_dodges_reclaim_glob(self): + # manifests live in a _vault child, not directly in log_dir + with tempfile.TemporaryDirectory() as d: + archive.write_manifest(archive.vault_dir_for(d), "b", {"x": 1}) + self.assertEqual([f for f in os.listdir(d)], ["_vault"]) + + +class TestOnlyArcnamesFilter(unittest.TestCase): + def _make_source(self, d): + for rel, content in [("a.txt", b"aaa"), ("b.txt", b"bbbb"), + ("sub/c.txt", b"ccc-c")]: + p = os.path.join(d, rel) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "wb") as fh: + fh.write(content) + + def test_none_archives_all(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._make_source(src) + _, _, checksums, members, _, _ = archive.make_zip_files( + src, ".*", tmp, "proj") + self.assertEqual(set(members), {"a.txt", "b.txt", "sub/c.txt"}) + self.assertEqual(set(checksums), set(members)) + + def test_narrows_to_subset_with_matching_keys(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._make_source(src) + subset = {"a.txt", "sub/c.txt"} + _, _, checksums, members, _, _ = archive.make_zip_files( + src, ".*", tmp, "proj", only_arcnames=subset) + self.assertEqual(set(members), subset) + self.assertEqual(set(checksums), subset) # invariant #2: keys == arcnames + + def test_empty_object_raises(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as tmp: + self._make_source(src) + with self.assertRaises(RuntimeError): + archive.make_zip_files(src, ".*", tmp, "proj", + only_arcnames={"does/not/exist"}) + + +if __name__ == "__main__": + unittest.main()