108 lines
4.1 KiB
Python
108 lines
4.1 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_baseline_pins(self):
|
|
self.assertEqual(
|
|
source_guard(ROOT, "baseline")["base_commit"],
|
|
"cd3a657f9d3833e812c3cca173525f165b48710a",
|
|
)
|
|
|
|
def test_corrected_and_tamper_guard(self):
|
|
pins = json.loads((HERE / "data/les_flow/source_pins.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)
|
|
# Rebuild the reviewed candidate from actual pinned headers, not fixtures.
|
|
candidate = root / "header_candidate"
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(HERE / "les/prepare.py"),
|
|
"--repo",
|
|
str(ROOT),
|
|
"--output-dir",
|
|
str(candidate),
|
|
],
|
|
check=True,
|
|
)
|
|
shutil.copyfile(candidate / "patched" / HEADER, root / HEADER)
|
|
with patch(
|
|
"run_uniform_les_flow.subprocess.check_output",
|
|
return_value="isolated-fixture\n",
|
|
):
|
|
source_guard(root, "corrected")
|
|
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
|
|
source_guard(root, "baseline")
|
|
target = root / "src/CelerisLab/lbm/kernels/step/one_step_double.cu"
|
|
target.write_bytes(target.read_bytes() + b"\n// tamper\n")
|
|
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
|
|
source_guard(root, "corrected")
|
|
|
|
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()
|