462 lines
18 KiB
Python
462 lines
18 KiB
Python
"""CPU-only contract tests. These are not CUDA or DFG acceptance evidence."""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
import numpy as np
|
|
import unittest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
|
|
import run_refined_dfg as r
|
|
|
|
h, periodic = r.helpers(r.DEFAULT_SOURCE)
|
|
|
|
|
|
def test_source_pin():
|
|
historical = json.loads((r.HERE / "data/refined_dfg/source_manifest.json").read_text())
|
|
assert historical["commit"].startswith("cd3a657")
|
|
mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
|
|
assert mrt_history["commit"].startswith("3ea131e")
|
|
assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["commit"].startswith("a233c90")
|
|
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
|
|
r.verify_source(r.DEFAULT_SOURCE, "MRT")
|
|
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
|
|
r.verify_source(r.DEFAULT_SOURCE, "SRT")
|
|
|
|
|
|
def test_matched_units_and_planes():
|
|
p = r.plan(h)
|
|
c = p["root"]
|
|
f = p["uniform"]
|
|
assert c["nx"] == 881 and c["ny"] == 166
|
|
assert f["nx"] == 1761 and f["ny"] == 330
|
|
assert c["dx"] / 2 == f["dx"] and c["dt"] / 2 == f["dt"]
|
|
assert np.isclose(c["viscosity"] * c["dx"] ** 2 / c["dt"], 0.001)
|
|
assert np.isclose(f["viscosity"] * f["dx"] ** 2 / f["dt"], 0.001)
|
|
assert np.isclose((p["body_root_edge"][0] - 0.5) * c["dx"], 0.2)
|
|
assert np.isclose((p["body_root_edge"][1] - 1) * c["dx"], 0.2)
|
|
assert np.isclose((165 - 1) * c["dx"], 0.41)
|
|
assert np.isclose((880.5 - 0.5) * c["dx"], 2.2)
|
|
assert np.isclose(2 * p["radius_root"] * c["dx"], 0.1)
|
|
|
|
|
|
def test_same_numerics():
|
|
c, _, _ = r.config(h, 40)
|
|
f, _, _ = r.config(h, 80)
|
|
assert c["method"] == f["method"]
|
|
m = c["method"]
|
|
assert m["collision"] == "SRT" and m["streaming"] == "double_buffer"
|
|
assert m["store_precision"] == "FP32" and m["ddf_shifting"]
|
|
assert not m["les"]["enabled"]
|
|
assert m["inlet"]["scheme"] == "zou_he_local"
|
|
assert m["outlet"]["mode"] == "neq_extrap" and m["outlet"]["srt_neq_damp"] == 0.5
|
|
assert not m["outlet"]["backflow_clamp"]
|
|
|
|
|
|
def test_actual_geometry_guards():
|
|
cfg, pair, body, host = r.host_geometry(r.DEFAULT_SOURCE, h)
|
|
assert pair.fine_shape == (240, 568)
|
|
assert np.allclose(
|
|
host.objects[0].geometry.center,
|
|
[
|
|
(80.5 - pair.fine_origin[0]) / 0.5 - 0.5,
|
|
(81 - pair.fine_origin[1]) / 0.5 - 0.5,
|
|
],
|
|
)
|
|
assert cfg.nx % r.BASE_BLOCK[1] == 0 and cfg.ny % r.BASE_BLOCK[0] == 0
|
|
assert len(host.boundary_cells) > 0
|
|
|
|
|
|
def test_physical_leaf_mass_excludes_covered_and_halo():
|
|
c = np.full((9, 4, 4), 1 / 9)
|
|
f = np.full((9, 8, 8), 1 / 9)
|
|
cm = np.ones((4, 4), bool)
|
|
cm[1:3, 1:3] = False
|
|
fm = np.zeros((8, 8), bool)
|
|
fm[2:6, 2:6] = True
|
|
c[:, ~cm] = np.nan
|
|
f[:, ~fm] = np.nan
|
|
assert np.isclose(r.leaf_mass([c, f], 2, [cm, fm]), 64)
|
|
f[:, 3, 3] = np.nan
|
|
with unittest.TestCase().assertRaisesRegex(ValueError, "Nonfinite"):
|
|
r.leaf_mass([c, f], 2, [cm, fm])
|
|
|
|
|
|
def test_force_impulse_interval_scaling():
|
|
from CelerisLab.lbm.raw_exchange import summarize_exchange
|
|
|
|
# Two fine substeps each j=(2,4), h=.5 => root impulse=(1,2).
|
|
result = summarize_exchange(
|
|
[[1.0, 2.0, 0.0, 0.0]],
|
|
[0],
|
|
[False],
|
|
body_id=0,
|
|
level=1,
|
|
h=0.5,
|
|
time_end=8,
|
|
window="last_base_step",
|
|
reference_center=(80.5, 81.0),
|
|
)
|
|
assert result["sample_count"] == 2 and result["time_start"] == 7
|
|
np.testing.assert_allclose(result["total"]["force"], [1, 2])
|
|
assert result["units"] == "base_lattice_per_unit_depth"
|
|
# Force unit scales dx^3/dt^2, thus uniform fine force is twice root.
|
|
root_cd = 1 / (0.5 * r.SPEED**2 * 40)
|
|
fine_cd = 2 / (0.5 * r.SPEED**2 * 80)
|
|
assert root_cd == fine_cd
|
|
|
|
|
|
def test_pressure_local_frame_and_units():
|
|
_, pair, _, _ = r.host_geometry(r.DEFAULT_SOURCE, h)
|
|
center = [
|
|
(80.5 - pair.fine_origin[0]) / 0.5 - 0.5,
|
|
(81 - pair.fine_origin[1]) / 0.5 - 0.5,
|
|
]
|
|
p = {"center": center, "radius": 40.0, "mean_speed": r.SPEED}
|
|
yy, xx = np.indices(pair.fine_shape)
|
|
rho = 1 + 1e-7 * xx
|
|
probes = h.pressure_probes(rho, np.ones(pair.fine_shape, bool), p)
|
|
assert np.isclose(probes["dp"], -80e-7 / (3 * r.SPEED**2))
|
|
|
|
|
|
def test_initializer_coordinates_and_rest_ghosts():
|
|
g = r.initial_state((166, 881), (0, 0), 1, root=True)
|
|
rho, ux, uy = r.macros(g)
|
|
assert np.max(abs(rho - 1)) < 1e-14
|
|
assert np.max(abs(uy)) == 0
|
|
assert np.max(abs(ux[[0, -1]])) == 0
|
|
assert np.isclose(ux[81, 50], 6 * r.SPEED * (80.5 / 164) * (1 - 80.5 / 164))
|
|
|
|
|
|
def test_axis_specific_sampling_phase():
|
|
_, pair, _, _ = r.host_geometry(r.DEFAULT_SOURCE, h)
|
|
assert pair.fine_origin == (38.0, 22.0)
|
|
j = np.arange(pair.fine_shape[1])
|
|
k = np.arange(pair.fine_shape[0])
|
|
fine_x = (pair.fine_origin[0] + (j + 0.5) / 2 - 0.5) * 0.0025
|
|
fine_y = (pair.fine_origin[1] + (k + 0.5) / 2 - 1) * 0.0025
|
|
np.testing.assert_allclose(fine_x / 0.00125, j + 75.5, atol=1e-12)
|
|
np.testing.assert_allclose(fine_y / 0.00125, (k + 43) - 0.5, atol=1e-12)
|
|
np.testing.assert_allclose(fine_x - (j + 75) * 0.00125, 0.5 * 0.00125, atol=1e-15)
|
|
np.testing.assert_allclose(fine_y - ((k + 43) - 0.5) * 0.00125, 0, atol=1e-15)
|
|
|
|
|
|
def test_full_requires_completed_pilot():
|
|
with unittest.TestCase().assertRaisesRegex(ValueError, "pilot-results"):
|
|
r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
|
|
|
|
|
|
def pilot_fixture(path, observation_mode="window"):
|
|
"""Synthetic contract fixture only; never a physical result or run receipt."""
|
|
manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
|
|
if observation_mode == "window" else
|
|
json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
|
|
(path / "plan.json").write_text(
|
|
json.dumps(
|
|
dict(
|
|
source_manifest=manifest,
|
|
collision="MRT",
|
|
observation_mode=observation_mode,
|
|
bounds_root_edge=list(r.BOUNDS),
|
|
runner_sha256=r.runner_hash(),
|
|
)
|
|
)
|
|
)
|
|
for case, diameter in (("uniform", 80), ("refined", 40)):
|
|
dest = path / case
|
|
dest.mkdir()
|
|
interval = round(0.02 * diameter / r.SPEED)
|
|
rows = [
|
|
dict(
|
|
step=(i + 1) * interval,
|
|
tstar=(i + 1) * 0.02,
|
|
tstar_force=(i + 0.5) * 0.02,
|
|
cd=3.2,
|
|
cl=0.1,
|
|
dp=2.0,
|
|
mass=0.8,
|
|
rho_min=0.99,
|
|
rho_max=1.01,
|
|
max_mach=0.03,
|
|
flux_in=0.4,
|
|
flux_out=0.4,
|
|
rho_front=1.0,
|
|
rho_back=1.0,
|
|
)
|
|
for i in range(250)
|
|
]
|
|
(dest / "summary.json").write_text(
|
|
json.dumps(
|
|
dict(
|
|
case=case,
|
|
collision="MRT",
|
|
observation_mode=observation_mode,
|
|
mode="pilot",
|
|
status="pilot_complete",
|
|
healthy=True,
|
|
steps=250 * interval,
|
|
history=rows,
|
|
startup_seconds=1.0,
|
|
wall_seconds=11.0,
|
|
step_seconds=9.0,
|
|
)
|
|
)
|
|
)
|
|
(dest / "ready.json").write_text(
|
|
json.dumps(
|
|
dict(
|
|
case=case,
|
|
collision="MRT",
|
|
observation_mode=observation_mode,
|
|
mode="pilot",
|
|
step=interval,
|
|
runner_sha256=r.runner_hash(),
|
|
all_runtime_and_observation_paths_exercised=True,
|
|
)
|
|
)
|
|
)
|
|
return manifest
|
|
|
|
|
|
def test_pilot_validation_and_calibration():
|
|
from tempfile import TemporaryDirectory
|
|
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp)
|
|
manifest = pilot_fixture(path)
|
|
assert r.verify_pilot(path, manifest, "MRT", "window") == 3606.0
|
|
|
|
|
|
def test_reject_malformed_pilot_receipts():
|
|
from tempfile import TemporaryDirectory
|
|
from copy import deepcopy
|
|
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp)
|
|
manifest = pilot_fixture(path)
|
|
receipt = path / "uniform/summary.json"
|
|
valid = json.loads(receipt.read_text())
|
|
mutations = [
|
|
lambda s: s.update(case="refined"),
|
|
lambda s: s.update(mode="full"),
|
|
lambda s: s.update(healthy="true"),
|
|
lambda s: s.update(steps=1),
|
|
lambda s: s["history"].pop(),
|
|
lambda s: s["history"][1].update(step=160),
|
|
lambda s: s["history"][-1].update(tstar=float("nan")),
|
|
lambda s: s["history"][0].update(cd=float("inf")),
|
|
lambda s: s["history"][0].update(max_mach=0.3),
|
|
lambda s: s["history"][0].update(tstar_force=0.02),
|
|
lambda s: s.update(wall_seconds=0),
|
|
]
|
|
for mutate in mutations:
|
|
bad = deepcopy(valid)
|
|
mutate(bad)
|
|
receipt.write_text(json.dumps(bad))
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", "window")
|
|
# Exact reviewer regression: previously accepted NaN endpoint.
|
|
receipt.write_text(
|
|
json.dumps(
|
|
dict(
|
|
status="pilot_complete",
|
|
healthy=True,
|
|
history=[{"tstar": float("nan")}],
|
|
)
|
|
)
|
|
)
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", "window")
|
|
receipt.write_text(json.dumps(valid))
|
|
(path / "uniform/ready.json").write_text("{}")
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", "window")
|
|
|
|
|
|
def test_strict_finite_json_and_deadline():
|
|
from tempfile import TemporaryDirectory
|
|
|
|
with TemporaryDirectory() as tmp:
|
|
p = Path(tmp) / "bad.json"
|
|
for data in ('{"x": NaN}', '{"x": Infinity}', '{"x": 1e999}'):
|
|
p.write_text(data)
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.read_finite_json(p)
|
|
with unittest.TestCase().assertRaises(TimeoutError):
|
|
r.check_deadline(0)
|
|
r.check_deadline(float("inf"))
|
|
|
|
|
|
def test_pressure_probes_actual_body_mask():
|
|
from CelerisLab.lbm.descriptors import SOLID
|
|
from CelerisLab.lbm.refinement import ACTIVE
|
|
|
|
_, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h)
|
|
fluid = (pair.fine_kind == ACTIVE) & (
|
|
(host.flags.reshape(pair.fine_shape) & SOLID) == 0
|
|
)
|
|
p = dict(center=host.objects[0].geometry.center, radius=40.0, mean_speed=r.SPEED)
|
|
assert h.pressure_probes(np.ones(pair.fine_shape), fluid, p)["dp"] == 0.0
|
|
|
|
|
|
def test_matched_mrt_selection():
|
|
for diameter in (40, 80):
|
|
srt, sbody, sp = r.config(h, diameter)
|
|
mrt, mbody, mp = r.config(h, diameter, "MRT")
|
|
assert mrt["method"]["collision"] == "MRT"
|
|
mrt["method"]["collision"] = "SRT"
|
|
assert mrt == srt and mbody == sbody and mp == sp
|
|
srt_plan, mrt_plan = r.plan(h), r.plan(h, "MRT")
|
|
for key in ("root", "uniform", "bounds_root_edge", "finest_dx_m", "finest_dt_s"):
|
|
assert srt_plan[key] == mrt_plan[key]
|
|
assert "q2/6/8" in mrt_plan["outlet"] and "all nine" in srt_plan["outlet"]
|
|
cfg, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT")
|
|
assert cfg.collision == "MRT" and pair.fine_shape == (240, 568)
|
|
assert len(host.boundary_cells) > 0
|
|
|
|
|
|
def test_reject_collision_receipts():
|
|
from tempfile import TemporaryDirectory
|
|
|
|
for rel in ("plan.json", "uniform/summary.json", "refined/summary.json",
|
|
"uniform/ready.json", "refined/ready.json"):
|
|
for invalid in ("SRT", None):
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp)
|
|
manifest = pilot_fixture(path)
|
|
receipt = path / rel
|
|
record = json.loads(receipt.read_text())
|
|
if invalid is None:
|
|
record.pop("collision")
|
|
else:
|
|
record["collision"] = invalid
|
|
receipt.write_text(json.dumps(record))
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", "window")
|
|
|
|
|
|
def test_record_actual_checkout_separately():
|
|
from source_provenance import checkout_provenance
|
|
|
|
actual = checkout_provenance(r.DEFAULT_SOURCE)
|
|
assert len(actual["checkout_revision"]) == 40
|
|
assert isinstance(actual["checkout_dirty"], bool)
|
|
assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["production_origin_commit"].startswith("d930")
|
|
|
|
|
|
def test_window_plan_preserves_physics():
|
|
for collision in ("SRT", "MRT"):
|
|
baseline = r.plan(h, collision)
|
|
window = r.plan(h, collision, "window")
|
|
assert baseline.pop("observation_mode") == "last_base_step"
|
|
assert window.pop("observation_mode") == "window"
|
|
assert baseline == window
|
|
assert r.verify_source(r.DEFAULT_SOURCE, collision, "window")["observation_mode"] == "window"
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_source(r.DEFAULT_SOURCE, "MRT", "unknown")
|
|
|
|
|
|
def test_reject_observation_mode_receipts():
|
|
from tempfile import TemporaryDirectory
|
|
|
|
for mode, other in (("window", "last_base_step"), ("last_base_step", "window")):
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp)
|
|
manifest = pilot_fixture(path, mode)
|
|
assert r.verify_pilot(path, manifest, "MRT", mode) == 3606.0
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", other)
|
|
for rel in ("plan.json", "uniform/summary.json", "refined/summary.json",
|
|
"uniform/ready.json", "refined/ready.json"):
|
|
for invalid in (other, None):
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp)
|
|
manifest = pilot_fixture(path, mode)
|
|
receipt = path / rel
|
|
record = json.loads(receipt.read_text())
|
|
if invalid is None:
|
|
record.pop("observation_mode")
|
|
else:
|
|
record["observation_mode"] = invalid
|
|
receipt.write_text(json.dumps(record))
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
r.verify_pilot(path, manifest, "MRT", mode)
|
|
|
|
|
|
class ExchangeFixture:
|
|
"""Synthetic interval ledger for orchestration tests, not a solver or GPU result."""
|
|
def __init__(self):
|
|
self.step_count = 11
|
|
self.events = []
|
|
self.begin = None
|
|
self.override = {}
|
|
|
|
def start_exchange_window(self):
|
|
self.begin = self.step_count
|
|
self.events.append(("start", self.step_count))
|
|
|
|
def step(self, n):
|
|
self.events.append(("step", n))
|
|
self.step_count += n
|
|
|
|
def result(self, start, window):
|
|
from CelerisLab.lbm.raw_exchange import summarize_exchange
|
|
|
|
total = sum(range(start + 1, self.step_count + 1))
|
|
result = summarize_exchange([[total, -2*total, 0., 0.]], [0], [False],
|
|
body_id=0, level=1, h=.5, time_end=self.step_count, window=window,
|
|
reference_center=(80.5, 81.),
|
|
**({"duration": self.step_count-start} if window == "window" else {}))
|
|
result.update(self.override)
|
|
return result
|
|
|
|
def read_exchange_window(self, body_id, *, level):
|
|
assert body_id == 0 and level == 1
|
|
self.events.append(("read_window", self.step_count))
|
|
return self.result(self.begin, "window")
|
|
|
|
def read_raw_exchange(self, body_id, *, level, window):
|
|
assert body_id == 0 and level == 1 and window == "last_base_step"
|
|
self.events.append(("read_raw", self.step_count))
|
|
return self.result(self.step_count-1, window)
|
|
|
|
|
|
def test_window_interval_impulse_sampling_and_units():
|
|
base, window = ExchangeFixture(), ExchangeFixture()
|
|
for n in (80, 2000, 3): # Pilot/dense, early full, and final partial interval.
|
|
begin = base.step_count
|
|
a = r.advance_refined_interval(base, n, float("inf"), "last_base_step")
|
|
w = r.advance_refined_interval(window, n, float("inf"), "window")
|
|
np.testing.assert_array_equal(w, a)
|
|
expected = sum(range(begin+1, begin+n+1))
|
|
np.testing.assert_array_equal(w, [expected, -2*expected])
|
|
den = .5 * r.SPEED**2 * 40
|
|
np.testing.assert_array_equal(w/n/den, a/n/den)
|
|
assert window.step_count == base.step_count == begin+n
|
|
assert window.events[-3:] == [("start", begin), ("step", n), ("read_window", begin+n)]
|
|
assert (begin+n/2)*r.SPEED/40 == (window.step_count-n/2)*r.SPEED/40
|
|
assert not any(event[0] == "read_raw" for event in window.events)
|
|
|
|
|
|
def test_window_contract_rejection():
|
|
for key, value in (("units", "local"), ("window", "last_base_step"),
|
|
("sample_count", 2), ("time_start", 0), ("time_end", 0), ("dt", 1.)):
|
|
backend = ExchangeFixture()
|
|
backend.override[key] = value
|
|
with unittest.TestCase().assertRaisesRegex(RuntimeError, "window force"):
|
|
r.advance_refined_interval(backend, 80, float("inf"), "window")
|
|
backend = ExchangeFixture()
|
|
with unittest.TestCase().assertRaises(TimeoutError):
|
|
r.advance_refined_interval(backend, 80, 0, "window")
|
|
assert not backend.events
|
|
|
|
|
|
if __name__ == "__main__":
|
|
suite = unittest.TestSuite(
|
|
unittest.FunctionTestCase(fn)
|
|
for name, fn in sorted(globals().items())
|
|
if name.startswith("test_")
|
|
)
|
|
result = unittest.TextTestRunner(verbosity=2).run(suite)
|
|
raise SystemExit(not result.wasSuccessful())
|