From 1abc8e0a964219a0abdff87378eb3c780e188555 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Tue, 30 Jun 2026 19:37:38 -0400 Subject: [PATCH] feat(exclusion): identity classifier + git coverage gate primitives (no behavior change) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First slice of docs/EXCLUSION_SPEC.md. Pure, additive helpers — NOTHING in the archive flow calls them yet (behavior-neutral, mirrors PR #8's primitives slice): - compile_exclude_spec / classify_tier — the 3-tier identity classifier (DATA allowlist first + un-overridable, then EXCLUDE, then OPTIONAL first-match, else DATA-default). Rejects bare-extension rules in the subtractive tiers at compile time (the spec's "never classify by extension" rule). DATA precedence + the greenhouse_climate lookahead are enforced. - git_coverage_preconditions / git_file_verdict — the fail-safe git gate. Per-run preconditions (repo+origin present, origin off the protected mount, durable-remote attestation, no git-LFS, fresh fetch, clean subtree) and the per-file proof (L/H/C/P/B/A) returning COVERED|UNTRACKED|DIRTY|STALE|ERROR. ONLY COVERED permits exclusion; every uncertain/error verdict -> back up. - spec_sha256 / _parse_duration helpers. Imports stay inside function bodies (Globus Compute / dill, invariant #1); the compiled spec (regex objects) is local-only, the raw dict crosses the boundary. Tests: tests/test_exclusion_primitives.py (+39) — classifier precedence incl. the greenhouse data-vs-_metadata-twin counterexample, the bare-extension guard, and the full git-gate matrix against real throwaway repos with a local bare remote. Suite 60 -> 99 green. Co-Authored-By: Claude Opus 4.8 --- archive.py | 418 ++++++++++++++++++++++++++++ tests/test_exclusion_primitives.py | 432 +++++++++++++++++++++++++++++ 2 files changed, 850 insertions(+) create mode 100644 tests/test_exclusion_primitives.py diff --git a/archive.py b/archive.py index 22bc1e0..ce3cb2a 100644 --- a/archive.py +++ b/archive.py @@ -456,6 +456,424 @@ def decode_blonde(blonde): } +# --------------------------------------------------------------------------- +# Exclusion primitives (see docs/EXCLUSION_SPEC.md). +# +# A 3-tier identity classifier + a git coverage gate. CARDINAL RULE: it must be +# IMPOSSIBLE to silently drop a data file. A file is dropped ONLY when its tier +# is EXCLUDE (or OPTIONAL explicitly opted-out) AND the git gate returns a +# positive COVERED proof; every other tier and every uncertain/error verdict -> +# back it up (INV-A/B/D). Classification is by ROLE-bearing relpath regex, NEVER +# by extension — a bare `\.json$` / `\.md$` rule is rejected at compile time. +# +# These are pure, additive helpers. NOTHING in the archive flow calls them yet — +# this is behavior-neutral; a follow-up wires should_exclude into the four +# member-set producers + reclaim re-gate (INV-E/G). All imports stay inside the +# function bodies (Globus Compute / dill, invariant #1); the compiled spec holds +# regex objects and is LOCAL-only, while the raw spec dict is what crosses the +# Compute boundary (recompiled on the far side). +# --------------------------------------------------------------------------- + +# Tiers +EXCL_DATA = "DATA" +EXCL_OPTIONAL = "OPTIONAL" +EXCL_EXCLUDE = "EXCLUDE" + +# Per-file git coverage verdicts. Only COVERED permits exclusion. +COV_COVERED = "COVERED" +COV_UNTRACKED = "UNTRACKED" +COV_DIRTY = "DIRTY" +COV_STALE = "STALE" +COV_ERROR = "ERROR" + +# A committed blob that begins with this is a git-LFS pointer, not real bytes. +_LFS_POINTER_PREFIX = b"version https://git-lfs" + +# Generic data-shaped filenames carrying NO metadata role token. A correct +# EXCLUDE/OPTIONAL rule (status/parse-warnings/_metadata/ro-crate/catalog/...) +# matches none of these; a bare-extension rule (`.*\.json$`, `\.md$`) matches +# them — so we reject any exclude/optional rule that does, enforcing the spec's +# "never classify by extension" rule at compile time. +_BARE_EXT_PROBES = ( + "genericdatafile.json", "mydata.json", "results.json", "values.json", + "notes.md", "readme.md", "sample.csv", "picture.png", "table.xlsx", + "a/b/genericdatafile.json", "deep/path/notes.md", +) + + +def spec_sha256(spec): + """ + Stable SHA-256 over the exclusion spec's semantic content (canonical JSON: + sorted keys, no insignificant whitespace). Recorded in the run log and used + by the resume guard — a staged zip built under a different (or absent) spec + must force a re-zip rather than silently mixing exclusion regimes (INV-G). + Returns None for an empty/absent spec. + """ + import hashlib + import json + if not spec: + return None + payload = json.dumps(spec, sort_keys=True, separators=(",", ":"), + ensure_ascii=True) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _parse_duration(text, default_seconds=86400): + """Parse '24h' / '30m' / '90s' / '2d' / '3600' into whole seconds. + Bare integers are seconds. Unknown/empty -> default_seconds.""" + if text is None: + return default_seconds + s = str(text).strip().lower() + if not s: + return default_seconds + units = {"s": 1, "m": 60, "h": 3600, "d": 86400} + if s[-1] in units: + try: + return int(float(s[:-1]) * units[s[-1]]) + except ValueError: + return default_seconds + try: + return int(s) + except ValueError: + return default_seconds + + +def compile_exclude_spec(spec, exclude_optional=()): + """ + Validate and compile an exclusion spec dict (the JSON from `exclude_spec`). + Returns a compiled dict (with re objects) usable by classify_tier, or None + for an empty/absent spec (no exclusion this run — INV-D default-safe). + + Raises ValueError on a malformed spec: a rule missing id/kind/pattern, an + unknown `kind`, an uncompilable regex, or a FORBIDDEN bare-extension rule in + the EXCLUDE/OPTIONAL tiers (the spec's hard "never classify by extension" + rule). `exclude_optional` is the per-asset list of OPTIONAL rule ids to drop; + each OPTIONAL rule's effective disposition is resolved here ('drop' if its id + is opted out, else 'backup'). + """ + import re + if not spec: + return None + if not isinstance(spec, dict): + raise ValueError("exclude_spec must be a JSON object") + + opted_out = set(exclude_optional or ()) + + def _compile_rules(key, tier): + out = [] + for rule in spec.get(key, []) or []: + if not isinstance(rule, dict): + raise ValueError(f"{key}: each rule must be an object, got {rule!r}") + rid = rule.get("id") + kind = rule.get("kind", "regex") + pattern = rule.get("pattern") + note = rule.get("note", "") + if not rid or pattern is None: + raise ValueError(f"{key}: rule missing 'id' or 'pattern': {rule!r}") + if kind != "regex": + raise ValueError(f"{key}: rule {rid!r} has unknown kind {kind!r} " + f"(only 'regex' is supported)") + try: + rx = re.compile(pattern) + except re.error as e: + raise ValueError(f"{key}: rule {rid!r} has an invalid regex " + f"{pattern!r}: {e}") + # Forbid bare-extension rules in the subtractive tiers. A role-bearing + # rule matches none of the generic data probes; a `.*\.json$`-style + # rule matches them and could sweep up real data-as-JSON. + if tier in (EXCL_EXCLUDE, EXCL_OPTIONAL): + hit = next((p for p in _BARE_EXT_PROBES if rx.search(p)), None) + if hit is not None: + raise ValueError( + f"{key}: rule {rid!r} pattern {pattern!r} is a forbidden " + f"bare-extension/too-broad rule — it matches the generic " + f"data file {hit!r}. Classify by a role-bearing name shape " + f"(e.g. '_status', '_metadata', 'ro-crate'), never by " + f"extension (EXCLUSION_SPEC.md §2).") + if tier == EXCL_OPTIONAL: + disposition = "drop" if rid in opted_out else "backup" + out.append((rid, rx, disposition, note)) + else: + out.append((rid, rx, note)) + return out + + coverage = spec.get("coverage", {}) or {} + if not isinstance(coverage, dict): + raise ValueError("exclude_spec 'coverage' must be an object") + + return { + "version": spec.get("version"), + "spec_sha256": spec_sha256(spec), + "exclude_optional": sorted(opted_out), + "data_identity": _compile_rules("data_identity", EXCL_DATA), + "exclude_identity": _compile_rules("exclude_identity", EXCL_EXCLUDE), + "optional_identity": _compile_rules("optional_identity", EXCL_OPTIONAL), + "coverage": coverage, + } + + +def classify_tier(arcname, compiled): + """ + Assign exactly one tier to an arcname (a relpath from source_folder), with + the spec's precedence (EXCLUSION_SPEC.md §1): + + 1. DATA allowlist — highest, UN-OVERRIDABLE. First data_identity match + wins and assignment STOPS (protects greenhouse_climate + data sitting beside *_metadata.json). + 2. EXCLUDE identity -> candidate EXCLUDE. + 3. OPTIONAL identity -> first-listed match wins; disposition from the rule. + 4. Unmatched -> DATA (default). Unknown is data. + + Returns a dict: + {tier, rule_id, note, candidate, data_kind, disposition} + + `candidate` is True iff the file is a candidate for exclusion (tier EXCLUDE, + or OPTIONAL whose effective disposition is 'drop'). A candidate is dropped + ONLY after the git gate returns COVERED — classification never excludes on + its own. `data_kind` is 'DATA-allowlist' (matched a data rule) or + 'DATA-default' (unmatched) for audit logging of contested files. + """ + if compiled is None: + return {"tier": EXCL_DATA, "rule_id": None, "note": "no exclude spec", + "candidate": False, "data_kind": "DATA-default", + "disposition": "backup"} + + for (rid, rx, note) in compiled["data_identity"]: + if rx.search(arcname): + return {"tier": EXCL_DATA, "rule_id": rid, "note": note, + "candidate": False, "data_kind": "DATA-allowlist", + "disposition": "backup"} + + for (rid, rx, note) in compiled["exclude_identity"]: + if rx.search(arcname): + return {"tier": EXCL_EXCLUDE, "rule_id": rid, "note": note, + "candidate": True, "data_kind": None, + "disposition": "drop"} + + for (rid, rx, disposition, note) in compiled["optional_identity"]: + if rx.search(arcname): + return {"tier": EXCL_OPTIONAL, "rule_id": rid, "note": note, + "candidate": (disposition == "drop"), "data_kind": None, + "disposition": disposition} + + return {"tier": EXCL_DATA, "rule_id": None, "note": "unmatched — default data", + "candidate": False, "data_kind": "DATA-default", + "disposition": "backup"} + + +def _git(repo_root, *args): + """Run `git -C ` capturing text output. Returns + (returncode, stdout, stderr). Never raises on a non-zero git exit.""" + import subprocess + try: + r = subprocess.run(["git", "-C", repo_root, *args], + capture_output=True, text=True) + return r.returncode, r.stdout, r.stderr + except (OSError, ValueError) as e: + return 127, "", str(e) + + +def git_coverage_preconditions(coverage, repo_root, asset_subtree, now=None): + """ + Evaluate the once-per-run coverage preconditions (EXCLUSION_SPEC.md §3). If + ANY fails, exclusion is OFF for the whole run (back up everything). Returns + (enabled: bool, reason: str, ctx: dict). ctx (when enabled) carries + repo_head, remote_ref, and fetch_ts for the per-file checks + the log. + + DEFAULT-SAFE: missing repo / origin / attestation, origin under a forbidden + mount, git-LFS in use, a stale-or-failed fetch, or a dirty asset subtree all + return enabled=False. Only an all-clear returns enabled=True. + """ + import os + import time + + cov = coverage or {} + ctx = {} + now = time.time() if now is None else now + + # 1a. Repo present. + rc, out, _ = _git(repo_root, "rev-parse", "--show-toplevel") + if rc != 0 or not out.strip(): + return (False, f"git repo not found at {repo_root}", ctx) + toplevel = os.path.normpath(out.strip()) + ctx["repo_root"] = toplevel + + # 1b. origin set. + rc, out, _ = _git(repo_root, "remote", "get-url", "origin") + if rc != 0 or not out.strip(): + return (False, "git remote 'origin' is not set", ctx) + origin_url = out.strip() + ctx["origin_url"] = origin_url + + # 1c. origin not under a protected/forbidden mount (e.g. /depot). A .git on + # the same doomed filesystem is not redundancy. Only applies to path-like + # origins; ssh/https remotes are off the mount by construction. + forbidden = cov.get("forbid_remote_under", ["/depot/"]) or [] + origin_path = origin_url + if origin_path.startswith("file://"): + origin_path = origin_path[len("file://"):] + if origin_path.startswith("/"): + op = os.path.normpath(origin_path) + for bad in forbidden: + b = os.path.normpath(bad).rstrip("/") + if op == b or op.startswith(b + "/"): + return (False, f"origin {origin_url!r} is under protected mount " + f"{bad!r} — not durable redundancy", ctx) + + # 1d. Durable-remote attestation (operator asserts the remote is itself + # backed up). require_pushed defaults on; turning it off needs the + # attestation AND prints a loud banner at the wiring layer. + require_pushed = cov.get("require_pushed", True) + attested = bool(cov.get("remote_durable_attestation", False)) + if require_pushed and not attested: + return (False, "remote_durable_attestation is not set (require_pushed) — " + "cannot treat the remote as a durable backup", ctx) + ctx["require_pushed"] = require_pushed + ctx["remote_durable_attestation"] = attested + + # 2. No git-LFS (an LFS path is 'tracked' but HEAD stores only a pointer). + gitattributes = os.path.join(toplevel, ".gitattributes") + if os.path.exists(gitattributes): + try: + with open(gitattributes, "r", encoding="utf-8", errors="replace") as fh: + if "filter=lfs" in fh.read(): + return (False, "git-LFS in use (.gitattributes filter=lfs) — " + "HEAD stores pointers, not bytes", ctx) + except OSError as e: + return (False, f"could not read .gitattributes: {e}", ctx) + + # 3. Fresh remote view. Try a fetch; on success the view is current. On + # failure, fall back to FETCH_HEAD age and require it within max_fetch_age. + max_age = _parse_duration(cov.get("max_fetch_age", "24h")) + rc, _, ferr = _git(repo_root, "fetch", "--quiet") + if rc == 0: + ctx["fetch_ts"] = now + else: + fetch_head = os.path.join(toplevel, ".git", "FETCH_HEAD") + try: + age = now - os.path.getmtime(fetch_head) + except OSError: + return (False, f"git fetch failed and no prior fetch found: " + f"{ferr.strip()}", ctx) + if age > max_age: + return (False, f"git fetch failed and last fetch is {int(age)}s old " + f"(> {max_age}s): {ferr.strip()}", ctx) + ctx["fetch_ts"] = now - age + + # remote-tracking ref to check 'pushed' against (origin/HEAD -> origin/main). + rc, out, _ = _git(repo_root, "rev-parse", "--abbrev-ref", "origin/HEAD") + remote_ref = out.strip() if (rc == 0 and out.strip()) else None + if not remote_ref: + # Fall back to the current branch's upstream. + rc, out, _ = _git(repo_root, "rev-parse", "--abbrev-ref", "@{u}") + remote_ref = out.strip() if (rc == 0 and out.strip()) else None + if require_pushed and not remote_ref: + return (False, "no remote-tracking ref (origin/HEAD or @{u}) to verify " + "'pushed' against", ctx) + ctx["remote_ref"] = remote_ref + + rc, out, _ = _git(repo_root, "rev-parse", "HEAD") + ctx["repo_head"] = out.strip() if rc == 0 else None + + # 4. Working tree clean in the asset's subtree. + rc, out, _ = _git(repo_root, "status", "--porcelain", "--", asset_subtree) + if rc != 0: + return (False, f"git status failed for {asset_subtree!r}", ctx) + if out.strip(): + return (False, f"asset subtree {asset_subtree!r} has uncommitted changes", + ctx) + + return (True, "ok", ctx) + + +def git_file_verdict(repo_root, full_path, ctx): + """ + Per-file coverage proof (EXCLUSION_SPEC.md §3, steps L/H/C/P/B/A). Returns + (verdict, evidence) where verdict is one of COV_* and evidence is a dict for + the audit log. ONLY COV_COVERED permits exclusion; every other verdict (and + any error) routes the file to back-up. The tier check (T) is the caller's + responsibility; this runs the git proof for a candidate. + + Absence of proof is proof of "back it up": a missing blob, a dirty/staged + file, a blob not on the freshly-fetched remote, an LFS pointer, a size + mismatch, or a content-rewriting attribute all return a non-COVERED verdict. + """ + import os + + rel = os.path.relpath(os.path.abspath(full_path), repo_root) + ev = {"relpath": rel} + require_pushed = ctx.get("require_pushed", True) + remote_ref = ctx.get("remote_ref") + + # L — tracked in the index. + rc, _, _ = _git(repo_root, "ls-files", "--error-unmatch", "--", rel) + if rc != 0: + return (COV_UNTRACKED, ev) + + # H — blob exists at HEAD (committed, not merely staged). + rc, out, _ = _git(repo_root, "rev-parse", f"HEAD:{rel}") + if rc != 0 or not out.strip(): + return (COV_DIRTY, ev) + blob_oid = out.strip() + ev["blob_oid"] = blob_oid + + # C — on-disk bytes equal the committed blob (no local modification). + rc, out, _ = _git(repo_root, "status", "--porcelain", "--", rel) + if rc != 0 or out.strip(): + return (COV_DIRTY, ev) + + # P — committed blob reachable on the freshly-fetched remote ref (blob-level). + if require_pushed: + if not remote_ref: + return (COV_STALE, ev) + rc, out, _ = _git(repo_root, "rev-parse", f"{remote_ref}:{rel}") + if rc != 0 or out.strip() != blob_oid: + return (COV_STALE, ev) + ev["remote_ref"] = remote_ref + + # B — real bytes, not an LFS pointer: blob size == disk size, and the blob + # does not begin with the LFS pointer signature. + rc, out, _ = _git(repo_root, "cat-file", "-s", blob_oid) + try: + blob_size = int(out.strip()) + except ValueError: + return (COV_ERROR, ev) + try: + disk_size = os.path.getsize(full_path) + except OSError: + return (COV_ERROR, ev) + if blob_size != disk_size: + ev["reason"] = f"blob size {blob_size} != disk size {disk_size}" + return (COV_ERROR, ev) + import subprocess + try: + head = subprocess.run(["git", "-C", repo_root, "cat-file", "-p", blob_oid], + capture_output=True).stdout[:64] + except OSError: + return (COV_ERROR, ev) + if head.startswith(_LFS_POINTER_PREFIX): + ev["reason"] = "blob is a git-LFS pointer" + return (COV_ERROR, ev) + + # A — no content-rewriting attribute (a clean/EOL filter would make working + # bytes differ from the stored blob while --porcelain still reports clean). + rc, out, _ = _git(repo_root, "check-attr", "filter", "diff", "text", "--", rel) + for line in out.splitlines(): + # format: ": : " + parts = line.rsplit(": ", 2) + if len(parts) != 3: + continue + _, attr, value = parts + if attr == "filter" and value not in ("unspecified", "unset"): + ev["reason"] = f"content filter attr {value!r}" + return (COV_ERROR, ev) + if attr == "text" and value in ("set", "auto"): + ev["reason"] = "EOL-normalizing text attribute" + return (COV_ERROR, ev) + + return (COV_COVERED, ev) + + def make_zip_files(source_folder, file_pattern, tmp_dir, project_name, compression="deflate", allow_empty_files=False): """ diff --git a/tests/test_exclusion_primitives.py b/tests/test_exclusion_primitives.py new file mode 100644 index 0000000..139555c --- /dev/null +++ b/tests/test_exclusion_primitives.py @@ -0,0 +1,432 @@ +""" +Tests for the exclusion primitives in archive.py (see docs/EXCLUSION_SPEC.md): +the 3-tier identity classifier (compile_exclude_spec / classify_tier) and the +git coverage gate (git_coverage_preconditions / git_file_verdict). + +These primitives are behavior-neutral — nothing in the archive flow calls them +yet — so these tests exercise them directly. The git-gate tests build real +throwaway git repos (with a local bare "remote") under tmp; everything is local, +no network. + +Run: python3 -m unittest discover -s tests +""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import archive + + +# --------------------------------------------------------------------------- +# A spec mirroring docs/EXCLUSION_SPEC.md §5 (trimmed to what the tests need). +# --------------------------------------------------------------------------- +SPEC = { + "version": 1, + "coverage": { + "oracle": "git", + "repo_root": "/repo", + "require_pushed": True, + "remote_durable_attestation": True, + "forbid_remote_under": ["/depot/"], + "max_fetch_age": "24h", + }, + "data_identity": [ + {"id": "greenhouse_climate", "kind": "regex", + "pattern": r"(^|/)greenhouse_climate_(?!.*_metadata)[^/]*\.json$", + "note": "greenhouse climate DATA"}, + {"id": "tabular", "kind": "regex", "pattern": r"\.(xlsx|xls|csv|tsv)$", + "note": "tabular data"}, + {"id": "imagery", "kind": "regex", + "pattern": r"\.(png|jpg|jpeg|tif|tiff|raw|dng|mp4|mov)$", + "note": "imagery"}, + ], + "exclude_identity": [ + {"id": "status", "kind": "regex", "pattern": r"(^|/)[^/]+_status\.json$", + "note": "status"}, + {"id": "parse_warnings", "kind": "regex", + "pattern": r"(^|/)[^/]+_parse-warnings\.json$", "note": "parse-warnings"}, + {"id": "ro_crate", "kind": "regex", + "pattern": r"(^|/)ro-crate-metadata\.json$", "note": "RO-Crate"}, + {"id": "metadata_sidecar", "kind": "regex", + "pattern": r"(^|/)[^/]+_metadata\.(json|md)$", "note": "metadata sidecar"}, + {"id": "catalog", "kind": "regex", + "pattern": r"(^|/)(catalog|manifest)\.json$", "note": "catalog/manifest"}, + ], + "optional_identity": [ + {"id": "aux_info", "kind": "regex", "pattern": r"(^|/)aux_info/.*\.json$", + "default": "backup", "note": "aux QC"}, + {"id": "omics_provenance", "kind": "regex", + "pattern": r"(^|/)(run_info|meta_info|cmd_info)\.json$", + "default": "backup", "note": "tool provenance"}, + {"id": "fastp", "kind": "regex", "pattern": r"\.fastp\.json$", + "default": "backup", "note": "fastp QC"}, + ], +} + + +class TestCompileSpec(unittest.TestCase): + def test_empty_spec_is_noop(self): + self.assertIsNone(archive.compile_exclude_spec(None)) + self.assertIsNone(archive.compile_exclude_spec({})) + + def test_valid_spec_compiles(self): + c = archive.compile_exclude_spec(SPEC) + self.assertEqual(c["version"], 1) + self.assertEqual(len(c["exclude_identity"]), 5) + self.assertEqual(len(c["data_identity"]), 3) + self.assertTrue(c["spec_sha256"]) + + def test_unknown_kind_rejected(self): + bad = {"exclude_identity": [{"id": "x", "kind": "glob", "pattern": "*"}]} + with self.assertRaises(ValueError): + archive.compile_exclude_spec(bad) + + def test_bad_regex_rejected(self): + bad = {"exclude_identity": [{"id": "x", "kind": "regex", "pattern": "(unclosed"}]} + with self.assertRaises(ValueError): + archive.compile_exclude_spec(bad) + + def test_missing_id_or_pattern_rejected(self): + with self.assertRaises(ValueError): + archive.compile_exclude_spec( + {"exclude_identity": [{"kind": "regex", "pattern": r"_status\.json$"}]}) + with self.assertRaises(ValueError): + archive.compile_exclude_spec( + {"exclude_identity": [{"id": "x", "kind": "regex"}]}) + + def test_bare_extension_exclude_rule_forbidden(self): + # The whole point of the spec: never classify by extension in the + # subtractive tiers. A `.*\.json$` exclude would sweep up data-as-JSON. + for pat in (r"\.json$", r".*\.json$", r"\.md$", r"(^|/).*\.json$"): + with self.assertRaises(ValueError, msg=f"{pat} should be rejected"): + archive.compile_exclude_spec( + {"exclude_identity": [{"id": "x", "kind": "regex", "pattern": pat}]}) + + def test_bare_extension_optional_rule_forbidden(self): + with self.assertRaises(ValueError): + archive.compile_exclude_spec( + {"optional_identity": [{"id": "x", "kind": "regex", + "pattern": r"\.json$", "default": "backup"}]}) + + def test_extension_rule_allowed_in_data_tier(self): + # DATA identity rules ARE legitimately extension-based (tabular/imagery); + # only EXCLUDE/OPTIONAL forbid bare-extension rules. + c = archive.compile_exclude_spec( + {"data_identity": [{"id": "x", "kind": "regex", "pattern": r"\.xlsx$"}]}) + self.assertEqual(len(c["data_identity"]), 1) + + def test_optional_opt_out_resolves_disposition(self): + c = archive.compile_exclude_spec(SPEC, exclude_optional=["fastp"]) + disp = {rid: d for (rid, _rx, d, _n) in c["optional_identity"]} + self.assertEqual(disp["fastp"], "drop") + self.assertEqual(disp["aux_info"], "backup") + self.assertEqual(c["exclude_optional"], ["fastp"]) + + +class TestSpecSha(unittest.TestCase): + def test_stable_and_order_independent(self): + a = {"version": 1, "data_identity": [{"id": "x", "pattern": "a"}]} + b = {"data_identity": [{"pattern": "a", "id": "x"}], "version": 1} + self.assertEqual(archive.spec_sha256(a), archive.spec_sha256(b)) + + def test_changes_with_content(self): + a = archive.spec_sha256({"version": 1}) + b = archive.spec_sha256({"version": 2}) + self.assertNotEqual(a, b) + + def test_none_for_empty(self): + self.assertIsNone(archive.spec_sha256(None)) + self.assertIsNone(archive.spec_sha256({})) + + +class TestParseDuration(unittest.TestCase): + def test_units(self): + self.assertEqual(archive._parse_duration("90s"), 90) + self.assertEqual(archive._parse_duration("30m"), 1800) + self.assertEqual(archive._parse_duration("24h"), 86400) + self.assertEqual(archive._parse_duration("2d"), 172800) + + def test_bare_integer_is_seconds(self): + self.assertEqual(archive._parse_duration("3600"), 3600) + + def test_junk_returns_default(self): + self.assertEqual(archive._parse_duration("", default_seconds=5), 5) + self.assertEqual(archive._parse_duration("nonsense", default_seconds=5), 5) + self.assertEqual(archive._parse_duration(None, default_seconds=5), 5) + + +class TestClassifyTier(unittest.TestCase): + def setUp(self): + self.c = archive.compile_exclude_spec(SPEC, exclude_optional=["fastp"]) + + def test_none_compiled_is_data_default(self): + r = archive.classify_tier("anything.json", None) + self.assertEqual(r["tier"], archive.EXCL_DATA) + self.assertFalse(r["candidate"]) + + def test_greenhouse_climate_data_protected(self): + # The cardinal counterexample: climate DATA arrives as *.json beside + # *_metadata.json. DATA allowlist wins, un-overridable. + r = archive.classify_tier( + "metadata/greenhouse/greenhouse_climate_2026-06.json", self.c) + self.assertEqual(r["tier"], archive.EXCL_DATA) + self.assertEqual(r["rule_id"], "greenhouse_climate") + self.assertEqual(r["data_kind"], "DATA-allowlist") + self.assertFalse(r["candidate"]) + + def test_greenhouse_metadata_twin_excluded(self): + # The lookahead must let the _metadata twin fall through to EXCLUDE. + r = archive.classify_tier( + "metadata/greenhouse/greenhouse_climate_2026_metadata.json", self.c) + self.assertEqual(r["tier"], archive.EXCL_EXCLUDE) + self.assertEqual(r["rule_id"], "metadata_sidecar") + self.assertTrue(r["candidate"]) + + def test_status_requires_prefix(self): + self.assertEqual( + archive.classify_tier("X0A/X0A_status.json", self.c)["tier"], + archive.EXCL_EXCLUDE) + # A bare status.json (no experiment prefix) falls to DATA-default. + bare = archive.classify_tier("X0A/status.json", self.c) + self.assertEqual(bare["tier"], archive.EXCL_DATA) + self.assertEqual(bare["data_kind"], "DATA-default") + + def test_exclude_identities(self): + for arc, rid in [ + ("X0A/X0A_parse-warnings.json", "parse_warnings"), + ("d/ro-crate-metadata.json", "ro_crate"), + ("X0A/X0A_metadata.json", "metadata_sidecar"), + ("X0A/X0A_metadata.md", "metadata_sidecar"), + ("X0A/catalog.json", "catalog"), + ("X0A/manifest.json", "catalog"), + ]: + r = archive.classify_tier(arc, self.c) + self.assertEqual(r["tier"], archive.EXCL_EXCLUDE, arc) + self.assertEqual(r["rule_id"], rid, arc) + self.assertTrue(r["candidate"], arc) + + def test_unmatched_is_data(self): + r = archive.classify_tier("X0A/raw/IMG_001.dat", self.c) + self.assertEqual(r["tier"], archive.EXCL_DATA) + self.assertEqual(r["data_kind"], "DATA-default") + + def test_data_extension_rules(self): + for arc in ["x/table.xlsx", "x/scan.tiff", "x/photo.PNG".lower()]: + self.assertEqual(archive.classify_tier(arc, self.c)["tier"], + archive.EXCL_DATA, arc) + + def test_optional_first_match_wins(self): + # aux_info/run_info.json matches BOTH aux_info and omics_provenance; + # the first-listed (aux_info) must win. + r = archive.classify_tier("salmon/aux_info/run_info.json", self.c) + self.assertEqual(r["tier"], archive.EXCL_OPTIONAL) + self.assertEqual(r["rule_id"], "aux_info") + + def test_optional_default_backup_not_candidate(self): + r = archive.classify_tier("salmon/run_info.json", self.c) + self.assertEqual(r["tier"], archive.EXCL_OPTIONAL) + self.assertFalse(r["candidate"]) # default backup, not opted out + + def test_optional_opt_out_becomes_candidate(self): + r = archive.classify_tier("reads/sample.fastp.json", self.c) + self.assertEqual(r["tier"], archive.EXCL_OPTIONAL) + self.assertTrue(r["candidate"]) # fastp opted out in setUp + + +# --------------------------------------------------------------------------- +# Git coverage gate — real throwaway repos with a local bare "remote". +# --------------------------------------------------------------------------- +def _git(cwd, *args, check=True): + r = subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True) + if check and r.returncode != 0: + raise RuntimeError(f"git {' '.join(args)} failed: {r.stderr}") + return r + + +def _have_git(): + try: + subprocess.run(["git", "--version"], capture_output=True) + return True + except OSError: + return False + + +@unittest.skipUnless(_have_git(), "git not available") +class TestGitGate(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.remote = os.path.join(self.tmp, "remote.git") + self.work = os.path.join(self.tmp, "work") + subprocess.run(["git", "init", "--bare", "-b", "main", self.remote], + capture_output=True, check=True) + subprocess.run(["git", "init", "-b", "main", self.work], + capture_output=True, check=True) + _git(self.work, "config", "user.email", "t@t") + _git(self.work, "config", "user.name", "t") + _git(self.work, "config", "core.autocrlf", "false") + _git(self.work, "config", "commit.gpgsign", "false") + _git(self.work, "remote", "add", "origin", self.remote) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def _write(self, rel, content="hello\n"): + p = os.path.join(self.work, rel) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "w") as fh: + fh.write(content) + return p + + def _commit(self, msg="c"): + _git(self.work, "add", "-A") + _git(self.work, "commit", "-m", msg) + + def _push(self): + _git(self.work, "push", "-u", "origin", "main") + + # ---- per-file verdicts (ctx built directly to isolate from preconditions) -- + def _ctx(self, require_pushed=True, remote_ref="origin/main"): + return {"require_pushed": require_pushed, "remote_ref": remote_ref} + + def test_covered(self): + p = self._write("X0A/X0A_status.json") + self._commit() + self._push() + verdict, ev = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_COVERED) + self.assertIn("blob_oid", ev) + self.assertEqual(ev["remote_ref"], "origin/main") + + def test_untracked(self): + p = self._write("X0A/new.json") + # never added + verdict, _ = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_UNTRACKED) + + def test_staged_only_is_dirty(self): + p = self._write("X0A/X0A_status.json") + _git(self.work, "add", "-A") # staged, not committed + verdict, _ = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_DIRTY) + + def test_modified_after_commit_is_dirty(self): + p = self._write("X0A/X0A_status.json") + self._commit() + self._push() + with open(p, "w") as fh: + fh.write("changed\n") # dirty the working copy + verdict, _ = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_DIRTY) + + def test_committed_not_pushed_is_stale(self): + self._write("base.txt") + self._commit() + self._push() # origin/main now exists + p = self._write("X0A/X0A_status.json") + self._commit() # committed locally, NOT pushed + verdict, _ = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_STALE) + + def test_lfs_pointer_blob_is_error(self): + # A committed file whose blob content is an LFS pointer: clean, pushed, + # but B must catch that the bytes are a pointer, not real content. + p = self._write("X0A/big.bin", + "version https://git-lfs.github.com/spec/v1\n" + "oid sha256:deadbeef\nsize 12345\n") + self._commit() + self._push() + verdict, ev = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_ERROR) + self.assertIn("LFS", ev.get("reason", "")) + + def test_eol_filter_attribute_is_error(self): + # A content-rewriting `text` attribute would make working bytes differ + # from the stored blob while --porcelain still reports clean. + self._write(".gitattributes", "*.dat text\n") + p = self._write("X0A/d.dat", "data\n") + self._commit() + self._push() + verdict, ev = archive.git_file_verdict(self.work, p, self._ctx()) + self.assertEqual(verdict, archive.COV_ERROR) + + def test_require_pushed_false_allows_unpushed(self): + # With require_pushed off, a committed+clean local file is COVERED even + # without a remote (the dangerous override; gated by attestation upstream). + p = self._write("X0A/X0A_status.json") + self._commit() + verdict, _ = archive.git_file_verdict( + self.work, p, self._ctx(require_pushed=False, remote_ref=None)) + self.assertEqual(verdict, archive.COV_COVERED) + + # ---- run-level preconditions ------------------------------------------ + def _cov(self, **over): + cov = {"forbid_remote_under": ["/depot/"], + "remote_durable_attestation": True, "require_pushed": True, + "max_fetch_age": "24h"} + cov.update(over) + return cov + + def test_preconditions_happy_path(self): + self._write("X0A/X0A_status.json") + self._commit() + self._push() + ok, reason, ctx = archive.git_coverage_preconditions( + self._cov(), self.work, self.work) + self.assertTrue(ok, reason) + self.assertTrue(ctx["remote_ref"]) + self.assertTrue(ctx["repo_head"]) + + def test_preconditions_no_repo(self): + ok, reason, _ = archive.git_coverage_preconditions( + self._cov(), os.path.join(self.tmp, "nope"), os.path.join(self.tmp, "nope")) + self.assertFalse(ok) + + def test_preconditions_no_attestation(self): + self._write("a.txt") + self._commit() + self._push() + ok, reason, _ = archive.git_coverage_preconditions( + self._cov(remote_durable_attestation=False), self.work, self.work) + self.assertFalse(ok) + self.assertIn("attestation", reason) + + def test_preconditions_origin_under_forbidden_mount(self): + self._write("a.txt") + self._commit() + self._push() + # forbid the temp dir that actually contains the bare remote → origin is + # "under" it → must disable (stands in for the real /depot rule). + ok, reason, _ = archive.git_coverage_preconditions( + self._cov(forbid_remote_under=[self.tmp]), self.work, self.work) + self.assertFalse(ok) + self.assertIn("protected mount", reason) + + def test_preconditions_lfs_disables(self): + self._write(".gitattributes", "*.bin filter=lfs diff=lfs merge=lfs -text\n") + self._write("a.txt") + self._commit() + self._push() + ok, reason, _ = archive.git_coverage_preconditions( + self._cov(), self.work, self.work) + self.assertFalse(ok) + self.assertIn("LFS", reason) + + def test_preconditions_dirty_subtree_disables(self): + self._write("X0A/a.txt") + self._commit() + self._push() + self._write("X0A/dirty.txt", "uncommitted\n") # untracked change in subtree + ok, reason, _ = archive.git_coverage_preconditions( + self._cov(), self.work, os.path.join(self.work, "X0A")) + self.assertFalse(ok) + self.assertIn("uncommitted", reason) + + +if __name__ == "__main__": + unittest.main()