CelerisLab/tests/unit/test_les_flow_runner.py

"""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()