From f02355a2b7e2d8c60753e0da16b3f71e17cde152 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 18:27:13 +0000 Subject: [PATCH] test: LES a135252 guard tests use the pinned tree, not today's src The a135252 manifest is historical qualification: it pins the committed corrected tree at a135252. The guard is correct; the four tests used the current working tree as the 'pinned' fixture, so 87c3a31's jets.py change broke them. Fixtures are now materialized from a135252 with git; the corrected acceptance, baseline rejection, four tamper rejections and the reversed-header baseline control are unchanged. A new test records that the guard refuses today's src where it differs from the pin. Missing history skips with the reason, or fails when CELERIS_REQUIRE_HISTORY=1 (release gating) via history_unavailable(). Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_les_flow_runner.py | 66 ++++++++++++++++++--------- tests/validation/source_provenance.py | 20 ++++++++ 2 files changed, 64 insertions(+), 22 deletions(-) diff --git a/tests/unit/test_les_flow_runner.py b/tests/unit/test_les_flow_runner.py index 5063f60..b3fa3c4 100644 --- a/tests/unit/test_les_flow_runner.py +++ b/tests/unit/test_les_flow_runner.py @@ -13,8 +13,15 @@ from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation")) from run_uniform_les_flow import source_guard, decode_macros, HERE, HEADER +from source_provenance import (HistoryUnavailable, historical_files, historical_sha256, + historical_tree, history_unavailable) ROOT = HERE.parents[1] +A135252 = "a135252ae78b47b57d7e77d19001700b3bedac8b" + + +def a135252_pins(): + return json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text()) class GuardTests(unittest.TestCase): @@ -41,20 +48,35 @@ class GuardTests(unittest.TestCase): with self.assertRaisesRegex(RuntimeError, "velocity/profile"): decode_macros(invalid, w, cx, cy) - def test_current_corrected_pins(self): - self.assertEqual( - source_guard(ROOT, "corrected", "a135252")["base_commit"], - "a135252ae78b47b57d7e77d19001700b3bedac8b", - ) + def materialize(self, root): + """The pinned a135252 tree, read with git (the guard's own qualification).""" + try: + historical_files(ROOT, A135252, a135252_pins()["sources"], root) + except HistoryUnavailable as error: + history_unavailable(self, error) - def test_corrected_and_tamper_guard(self): - pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text()) + def test_a135252_pins_qualify_their_own_commit(self): with TemporaryDirectory() as directory: root = Path(directory) - for name in pins["sources"]: - dest = root / name - dest.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(ROOT / name, dest) + self.materialize(root) + with patch("run_uniform_les_flow.subprocess.check_output", + return_value="isolated-fixture\n"): + self.assertEqual(source_guard(root, "corrected", "a135252")["base_commit"], A135252) + + def test_current_tree_is_not_the_a135252_qualification(self): + """Explicit record: the guard refuses today's src wherever it differs from the pin.""" + pins = a135252_pins()["sources"] + changed = sorted(n for n, sha in pins.items() if (ROOT / n).exists() + and hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != sha) + if not changed: + self.skipTest("current src equals the a135252 pin") + with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): + source_guard(ROOT, "corrected", "a135252") + + def test_corrected_and_tamper_guard(self): + with TemporaryDirectory() as directory: + root = Path(directory) + self.materialize(root) with patch( "run_uniform_les_flow.subprocess.check_output", return_value="isolated-fixture\n", @@ -85,25 +107,25 @@ class GuardTests(unittest.TestCase): with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(ROOT, "corrected") # Omission still uses the historical profile. - def test_current_manifest_pins_actual_corrected_tree(self): - pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text()) - tracked = subprocess.check_output(["git", "-C", str(ROOT), "ls-files", "src"], text=True) - self.assertEqual(set(pins["sources"]), set(tracked.splitlines())) - for name, expected in pins["sources"].items(): - self.assertEqual(hashlib.sha256((ROOT / name).read_bytes()).hexdigest(), expected) + def test_manifest_pins_the_a135252_corrected_tree(self): + pins = a135252_pins() + self.assertEqual(pins["base_commit"], A135252) self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"]) self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"]) with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(ROOT, "baseline", "a135252") + try: + names = historical_tree(ROOT, A135252) + actual = historical_sha256(ROOT, A135252, pins["sources"]) + except HistoryUnavailable as error: + history_unavailable(self, error) + self.assertEqual(set(pins["sources"]), set(names)) + self.assertEqual(actual, pins["sources"]) def test_named_baseline_accepts_only_reversed_header(self): - pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text()) with TemporaryDirectory() as directory: root = Path(directory) - for name in pins["sources"]: - dest = root / name - dest.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(ROOT / name, dest) + self.materialize(root) subprocess.run(["git", "apply", "--reverse", str(HERE / "les/data/smagorinsky_normalization.patch")], cwd=root, check=True) with patch("run_uniform_les_flow.subprocess.check_output", return_value="baseline-fixture\n"): source_guard(root, "baseline", "a135252") diff --git a/tests/validation/source_provenance.py b/tests/validation/source_provenance.py index 1fcb965..cee5d51 100644 --- a/tests/validation/source_provenance.py +++ b/tests/validation/source_provenance.py @@ -48,3 +48,23 @@ def historical_tree(repo, commit, prefix="src"): historical_sha256(repo, commit, []) return subprocess.check_output(["git", "-C", str(repo), "ls-tree", "-r", "--name-only", commit, "--", prefix], text=True).splitlines() + + +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, []) + for path in paths: + 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) + + +def history_unavailable(testcase, error): + """Skip with the reason, or fail when CELERIS_REQUIRE_HISTORY=1 (release gating).""" + import os + message = f"historical integrity NOT checked: {error}" + if os.environ.get("CELERIS_REQUIRE_HISTORY") == "1": + raise AssertionError(message) + testcase.skipTest(message)