test: integrate explicitly repinned uniform LES shear diagnostic

tests/unit/test_les_flow_oracle.py

@@ -0,0 +1,70 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
+import json
+import unittest
+import numpy as np
+from les_flow_oracle import H, U, NU, TIMES, solve, relative, amplitude
+
+
+class OracleTests(unittest.TestCase):
+    def test_linear_exact(self):
+        y, states = solve(0, n=256)
+        for t, u in states.items():
+            exact = U * np.sin(np.pi * y / H) * np.exp(-NU * (np.pi / H) ** 2 * t)
+            self.assertLess(relative(u, exact), 1e-5)
+
+    def test_nonlinear_refinement_symmetry_and_dissipation(self):
+        y, states = solve(1, n=256)
+        yc, coarse = solve(1, n=128)
+        energy = U**2 / 2
+        for t, u in states.items():
+            self.assertTrue(np.all(np.isfinite(u)))
+            self.assertGreaterEqual(float(u.min()), 0)
+            self.assertLessEqual(float(u.max()), U)
+            np.testing.assert_allclose(u, u[::-1], atol=1e-14)
+            self.assertLess(float(np.mean(u * u)), energy)
+            energy = float(np.mean(u * u))
+            self.assertLess(relative(np.interp(yc, y, u), coarse[t]), 8e-5)
+
+    def test_filter_is_independent_of_oracle_grid(self):
+        y, a = solve(1, n=128, times=(TIMES[-1],))
+        _, b = solve(0.5, n=128, times=(TIMES[-1],), delta=2)
+        np.testing.assert_array_equal(a[TIMES[-1]], b[TIMES[-1]])
+
+    def test_fixed_grid_timestep_halving(self):
+        errors = []
+        for cs in (0.0, 0.16, 1.0):
+            _, full = solve(cs, n=256, cfl=0.4)
+            _, half = solve(cs, n=256, cfl=0.2)
+            errors.extend(relative(full[t], half[t]) for t in TIMES)
+        self.assertLess(max(errors), 1e-8)
+        print(
+            json.dumps({"fixed_grid_timestep_half_max_relative": max(errors)}),
+            flush=True,
+        )
+
+    def test_signal_and_wrong_convention_separate(self):
+        y, std = solve(1, n=256, times=(TIMES[-1],))
+        _, wrong = solve(2 ** (-0.25), n=256, times=(TIMES[-1],))
+        t = TIMES[-1]
+        base = U * np.exp(-NU * (np.pi / H) ** 2 * t)
+        signal = (base - amplitude(y, std[t])) / base
+        separation = relative(wrong[t], std[t])
+        print(
+            json.dumps(
+                {
+                    "final_time": t,
+                    "standard_sgs_amplitude_signal": signal,
+                    "wrong_convention_profile_separation": separation,
+                }
+            ),
+            flush=True,
+        )
+        self.assertGreater(signal, 0.04)
+        self.assertGreater(separation, 0.01)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_les_flow_runner.py

@@ -0,0 +1,107 @@
+"""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()