137 lines
6.4 KiB
Python
137 lines
6.4 KiB
Python
"""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
|
|
|
|
ROOT = HERE.parents[1]
|
|
|
|
|
|
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 test_current_corrected_pins(self):
|
|
self.assertEqual(
|
|
source_guard(ROOT, "corrected", "a135252")["base_commit"],
|
|
"a135252ae78b47b57d7e77d19001700b3bedac8b",
|
|
)
|
|
|
|
def test_corrected_and_tamper_guard(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)
|
|
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_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)
|
|
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")
|
|
|
|
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)
|
|
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()
|