From 8cae58cdf10568cdef361dbb40667539453c84fd Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 18:36:28 +0000 Subject: [PATCH] test: fix V4 dirty mock, reject in-checkout output, typed blob errors - test_dfg_re20: the dirty-checkout test now supplies all three git responses (HEAD:src, HEAD, status) and asserts the call order, so it reaches the clean-checkout rejection instead of StopIteration. - run_dfg_re20: --output inside the source checkout is refused before provenance, plan writing or any CUDA work (receipts would dirty the recorded source); tested with the checkout root and nested paths. - source_provenance.historical_files: validates all paths first and reports an unreadable blob as HistoryUnavailable, not a raw CalledProcessError; tested with a failing git show. No physics thresholds changed. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_dfg_re20.py | 23 +++++++++++++++++-- .../unit/test_validation_source_provenance.py | 13 +++++++++++ tests/validation/run_dfg_re20.py | 8 +++++++ tests/validation/source_provenance.py | 9 +++++--- 4 files changed, 48 insertions(+), 5 deletions(-) diff --git a/tests/unit/test_dfg_re20.py b/tests/unit/test_dfg_re20.py index fb85577..6fe60b4 100644 --- a/tests/unit/test_dfg_re20.py +++ b/tests/unit/test_dfg_re20.py @@ -104,10 +104,29 @@ class Re20Provenance(unittest.TestCase): def test_dirty_checkout_is_refused(self): from unittest.mock import patch - with patch("source_provenance.subprocess.check_output", - side_effect=["f" * 40 + "\n", " M src/x.py\n"]): + # Three git calls in order: HEAD:src, HEAD, status --porcelain. + with patch("subprocess.check_output", + side_effect=["e" * 40 + "\n", "f" * 40 + "\n", " M src/x.py\n"]) as git: with self.assertRaisesRegex(RuntimeError, "clean checkout"): v.provenance(rr.DEFAULT_SOURCE) + self.assertEqual(git.call_count, 3) + self.assertEqual([c.args[0][-1] for c in git.call_args_list[:2]], ["HEAD:src", "HEAD"]) + self.assertIn("--porcelain", git.call_args_list[2].args[0]) + + def test_output_inside_checkout_is_refused_before_any_work(self): + import tempfile + from pathlib import Path + from unittest.mock import patch + source = rr.DEFAULT_SOURCE.resolve() + for output in (source, source / "re20-out", source / "src" / "x"): + argv = ["run_dfg_re20.py", "--plan", "--source", str(source), "--output", str(output)] + with patch("sys.argv", argv), \ + patch.object(v, "provenance", side_effect=AssertionError("work started")): + with self.assertRaisesRegex(SystemExit, "outside the source checkout"): + v.main() + self.assertFalse((source / "re20-out").exists()) + with tempfile.TemporaryDirectory() as d: + v.reject_output_inside(Path(d) / "out", source) # outside: accepted if __name__ == "__main__": diff --git a/tests/unit/test_validation_source_provenance.py b/tests/unit/test_validation_source_provenance.py index 0b9d42b..df26033 100644 --- a/tests/unit/test_validation_source_provenance.py +++ b/tests/unit/test_validation_source_provenance.py @@ -98,3 +98,16 @@ def test_historical_check_rejects_tampering_and_reports_missing_history(tmp_path verify_manifest(ROOT, ORIGIN, tampered) # one flipped digest is caught with pytest.raises(ValueError, match="not part of"): historical_sha256(ROOT, ORIGIN, ["src/CelerisLab/does_not_exist.py"]) + + +def test_historical_files_reports_unreadable_blob_as_history_unavailable(tmp_path): + """A blob missing from the clone is HistoryUnavailable, not a raw CalledProcessError.""" + import source_provenance + from unittest.mock import patch + + missing = subprocess.CompletedProcess(["git", "show"], 128, b"", b"fatal: missing blob") + with patch.object(source_provenance, "historical_sha256", return_value={"src/a.py": "0" * 64}), \ + patch.object(source_provenance.subprocess, "run", return_value=missing): + with pytest.raises(HistoryUnavailable, match="blob not in this clone"): + source_provenance.historical_files(ROOT, ORIGIN, ["src/a.py"], tmp_path) + assert not (tmp_path / "src/a.py").exists() diff --git a/tests/validation/run_dfg_re20.py b/tests/validation/run_dfg_re20.py index 9e6099d..4b26c7b 100644 --- a/tests/validation/run_dfg_re20.py +++ b/tests/validation/run_dfg_re20.py @@ -225,6 +225,13 @@ def provenance(source): return record +def reject_output_inside(output, source): + """Receipts inside the checkout would make it dirty and change the recorded source.""" + output = Path(output).resolve() + if output == source or source in output.parents: + raise SystemExit(f"--output must be outside the source checkout {source}") + + def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--case", choices=["refined", "uniform", "both"], default="refined") @@ -235,6 +242,7 @@ def main(): parser.add_argument("--max-wall-seconds", type=float) a = parser.parse_args() a.source = a.source.resolve() + reject_output_inside(a.output, a.source) record = dict(plan(a.collision), schema=RESULT_SCHEMA, provenance=provenance(a.source)) a.output.mkdir(parents=True, exist_ok=False) (a.output / "plan.json").write_text(json.dumps(record, indent=2, allow_nan=False)) diff --git a/tests/validation/source_provenance.py b/tests/validation/source_provenance.py index 71f415e..d3dd07f 100644 --- a/tests/validation/source_provenance.py +++ b/tests/validation/source_provenance.py @@ -66,12 +66,15 @@ def historical_tree(repo, commit, prefix="src"): def historical_files(repo, commit, paths, dest): """Write each path as stored in `commit` under `dest` (HistoryUnavailable if missing).""" from pathlib import Path - historical_sha256(repo, commit, []) + historical_sha256(repo, commit, paths) # HistoryUnavailable / ValueError before writing for path in paths: + shown = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"], + capture_output=True) + if shown.returncode: + raise HistoryUnavailable(f"cannot read {commit}:{path} (blob not in this clone)") target = Path(dest) / path target.parent.mkdir(parents=True, exist_ok=True) - target.write_bytes(subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"], - check=True, capture_output=True).stdout) + target.write_bytes(shown.stdout) def history_unavailable(testcase, error):