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