From 537b61e8d997f927943470c5d379dfd5ab175f7a Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 13 Oct 2026 23:17:15 +0200 Subject: [PATCH] fix(hunt): diff producer records where a push removed lines; worktree checkout is proven complete Two instrument findings from the first access-control feature table (952 rows, 74 positive rows): 1. A fix that adds a guard is, replayed base-to-head, a push that removes it. The diff producer only kept hunks with added lines at head, so 38 of 74 positive rows had no hunk in their file at all and 69 of 74 were "not on a changed line", while 60% of ordinary rows were. The diff annotation separated the labels by construction and was blind to exactly the access-control shape. The producer (version 2) now records per file the head lines after which the push removed code; distance and on_changed_line count them, the model's hunks input renders them as removed_after_line, and diff.deletion_count is an engineered feature. 2. 150 rows (16%, every positive of one repository) read empty files from the worktree although git holds the blobs; the build ran while another process checked out from the same clone. _full_checkout now proves the checkout: git status must be clean, a dirty tree is restored once, then InstrumentFailure checkout_incomplete. The table script cross-checks every unread row against the blob size and refuses to screen a partly unread tree, and records the diff-scoped recall ceiling (positive rows a label-blind enumeration would reach). Co-Authored-By: Claude Fable 5.1 --- .../2026-10-13-access-control-screen/table.py | 149 ++++++++++++++++++ src/openultrasast/hunt/features.py | 2 + src/openultrasast/hunt/producers.py | 38 ++++- src/openultrasast/hunt/program.py | 14 +- src/openultrasast/hunt/runner.py | 8 +- src/openultrasast/hunt/table.py | 2 +- tests/test_hunt_features.py | 3 +- tests/test_hunt_producers.py | 58 +++++-- tests/test_hunt_program.py | 2 +- tests/test_hunt_runner.py | 19 +++ 10 files changed, 273 insertions(+), 22 deletions(-) create mode 100644 benchmarks/measurements/2026-10-13-access-control-screen/table.py diff --git a/benchmarks/measurements/2026-10-13-access-control-screen/table.py b/benchmarks/measurements/2026-10-13-access-control-screen/table.py new file mode 100644 index 0000000..68536b9 --- /dev/null +++ b/benchmarks/measurements/2026-10-13-access-control-screen/table.py @@ -0,0 +1,149 @@ +"""Feature table and separation screen over a frozen training pool, no model calls. + +For every change of the pool's slice, check out its head, enumerate the candidate rows (touched functions at head +plus the labelled fix site) and compute the four producers. Rows are labelled 1 at an accepted fix site, 0 on an +ordinary push and null elsewhere on a positive. The screen then scores every engineered feature alone against the +label (ROC-AUC) and reports the strongest feature per annotation. Repositories are the clusters: with few of them +the screen is a direction, not a result, and the output says how many there are. + + PYTHONPATH=. .venv/bin/python benchmarks/measurements/2026-10-13-access-control-screen/table.py \\ + benchmarks/valid/train/pool-train.toml --repos ~/ousast-results/valid/train/repos_hist.json --out +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +import tempfile +import time +import tomllib +from collections import Counter +from pathlib import Path + +from openultrasast.hunt import features +from openultrasast.hunt.producers import compute_rows +from openultrasast.hunt.runner import _full_checkout as full_checkout +from openultrasast.hunt.table import FeatureTable +from openultrasast.stats import separation + + +def _git(root: Path, *args: str) -> str: + return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True, text=True).stdout + + +def changes_of(pool: Path, slice_id: int) -> list[dict]: + data = tomllib.loads(pool.read_text()) + rows = next(value for value in data.values() if isinstance(value, list) and value and "changes" in value[0]) + out = [] + for row in rows: + if int(row.get("slice", 1)) != slice_id: + continue + for change in row["changes"]: + change = dict(change) + change.setdefault("repo", row.get("repository") or row.get("repo") or row["id"]) + change.setdefault("id", change["case_id"]) + out.append(change) + return out + + +def rows_for(change: dict, clone: Path) -> tuple[list[dict], dict]: + started = time.monotonic() + with tempfile.TemporaryDirectory(prefix="ousast-table-") as temporary: + root = Path(temporary) / "head" + _git(clone, "worktree", "add", "--detach", str(root), change["head"]) + try: + if _git(root, "rev-parse", "HEAD").strip() != change["head"]: + raise RuntimeError("head_mismatch") + full_checkout(root) + rows = compute_rows(root, change) + finally: + _git(clone, "worktree", "remove", "--force", str(root)) + return rows, {"seconds": round(time.monotonic() - started, 2), "rows": len(rows)} + + +def unread_rows(rows: list[dict], repos: dict[str, str]) -> int: + """Rows whose file the source producer could not read although git holds a non-empty blob at head. + + An empty or missing file in the worktree is a checkout failure, not a feature; the first build read 150 such + rows as feature-free (every positive of one repository among them) and nothing in the screen said so.""" + count = 0 + checked: dict[tuple[str, str, str], bool] = {} + for row in rows: + if row.get("provenance.source.reason") not in ("empty_source", "FileNotFoundError"): + continue + key = (row["repo"], row["head"], row["path"]) + if key not in checked: + clone = Path(repos[row["repo"]]).expanduser() + size = subprocess.run( + ["git", "-C", str(clone), "cat-file", "-s", f"{row['head']}:{row['path']}"], capture_output=True, text=True + ) + checked[key] = size.returncode == 0 and int(size.stdout.strip() or 0) > 0 + count += checked[key] + return count + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("pool", type=Path) + parser.add_argument("--repos", type=Path, required=True, help="JSON map repo -> local clone with history") + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--slice", type=int, default=1) + args = parser.parse_args(argv) + repos = json.loads(args.repos.read_text()) + args.out.mkdir(parents=True, exist_ok=True) + rows, log = [], [] + for change in changes_of(args.pool, args.slice): + clone = Path(repos[change["repo"]]).expanduser() + try: + produced, info = rows_for(change, clone) + except Exception as error: # noqa: BLE001 (type only; a message can carry a path) + info = {"failure": type(error).__name__} + produced = [] + rows.extend(produced) + log.append({"id": change["id"], "repo": change["repo"], "kind": change["kind"], **info}) + print(f"{change['id'][:16]} {change['kind']} {info}", flush=True) + table = FeatureTable(rows) + table.write_jsonl(args.out / "table.jsonl") + # The instrument gate: a producer that failed on every row saw an unread tree, not a feature-free one. + provenance = {name: sum(bool(row.get(f"provenance.{name}.ok")) for row in rows) for name in ("diff", "source", "semantic", "taxonomy")} + print("producer rows ok:", provenance, "of", len(rows), flush=True) + if rows and provenance["source"] == 0: + print("the source producer read nothing: refusing to screen an unread tree", file=sys.stderr) + return 2 + unread = unread_rows(rows, repos) + print("rows unread although git holds the file:", unread, flush=True) + if unread: + print("the checkout was incomplete for some rows: refusing to screen a partly unread tree", file=sys.stderr) + return 2 + (args.out / "log.json").write_text(json.dumps(log, indent=1)) + frame = features.feature_frame(table) + labelled = frame[frame["label"].isin([0, 1])] + summary = { + "producer_rows_ok": provenance, + "table_digest": table.digest(), + "rows": len(frame), + "labelled": {"positive": int((labelled["label"] == 1).sum()), "negative": int((labelled["label"] == 0).sum())}, + "repositories": int(labelled["repo"].nunique()), + "changes": int(labelled["change_id"].nunique()), + "positive_repositories": int(labelled[labelled["label"] == 1]["repo"].nunique()), + "source_failures": dict(Counter(row.get("provenance.source.reason") for row in rows if not row.get("provenance.source.ok"))), + "unread_rows": unread, + # The recall ceiling of a diff-scoped candidate set: positive rows a label-blind enumeration would reach. + "positive_rows_on_changed_line": int(labelled[(labelled["label"] == 1) & (labelled["diff.on_changed_line"] == 1)].shape[0]), + "positive_rows_with_deletions": int(labelled[(labelled["label"] == 1) & (labelled["diff.deletion_count"].fillna(0) > 0)].shape[0]), + } + columns = [name for name in features.ENGINEERED if name in frame.columns] + per_feature = separation(labelled, "label", columns) + per_feature["annotation"] = [features.ENGINEERED[name][1] for name in per_feature["column"]] + per_feature.to_csv(args.out / "separation.csv", index=False) + features.screen(labelled).to_csv(args.out / "screen.csv", index=False) + (args.out / "summary.json").write_text(json.dumps(summary, indent=1)) + print(json.dumps(summary)) + print(per_feature.sort_values("auc", key=lambda s: (s - 0.5).abs(), ascending=False).to_string(index=False)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/openultrasast/hunt/features.py b/src/openultrasast/hunt/features.py index 5e577d5..665d11c 100644 --- a/src/openultrasast/hunt/features.py +++ b/src/openultrasast/hunt/features.py @@ -20,6 +20,7 @@ ENGINEERED: dict[str, tuple[str, str]] = { "diff.on_changed_line": ("feature.diff.on_changed_line", "hunks"), "diff.hunk_count": ("feature.diff.hunks", "hunks"), "diff.hunk_lines": ("feature.diff.hunks", "hunks"), + "diff.deletion_count": ("feature.diff.deletions", "hunks"), "source.span_lines": ("feature.source.span", "function_text"), "source.is_module": ("feature.source.function", "change_summary"), "source.caller_count": ("feature.source.callers", "callers"), @@ -57,6 +58,7 @@ def _engineer(row: Mapping[str, Any]) -> dict[str, Any]: else int(bool(row["feature.diff.on_changed_line"])), "diff.hunk_count": _count(hunks), "diff.hunk_lines": None if hunks is None else sum(end - start + 1 for start, end in hunks), + "diff.deletion_count": _count(row.get("feature.diff.deletions")), "source.span_lines": None if not span else span[1] - span[0] + 1, "source.is_module": None if function is None else int(function == ""), "source.caller_count": _count(callers), diff --git a/src/openultrasast/hunt/producers.py b/src/openultrasast/hunt/producers.py index 4b9a8af..63c5afc 100644 --- a/src/openultrasast/hunt/producers.py +++ b/src/openultrasast/hunt/producers.py @@ -96,6 +96,19 @@ class Context: @cached_property def hunks(self) -> dict[str, list[list[int]]]: + """Line ranges per file that the push changed or added, at head.""" + return self._patch[0] + + @cached_property + def deletions(self) -> dict[str, list[int]]: + """Per file, the head lines after which the push removed lines (0 when it removed the top of the file). + + A fix that adds a guard is a push that removes it when replayed base-to-head, and that push has no added + line at head: without this column the diff features are blind to exactly that shape.""" + return self._patch[1] + + @cached_property + def _patch(self) -> tuple[dict[str, list[list[int]]], dict[str, list[int]]]: self.verify_head() # Exclude private files before git opens the patch; only hunk coordinates become features. patch = _git( @@ -112,6 +125,7 @@ class Context: ) self.diff_bytes = len(patch.encode("utf-8")) result: dict[str, list[list[int]]] = {} + removed: dict[str, list[int]] = {} path = None for text in patch.splitlines(): if text.startswith("+++ "): @@ -122,7 +136,9 @@ class Context: start, count = int(match[1]), int(match[2] or "1") if count: result.setdefault(path, []).append([start, start + count - 1]) - return result + else: + removed.setdefault(path, []).append(start) + return result, removed def ir(self, path: str) -> FileIR: if path not in self.irs: @@ -143,7 +159,7 @@ class Producer: compute: Compute -def _producer(name: str, compute: Compute) -> Producer: +def _producer(name: str, compute: Compute, version: str = "1") -> Producer: columns = tuple(f"feature.{name}.{column}" for column in FEATURE_NAMES[name]) provenance = tuple(f"provenance.{name}.{column}" for column in ("ok", "reason", "version", "bytes_read")) @@ -167,19 +183,22 @@ def _producer(name: str, compute: Compute) -> Producer: { f"provenance.{name}.ok": reason is None, f"provenance.{name}.reason": reason, - f"provenance.{name}.version": "1", + f"provenance.{name}.version": version, f"provenance.{name}.bytes_read": bytes_read, } ) return result - return Producer(name, "1", columns + provenance, guarded) + return Producer(name, version, columns + provenance, guarded) def _diff(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]: hunks = ctx.hunks.get(path, []) - distance = min((max(start - line, 0, line - end) for start, end in hunks), default=None) - return {"hunks": hunks, "on_changed_line": distance == 0, "distance": distance} + deletions = ctx.deletions.get(path, []) + # A removal sits between head lines `after` and `after + 1`; both lines count as touched. + ranges = hunks + [[max(after, 1), max(after, 1) + 1] for after in deletions] + distance = min((max(start - line, 0, line - end) for start, end in ranges), default=None) + return {"hunks": hunks, "deletions": deletions, "on_changed_line": distance == 0, "distance": distance} def _source(root: Path, change: Change, path: str, line: int, ctx: Context) -> dict[str, object]: @@ -279,7 +298,12 @@ def _taxonomy(root: Path, change: Change, path: str, line: int, ctx: Context) -> return {"families": [{"id": f.id, "definition": " ".join(f.description.split())} for f in load_families().families]} -PRODUCERS = (_producer("diff", _diff), _producer("source", _source), _producer("semantic", _semantic), _producer("taxonomy", _taxonomy)) +PRODUCERS = ( + _producer("diff", _diff, version="2"), + _producer("source", _source), + _producer("semantic", _semantic), + _producer("taxonomy", _taxonomy), +) def _candidates(root: Path, change: Change, ctx: Context) -> list[dict[str, Any]]: diff --git a/src/openultrasast/hunt/program.py b/src/openultrasast/hunt/program.py index 5cad5de..822efcf 100644 --- a/src/openultrasast/hunt/program.py +++ b/src/openultrasast/hunt/program.py @@ -31,7 +31,11 @@ BASELINE_INSTRUCTION = ( # Input name -> (feature column, renderer). Order is the order the model sees them in; masked columns are absent. INPUTS: tuple[tuple[str, str, str], ...] = ( ("change_summary", "", "The files and functions the push touched."), - ("hunks", "feature.diff.hunks", "Changed line ranges per file at head, as [start, end] pairs."), + ( + "hunks", + "feature.diff.hunks", + "Per file at head: changed line ranges as [start, end] pairs, and the lines after which the push removed code.", + ), ("function_text", "feature.source.span", "The enclosing function at head, numbered lines."), ("callers", "feature.source.callers", "Where the enclosing function is called from: path and line."), ("entry_points", "feature.source.entry_points", "In-repository entry points that reach the file."), @@ -82,7 +86,13 @@ def render_rows( touched = sorted({(row["path"], str(row.get("feature.source.function") or "")) for row in rows}) out["change_summary"] = "\n".join(f"{path}::{function}" for path, function in touched) or "(nothing readable)" if "hunks" in wanted: - hunks = {row["path"]: row.get("feature.diff.hunks") for row in rows if row.get("feature.diff.hunks")} + hunks: dict[str, dict[str, list]] = {} + for row in rows: + if row.get("feature.diff.hunks") or row.get("feature.diff.deletions"): + hunks[row["path"]] = { + "changed": row.get("feature.diff.hunks") or [], + "removed_after_line": row.get("feature.diff.deletions") or [], + } out["hunks"] = json.dumps(hunks, sort_keys=True) if hunks else "(no hunks)" if "function_text" in wanted: blocks = [] diff --git a/src/openultrasast/hunt/runner.py b/src/openultrasast/hunt/runner.py index ce7d2e0..dbffd55 100644 --- a/src/openultrasast/hunt/runner.py +++ b/src/openultrasast/hunt/runner.py @@ -169,9 +169,15 @@ def _full_checkout(root: Path) -> None: try: sparse = _git(root, "config", "--get", "core.sparseCheckout").strip() except subprocess.CalledProcessError: - return + sparse = "false" if sparse == "true": _git(root, "sparse-checkout", "disable") + # The proof: a fresh worktree has nothing modified. Truncated files (seen once while two processes checked out + # from the same clone) show up here as modified; restore them once, then refuse rather than read an unread tree. + if _git(root, "status", "--porcelain", "--untracked-files=no").strip(): + _git(root, "checkout", "--", ".") + if _git(root, "status", "--porcelain", "--untracked-files=no").strip(): + raise InstrumentFailure("checkout_incomplete") def _function_at(root: Path, finding: Mapping[str, Any]) -> str | None: diff --git a/src/openultrasast/hunt/table.py b/src/openultrasast/hunt/table.py index 5ba4480..360b78b 100644 --- a/src/openultrasast/hunt/table.py +++ b/src/openultrasast/hunt/table.py @@ -13,7 +13,7 @@ SCHEMA_VERSION = 1 KEY_COLUMNS = ("change_id", "repo", "head", "path", "line", "kind", "fold") LABEL_COLUMNS = ("label", "family", "fix_span", "label_ref", "verification") FEATURE_NAMES = { - "diff": ("hunks", "on_changed_line", "distance"), + "diff": ("hunks", "deletions", "on_changed_line", "distance"), "source": ("function", "span", "callers", "entry_points", "language"), "semantic": ("roles", "def_use", "obligations"), "taxonomy": ("families",), diff --git a/tests/test_hunt_features.py b/tests/test_hunt_features.py index cd463a3..24809cb 100644 --- a/tests/test_hunt_features.py +++ b/tests/test_hunt_features.py @@ -20,6 +20,7 @@ ROW = { "label": 1, "family": "injection", "feature.diff.hunks": [[3, 3], [10, 12]], + "feature.diff.deletions": [9], "feature.diff.on_changed_line": False, "feature.diff.distance": 2, "feature.source.function": "second", @@ -56,7 +57,7 @@ def test_feature_frame_engineers_counts_distances_and_flags(): frame = features.feature_frame([ROW, absent]) first = frame.iloc[0] assert first["diff.distance"] == 2 and first["diff.on_changed_line"] == 0 - assert first["diff.hunk_count"] == 2 and first["diff.hunk_lines"] == 4 + assert first["diff.hunk_count"] == 2 and first["diff.hunk_lines"] == 4 and first["diff.deletion_count"] == 1 assert first["source.span_lines"] == 3 and first["source.is_module"] == 0 assert first["source.caller_count"] == 3 and first["source.caller_files"] == 2 and first["source.entry_point_count"] == 1 assert (first["semantic.sources"], first["semantic.sinks"], first["semantic.sanitizers"]) == (1, 2, 1) diff --git a/tests/test_hunt_producers.py b/tests/test_hunt_producers.py index 611db3a..55cd523 100644 --- a/tests/test_hunt_producers.py +++ b/tests/test_hunt_producers.py @@ -20,8 +20,7 @@ def repository(tmp_path): git("init", "-q") git("config", "user.name", "Fixture") git("config", "user.email", "fixture@example.invalid") - initial = ('def first():\n value = input()\n return value\n\n' - 'def second():\n value = first()\n return value\n') + initial = "def first():\n value = input()\n return value\n\ndef second():\n value = first()\n return value\n" (tmp_path / "app.py").write_text(initial) (tmp_path / "notes.txt").write_text("GHSA-xxxx-xxxx-xxxx CVE-2024-0001\n") (tmp_path / "unknown.zzz").write_text("opaque\n") @@ -33,10 +32,22 @@ def repository(tmp_path): git("add", "app.py", "unknown.zzz") git("commit", "-qm", "fix XSS CVE-2024-0001") head = git("rev-parse", "HEAD") - data = {"id": "change", "repo": "fixture/repo", "base": base, "head": head, "kind": "vulnerability", "fold": 1, - "path": "app.py", "function": "second", "function_lines": [5, 7], "family": "injection", "case_id": "verified", - "advisory": "GHSA-xxxx-xxxx-xxxx", "commit_message": "fix XSS CVE-2024-0001", - "files": [{"path": "app.py", "head_lines": [[3, 3]]}]} + data = { + "id": "change", + "repo": "fixture/repo", + "base": base, + "head": head, + "kind": "vulnerability", + "fold": 1, + "path": "app.py", + "function": "second", + "function_lines": [5, 7], + "family": "injection", + "case_id": "verified", + "advisory": "GHSA-xxxx-xxxx-xxxx", + "commit_message": "fix XSS CVE-2024-0001", + "files": [{"path": "app.py", "head_lines": [[3, 3]]}], + } return tmp_path, data @@ -55,10 +66,11 @@ def test_every_column_candidates_and_cached_work(repository, monkeypatch): rows = compute_rows(root, change) assert len(calls) == 1 table = FeatureTable(rows) - assert len(table.feature_columns()) == 12 + assert len(table.feature_columns()) == 13 assert [r["label"] for r in rows] == [None, 1] first, second = rows assert first["feature.diff.hunks"] == [[3, 3]] + assert first["feature.diff.deletions"] == [] and first["provenance.diff.version"] == "2" assert first["feature.diff.distance"] == 2 assert not first["feature.diff.on_changed_line"] assert first["feature.source.function"] == "first" @@ -149,10 +161,38 @@ def test_head_mismatch_is_not_a_clean_measurement(repository): def test_obligation_result_uses_dominance_not_absence_of_findings(repository, monkeypatch): root, change = repository - text = ('from flask import request\n@app.route("/item")\ndef first():\n' - ' return User.query.get(request.args["id"])\n') + text = 'from flask import request\n@app.route("/item")\ndef first():\n return User.query.get(request.args["id"])\n' (root / "app.py").write_text(text) monkeypatch.setattr(module, "_git", lambda *args: change["head"]) row = compute_rows(root, {**change, "kind": "ordinary", "files": [{"path": "app.py", "head_lines": [[4, 4]]}]})[0] assert row["feature.semantic.obligations"] assert row["feature.semantic.obligations"][0]["discharged"] is False + + +def test_a_removed_guard_anchors_the_function_it_left(tmp_path): + """Replayed base-to-head, a fix that added a check is a push that removes it: no added line, one deletion anchor.""" + + def git(*args): + return subprocess.run(["git", "-C", str(tmp_path), *args], check=True, capture_output=True, text=True).stdout.strip() + + git("init", "-q") + git("config", "user.name", "Fixture") + git("config", "user.email", "fixture@example.invalid") + check = " if not user.is_admin:\n raise PermissionError\n" + guarded = "def first():\n return 1\n\n\ndef admin(user):\n" + check + " return delete_everything()\n" + (tmp_path / "app.py").write_text(guarded) + git("add", "app.py") + git("commit", "-qm", "fix: require admin") + base = git("rev-parse", "HEAD") + (tmp_path / "app.py").write_text(guarded.replace(check, "")) + git("add", "app.py") + git("commit", "-qm", "unguarded") + head = git("rev-parse", "HEAD") + change = {"id": "c", "case_id": "c-1", "repo": "fixture/repo", "base": base, "head": head, "kind": "ordinary", "fold": 1, + "files": [{"path": "app.py", "head_lines": [[1, 1], [5, 6]]}]} # fmt: skip + rows = compute_rows(tmp_path, change) + row = next(r for r in rows if r["feature.source.function"] == "admin") + assert row["feature.diff.hunks"] == [] and row["feature.diff.deletions"] == [5] + assert row["feature.diff.on_changed_line"] and row["feature.diff.distance"] == 0 + first = next(r for r in rows if r["feature.source.function"] == "first") + assert first["feature.diff.distance"] == 4 and not first["feature.diff.on_changed_line"] diff --git a/tests/test_hunt_program.py b/tests/test_hunt_program.py index d3b8389..416dc65 100644 --- a/tests/test_hunt_program.py +++ b/tests/test_hunt_program.py @@ -43,7 +43,7 @@ def test_signature_drops_masked_inputs_and_keeps_the_order(): def test_render_rows_numbers_the_function_and_omits_masked_inputs(): inputs = hp.render_rows(CHANGE, ROWS, TEXT) assert inputs["change_summary"] == "app.py::second" - assert json.loads(inputs["hunks"]) == {"app.py": [[3, 3]]} + assert json.loads(inputs["hunks"]) == {"app.py": {"changed": [[3, 3]], "removed_after_line": []}} assert " 5 def second():" in inputs["function_text"] and " 6 " in inputs["function_text"] assert "input()" not in inputs["function_text"] # only the span is shown assert inputs["families"].startswith("injection: ") diff --git a/tests/test_hunt_runner.py b/tests/test_hunt_runner.py index 8dbd343..b9e9fd0 100644 --- a/tests/test_hunt_runner.py +++ b/tests/test_hunt_runner.py @@ -108,3 +108,22 @@ def test_instrument_failure_is_recorded_with_a_type_only(repository, tmp_path): assert record["status"] == "instrument_failure" assert record["instrument"]["failure"] == "CalledProcessError" assert record["push"]["admission"]["defects"] == [] + + +def test_full_checkout_restores_a_truncated_file_and_refuses_an_unread_tree(repository, tmp_path, monkeypatch): + root, _manifest, changes = repository + worktree = tmp_path / "head" + subprocess.run( + ["git", "-C", str(root), "worktree", "add", "--detach", str(worktree), changes[0]["head"]], check=True, capture_output=True + ) + (worktree / "app.py").write_bytes(b"") + runner._full_checkout(worktree) + assert (worktree / "app.py").read_bytes() # the truncated file was restored, not read as empty + real = runner._git + + def still_dirty(target, *args): + return " M app.py\n" if args[:2] == ("status", "--porcelain") else real(target, *args) + + monkeypatch.setattr(runner, "_git", still_dirty) + with pytest.raises(runner.InstrumentFailure, match="checkout_incomplete"): + runner._full_checkout(worktree)