"""Host-only entrypoint/source-guard checks; never imports PyCUDA.""" import hashlib import json from pathlib import Path import shutil import subprocess import sys from tempfile import TemporaryDirectory import unittest import numpy as np 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): def test_decoded_density_and_velocity_guards(self): w = np.array([4 / 9] + [1 / 9] * 4 + [1 / 36] * 4) cx = np.array([0, 1, -1, 0, 0, 1, -1, -1, 1]) cy = np.array([0, 0, 0, 1, -1, 1, -1, 1, -1]) data = np.zeros((9, 4, 2)) profile, mass, vy, spread = decode_macros(data, w, cx, cy) np.testing.assert_allclose(profile, 0, atol=1e-16) self.assertAlmostEqual(mass, 1) for scale in (1, 2): invalid = np.broadcast_to(-scale * w[:, None, None], data.shape).copy() with self.assertRaisesRegex(RuntimeError, "density"): decode_macros(invalid, w, cx, cy) invalid = data.copy() invalid[0, 1, 0] = np.nan with self.assertRaisesRegex(RuntimeError, "populations"): decode_macros(invalid, w, cx, cy) # Finite density but overflowing moment must not produce an accepted NaN/Inf norm. invalid = data.copy() invalid[1, :, :] = 1e308 invalid[2, :, :] = -1e308 with self.assertRaisesRegex(RuntimeError, "velocity/profile"): decode_macros(invalid, w, cx, cy) 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_a135252_pins_qualify_their_own_commit(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"): 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", ): source_guard(root, "corrected", "a135252") with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(root, "baseline", "a135252") for name in ("kernels/step/one_step_double.cu", "kernels/blocks.cu", "refinement_stepper.py", "refinement_tables.py"): target = root / "src/CelerisLab/lbm" / name original = target.read_bytes() target.write_bytes(original + b"\n// tamper\n") with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(root, "corrected", "a135252") target.write_bytes(original) def test_historical_default_and_manifest_unchanged(self): historical = { "historical": ("source_pins.json", "87ca033619da1fc02bee822bfc5681b518a92a155c3c3d1dbbe5265c96066b39"), "dev8403956": ("source_pins_dev8403956.json", "58187908485f2af7c78677d53de98015878372444265ce9e667683b8ef5d410f"), } for experiment, (name, expected) in historical.items(): manifest = HERE / "data/les_flow" / name self.assertEqual(hashlib.sha256(manifest.read_bytes()).hexdigest(), expected) for variant in ("baseline", "corrected"): with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(ROOT, variant, experiment) with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(ROOT, "corrected") # Omission still uses the historical profile. 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): with TemporaryDirectory() as directory: root = Path(directory) 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") with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"): source_guard(root, "corrected", "a135252") def test_gpu_requires_opt_in_before_import(self): with TemporaryDirectory() as directory: result = subprocess.run( [ sys.executable, str(HERE / "run_uniform_les_flow.py"), "--root", str(ROOT), "--variant", "baseline", "--output", str(Path(directory) / "out.json"), ], capture_output=True, text=True, ) self.assertEqual(result.returncode, 2) self.assertIn("Explicit --run-gpu is required", result.stderr) self.assertNotIn("pycuda", result.stderr) self.assertFalse((Path(directory) / "out.json").exists()) if __name__ == "__main__": unittest.main()