diff --git a/tests/unit/test_jet_circle_cv.py b/tests/unit/test_jet_circle_cv.py index 57b78aa..a28331c 100644 --- a/tests/unit/test_jet_circle_cv.py +++ b/tests/unit/test_jet_circle_cv.py @@ -209,15 +209,15 @@ class Host(unittest.TestCase): done = v.finish_receipt(path, payload, recorded, lambda: dict(recorded)) self.assertEqual(json.loads(path.read_text()), done) self.assertEqual((done["run_status"], done["source_recheck"]), ("COMPLETE", "UNCHANGED")) - for recheck, after in ((lambda: dict(recorded, checkout_dirty=True), True), - (lambda: v.verify(Path(d)), None)): + for recheck, state in ((lambda: dict(recorded, checkout_dirty=True), "CHANGED"), + (lambda: v.verify(Path(d)), "RECHECK_FAILED")): path.write_text(json.dumps(payload)) with self.assertRaisesRegex(RuntimeError, "Source changed during the run"): v.finish_receipt(path, payload, recorded, recheck) bad = json.loads(path.read_text()) self.assertEqual((bad["run_status"], bad["source_recheck"]), - ("INVALID_SOURCE_CHANGED", "CHANGED")) - self.assertEqual(bad["source_after"] is None, after is None) + ("INVALID_SOURCE_CHANGED", state)) + self.assertEqual(bad["source_after"] is None, state == "RECHECK_FAILED") self.assertEqual(bad["windows"], [1, 2]) self.assertFalse(Path(str(path) + ".tmp").exists()) @@ -231,5 +231,16 @@ class Host(unittest.TestCase): self.assertIn("finish_receipt(", inspect.getsource(v.main)) + def test_output_inside_checkout_is_refused_before_source_or_cuda(self): + from unittest import mock + import run_jet_circle_cv as v + argv = ["run_jet_circle_cv.py", "--repo", str(REPO), "--output", str(REPO / "v1-out")] + with mock.patch("sys.argv", argv), mock.patch.dict("os.environ", CUDA_VISIBLE_DEVICES="0"), \ + mock.patch.object(v, "source_identity", side_effect=AssertionError("work started")): + with self.assertRaisesRegex(SystemExit, "outside the source checkout"): + v.main() + self.assertFalse((REPO / "v1-out").exists()) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_jet_cv_measurements.py b/tests/unit/test_jet_cv_measurements.py index 6ac59fb..39f7872 100644 --- a/tests/unit/test_jet_cv_measurements.py +++ b/tests/unit/test_jet_cv_measurements.py @@ -151,6 +151,8 @@ class Measurements(unittest.TestCase): (path / "samples_s2.json").write_text( json.dumps(dict(case=c, sample_file="samples_s2.npz")) ) + (path / f"re{c['re_d']}_s2.json").write_text(json.dumps( + dict(case=c, run_status="COMPLETE", source_recheck="UNCHANGED"))) np.savez( path / "samples_s2.npz", t_reference=t, @@ -163,6 +165,7 @@ class Measurements(unittest.TestCase): velocity_error=np.zeros(n), ) result = analyze(path) + self.assertEqual(result["receipt_gate"], "COMPLETE_SOURCE_UNCHANGED") self.assertEqual(len(result["windows"]), 32) np.testing.assert_allclose( result["windows"][0]["normalized_cv_residual"], 0, atol=1e-14 @@ -178,5 +181,36 @@ class Measurements(unittest.TestCase): np.testing.assert_allclose(integrate(t, y), [112, 8], atol=1e-15) + def test_receipt_gate_refuses_incomplete_invalid_and_unmarked_legacy(self): + import json + from analyze_jet_cv_continuation import receipt_gate + + c = case(2, steps=3200) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) + receipt = path / f"re{c['re_d']}_s2.json" + with self.assertRaisesRegex(ValueError, "No sibling run receipt"): + receipt_gate(path, c) + for fields, message in ( + (dict(run_status="INCOMPLETE"), "INCOMPLETE"), + (dict(run_status="INVALID_SOURCE_CHANGED", source_recheck="CHANGED"), "INVALID"), + (dict(run_status="COMPLETE", source_recheck="RECHECK_FAILED"), "RECHECK_FAILED"), + (dict(run_status="COMPLETE"), "source_recheck=None"), + (dict(), "Legacy run receipt"), + ): + receipt.write_text(json.dumps(dict(case=c, **fields))) + with self.assertRaisesRegex(ValueError, message): + receipt_gate(path, c) + receipt.write_text(json.dumps(dict(case=dict(c, scale=1), run_status="COMPLETE", + source_recheck="UNCHANGED"))) + with self.assertRaisesRegex(ValueError, "different case"): + receipt_gate(path, c) + receipt.write_text(json.dumps(dict(case=c))) + self.assertEqual(receipt_gate(path, c, legacy=True), "LEGACY_NO_FINAL_SOURCE_RECHECK") + receipt.write_text(json.dumps(dict(case=c, run_status="INCOMPLETE"))) + with self.assertRaisesRegex(ValueError, "INCOMPLETE"): + receipt_gate(path, c, legacy=True) # legacy opt-in never admits new statuses + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_validation_source_provenance.py b/tests/unit/test_validation_source_provenance.py index 54c3f4a..57d37bc 100644 --- a/tests/unit/test_validation_source_provenance.py +++ b/tests/unit/test_validation_source_provenance.py @@ -126,5 +126,14 @@ def test_final_source_check_unchanged_changed_and_failed_recheck(): raise RuntimeError("Source hash mismatch: src/x.py") failed = final_source_check(recorded, fails) - assert (failed["source_recheck"], failed["source_after"]) == ("CHANGED", None) + assert (failed["source_recheck"], failed["source_after"]) == ("RECHECK_FAILED", None) assert "Source hash mismatch" in failed["source_recheck_error"] + + +def test_reject_output_inside_checkout(tmp_path): + from source_provenance import reject_output_inside + + for output in (ROOT, ROOT / "out", ROOT / "src" / "x.json"): + with pytest.raises(SystemExit, match="outside the source checkout"): + reject_output_inside(output, ROOT) + reject_output_inside(tmp_path / "out", ROOT) diff --git a/tests/validation/analyze_jet_cv_continuation.py b/tests/validation/analyze_jet_cv_continuation.py index 032e038..03c466a 100644 --- a/tests/validation/analyze_jet_cv_continuation.py +++ b/tests/validation/analyze_jet_cv_continuation.py @@ -14,11 +14,35 @@ def integrate(t, y): ) -def analyze(directory): +def receipt_gate(directory, c, legacy=False): + """Side files are used only if the sibling run receipt finished with an unchanged source. + + Receipts from runners before the final source recheck have no run_status; they are + refused unless legacy=True (--legacy-receipts), and the result is labelled as such. + """ + path = Path(directory) / f"re{c['re_d']}_s{c['scale']}.json" + if not path.exists(): + raise ValueError(f"No sibling run receipt {path.name}") + receipt = json.loads(path.read_text()) + if receipt.get("case") != c: + raise ValueError("Sibling run receipt describes a different case") + if "run_status" not in receipt: + if not legacy: + raise ValueError("Legacy run receipt without run_status/final source recheck; " + "pass --legacy-receipts to analyze it as such") + return "LEGACY_NO_FINAL_SOURCE_RECHECK" + if receipt["run_status"] != "COMPLETE" or receipt.get("source_recheck") != "UNCHANGED": + raise ValueError(f"Run receipt is {receipt['run_status']} " + f"(source_recheck={receipt.get('source_recheck')})") + return "COMPLETE_SOURCE_UNCHANGED" + + +def analyze(directory, legacy=False): meta_path = next(Path(directory).glob("samples_s*.json")) meta = json.loads(meta_path.read_text()) c = meta["case"] s = c["scale"] + gate = receipt_gate(directory, c, legacy) sample_path = meta_path.parent / meta["sample_file"] checksum_verified = False if "sample_sha256" in meta: @@ -97,6 +121,7 @@ def analyze(directory): ) return dict( scale=s, + receipt_gate=gate, complete_schedule=bool(complete_schedule), sample_checksum_verified=checksum_verified, windows=records, @@ -115,8 +140,10 @@ def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument("directories", nargs="+", type=Path) p.add_argument("--output", type=Path, required=True) + p.add_argument("--legacy-receipts", action="store_true", + help="Accept receipts written before the final source recheck (labelled)") a = p.parse_args() - results = [analyze(d) for d in a.directories] + results = [analyze(d, a.legacy_receipts) for d in a.directories] payload = dict( status="TRANSIENT_DIAGNOSTIC_COMPARISON", results=results, diff --git a/tests/validation/run_jet_circle_cv.py b/tests/validation/run_jet_circle_cv.py index 6cc8bf3..52ae462 100644 --- a/tests/validation/run_jet_circle_cv.py +++ b/tests/validation/run_jet_circle_cv.py @@ -592,9 +592,9 @@ def run_case(c, ctx, output, provenance, continuation=False): physics_validation_pass=False, precision="FP32 evolution, FP64 host diagnostics", ) - (output / f"re{c['re_d']}_s{c['scale']}.json").write_text( - json.dumps(payload, indent=2) + "\n" - ) + receipt = output / f"re{c['re_d']}_s{c['scale']}.json" + Path(str(receipt) + ".tmp").write_text(json.dumps(payload, indent=2) + "\n") + Path(str(receipt) + ".tmp").replace(receipt) print( json.dumps( dict( @@ -670,6 +670,8 @@ def main(): ): p.error("Root must select one verified idle GPU via CUDA_VISIBLE_DEVICES") repo = a.repo.resolve() + from source_provenance import reject_output_inside + reject_output_inside(a.output, repo) provenance = source_identity(repo, a.current_checkout) sys.path.insert(0, str(repo / "src")) import CelerisLab diff --git a/tests/validation/source_provenance.py b/tests/validation/source_provenance.py index 807e8ae..c2b0b05 100644 --- a/tests/validation/source_provenance.py +++ b/tests/validation/source_provenance.py @@ -18,20 +18,28 @@ def checkout_provenance(repo): def final_source_check(recorded, recheck): """Re-derive the source identity after a run and compare it with the recorded one. - Returns receipt fields: source_recheck "UNCHANGED", or "CHANGED" with source_after - (None if the recheck itself failed) and the recheck error. Callers mark a receipt - complete only for "UNCHANGED"; a changed source invalidates the run. + Returns receipt fields: source_recheck "UNCHANGED"; "CHANGED" with source_after; or + "RECHECK_FAILED" (source_after None) with the error. Callers mark a receipt complete + only for "UNCHANGED"; anything else invalidates the run. """ try: after = recheck() - except Exception as error: # a failed recheck cannot confirm the source - return dict(source_recheck="CHANGED", source_after=None, + except Exception as error: # a failed recheck cannot confirm the source: invalid too + return dict(source_recheck="RECHECK_FAILED", source_after=None, source_recheck_error=f"{type(error).__name__}: {error}") if after == recorded: return dict(source_recheck="UNCHANGED") return dict(source_recheck="CHANGED", source_after=after, source_recheck_error=None) +def reject_output_inside(output, repo): + """Refuse run outputs inside the checkout before any work: they would dirty it.""" + from pathlib import Path + output, repo = Path(output).resolve(), Path(repo).resolve() + if output == repo or repo in output.parents: + raise SystemExit(f"--output must be outside the source checkout {repo}") + + class HistoryUnavailable(RuntimeError): """The clone cannot show the recorded commit (no git, shallow clone, installed copy).""" diff --git a/tests/validation/taylor_green/assess_projected_pairs.py b/tests/validation/taylor_green/assess_projected_pairs.py index dc3db00..b170a94 100644 --- a/tests/validation/taylor_green/assess_projected_pairs.py +++ b/tests/validation/taylor_green/assess_projected_pairs.py @@ -73,6 +73,27 @@ def valid_builds(record, n, depth): return True +# run_case.py (sha256) of the runners before the final source recheck (git: 9805383, +# 2cf9853 = the a135252 qualification runner, db3d098); they wrote schema 1 receipts. +LEGACY_RUN_CASE_SHA256 = { + "9805383": "fbceab8c1aaaf357195a4fb9e6a699a195326e480e274692a277ce3df54ee6d9", + "2cf9853": "0789d2e1ffa581078bd11e6251b660b2c268363171fda0c6b26ac41205f1de3a", + "db3d098": "618c14ba94584aa9c0b714e0e2064a746d2b882973a13424f09dd7e54c68f7f1", +} + + +def rechecked(r, current): + """Schema 2 receipts need source_recheck UNCHANGED; a missing recheck is accepted only + for frozen-mode schema 1 receipts written by a verified legacy runner.""" + if "invalid" in r: + return False + if r.get("schema") == 2: + return r.get("source_recheck") == "UNCHANGED" + harness = r.get("source", {}).get("harness_sha256") or {} + return (not current and r.get("schema") == 1 and "source_recheck" not in r + and harness.get("run_case.py") in LEGACY_RUN_CASE_SHA256.values()) + + def assess(records, expected_revision=None, current=False): """current=True assesses current-checkout receipts (run_case --current-checkout) with the same gates; it never accepts them as the frozen a135252 qualification, and the @@ -144,10 +165,7 @@ def assess(records, expected_revision=None, current=False): checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION and "source_mode" not in r and r.get("reference_harness") == REFERENCE_HARNESS) - # Final source recheck (run_case); receipts from before it existed carry no field - # and are accepted only in the frozen mode. - recheck = r.get("source_recheck") if current else r.get("source_recheck", "UNCHANGED") - checks[label + " source_recheck"] = recheck == "UNCHANGED" and "invalid" not in r + checks[label + " source_recheck"] = rechecked(r, current) checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION runtime = r.get("runtime_source", {}) package = Path(runtime.get("checkout", "")) / "src/CelerisLab" diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index ea903e1..2d644eb 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -182,6 +182,8 @@ def main(): p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required") repo = args.repo.resolve() sys.path[:0] = [str(repo / "src"), str(repo)] + from tests.validation.source_provenance import reject_output_inside + reject_output_inside(args.output, repo) source = verify_source(repo, args.expected_revision, current=args.current_checkout) runtime = runtime_source(repo) if args.dry_run: @@ -201,7 +203,7 @@ def main(): masks = [k == ACTIVE for k in h.kinds] main_steps, early_steps = sample_steps(args.n, args.early_pressure) report = dict( - schema=1, + schema=2, # 2: final source recheck (source_recheck) is recorded source=source, numerical_revision=None if args.current_checkout else NUMERICAL_REVISION, **({"source_mode": "current_checkout"} if args.current_checkout else {}), diff --git a/tests/validation/taylor_green/test_projected_pairs.py b/tests/validation/taylor_green/test_projected_pairs.py index 60dd4ff..170b0d9 100644 --- a/tests/validation/taylor_green/test_projected_pairs.py +++ b/tests/validation/taylor_green/test_projected_pairs.py @@ -18,7 +18,7 @@ import numpy as np import run_case from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION, - SOURCE_SHA256, assess) + LEGACY_RUN_CASE_SHA256, SOURCE_SHA256, assess) from oracle import LIMITS, TIMES, U, rectangles REPO = Path(__file__).resolve().parents[3] @@ -31,7 +31,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0 def source(): return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256, - harness_sha256={"run_case.py": "a" * 64}, + harness_sha256={"run_case.py": LEGACY_RUN_CASE_SHA256["2cf9853"]}, build_capture_sha256=run_case.BUILD_CAPTURE_SHA256) @@ -53,7 +53,7 @@ def records(early=False): h = build_hierarchy(cfg, refinements) main_steps, early_steps = run_case.sample_steps(n, early) out.append(dict( - re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)), + schema=1, re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)), viscosity=cfg.viscosity, tau_by_level=[0.5+3*c.viscosity for c in h.configs], refinements=refinements, numerics=NUMERICS.copy(), initialization=run_case.INITIALIZATION, source=source(), numerical_revision=NUMERICAL_REVISION, @@ -322,7 +322,7 @@ class ProjectedPairsTests(unittest.TestCase): source_mode="current_checkout", source_capture=run_case.CURRENT_CAPTURE) r["numerical_revision"], r["source_mode"] = None, "current_checkout" - r["source_recheck"] = "UNCHANGED" + r["schema"], r["source_recheck"] = 2, "UNCHANGED" ok = assess(current, MOCK_REVISION, current=True) self.assertTrue(ok["checks"]["clean_current_checkout_source"]) self.assertTrue(all(v for k, v in ok["checks"].items() if k.endswith(" reference"))) @@ -335,22 +335,61 @@ class ProjectedPairsTests(unittest.TestCase): self.assertTrue(assess(frozen, MOCK_REVISION)["checks"]["clean_a135252_numerical_source"]) def test_assessor_requires_unchanged_final_source(self): - frozen = records() + frozen = records() # schema 1 receipts from the verified legacy runner 2cf9853 label = lambda r: f"{r['n']}/{r['depth']}/{r['explosion']} source_recheck" self.assertTrue(all(assess(frozen, MOCK_REVISION)["checks"][label(r)] for r in frozen)) - for mutate in (dict(source_recheck="CHANGED"), dict(invalid="SOURCE_CHANGED_DURING_RUN")): - bad = deepcopy(frozen) - bad[0].update(mutate) - out = assess(bad, MOCK_REVISION) - self.assertFalse(out["checks"][label(bad[0])]) - self.assertFalse(out["passed"]) + new = deepcopy(frozen) + for r in new: + r.update(schema=2, source_recheck="UNCHANGED") + self.assertTrue(all(assess(new, MOCK_REVISION)["checks"][label(r)] for r in new)) + unknown_runner = dict(frozen[0]["source"], harness_sha256={"run_case.py": "a" * 64}) + for case, mutate in ( + ("schema 1, unverified runner", dict(source=unknown_runner)), + ("schema 2 without recheck", dict(schema=2)), + ("schema 2 changed", dict(schema=2, source_recheck="CHANGED")), + ("schema 2 recheck failed", dict(schema=2, source_recheck="RECHECK_FAILED")), + ("schema 1 with a recheck field", dict(source_recheck="UNCHANGED")), + ("unknown schema", dict(schema=3, source_recheck="UNCHANGED")), + ("no schema", dict(schema=None)), + ("invalid", dict(schema=2, source_recheck="UNCHANGED", + invalid="SOURCE_CHANGED_DURING_RUN")), + ): + with self.subTest(case): + bad = deepcopy(frozen) + bad[0].update(mutate) + out = assess(bad, MOCK_REVISION) + self.assertFalse(out["checks"][label(bad[0])]) + self.assertFalse(out["passed"]) current = deepcopy(frozen) for r in current: r["source"] = dict(r["source"], source_mode="current_checkout") r["numerical_revision"], r["source_mode"] = None, "current_checkout" - # Current-checkout receipts are new: a missing recheck is not accepted. + # Legacy acceptance is frozen-mode only. self.assertFalse(assess(current, MOCK_REVISION, current=True)["checks"][label(current[0])]) + def test_legacy_runner_hashes_are_the_recorded_commits(self): + sys.path.insert(0, str(REPO)) + from tests.validation.source_provenance import (HistoryUnavailable, historical_sha256, + history_unavailable) + try: + for commit, digest in LEGACY_RUN_CASE_SHA256.items(): + path = "tests/validation/taylor_green/run_case.py" + self.assertEqual(historical_sha256(REPO, commit, [path])[path], digest) + except HistoryUnavailable as error: + history_unavailable(self, error) + current = hashlib.sha256((REPO / "tests/validation/taylor_green/run_case.py").read_bytes()) + self.assertNotIn(current.hexdigest(), LEGACY_RUN_CASE_SHA256.values()) + + def test_output_inside_checkout_is_refused_before_source_check(self): + argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128", "--depth", "1", + "--output", str(REPO / "tg-out.json"), "--expected-revision", MOCK_REVISION, + "--dry-run"] + with patch.object(sys, "argv", argv), \ + patch.object(run_case, "verify_source", side_effect=AssertionError("work started")): + with self.assertRaisesRegex(SystemExit, "outside the source checkout"): + run_case.main() + self.assertFalse((REPO / "tg-out.json").exists()) + def test_explicit_opt_in_selector_and_dry_run_without_cuda(self): with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) / "case.json" @@ -483,13 +522,15 @@ class ProjectedPairsTests(unittest.TestCase): [call(REPO, MOCK_REVISION, current=False)] * (2 if rechecked else 1)) if fail in ("changed", "recheck_error"): # Full trajectory, but invalid: never a complete-looking receipt. - self.assertEqual(receipt["source_recheck"], "CHANGED") + self.assertEqual(receipt["source_recheck"], + "CHANGED" if fail == "changed" else "RECHECK_FAILED") self.assertEqual(receipt["invalid"], "SOURCE_CHANGED_DURING_RUN") self.assertFalse(receipt["completed"]) self.assertNotIn("screening_pass", receipt) self.assertEqual(receipt["source_after"] is None, fail == "recheck_error") continue self.assertEqual(receipt.get("source_recheck"), None if fail else "UNCHANGED") + self.assertEqual(receipt["schema"], 2) self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else []) self.assertEqual(receipt["source"], source()) self.assertEqual(receipt["runtime_identity"], RUNTIME)