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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
34b30c434f
commit
8cae58cdf1
@@ -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__":
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user