"""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 historical_manifest(collision="SRT", observation_mode="last_base_step"): """The manifest verify_source would select, without hashing the working tree. Receipt/plan logic tests use it; source integrity is checked once in test_source_pin. """ name = ("source_manifest_window.json" if observation_mode == "window" else "source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json") return json.loads((r.HERE / "data/refined_dfg" / name).read_text()) def test_source_pin(): historical = historical_manifest("SRT") assert historical["commit"].startswith("cd3a657") mrt_history = historical_manifest("MRT") assert mrt_history["commit"].startswith("3ea131e") assert historical_manifest("MRT", "window")["commit"].startswith("a233c90") from source_provenance import HistoryUnavailable, historical_sha256 for manifest in (historical, mrt_history, historical_manifest("MRT", "window")): try: # historical integrity: each manifest matches the tree of its own commit actual = historical_sha256(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"]) except HistoryUnavailable as error: unittest.TestCase().skipTest(f"historical integrity NOT checked: {error}") assert actual == manifest["sha256"], manifest["commit"] # Negative controls: the pinned modes refuse a working tree that differs from their pin. 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, historical_manifest("MRT", "window")) def pilot_fixture(path, observation_mode="window", jet_scenario="none", case_selection="both", uniform_diameter=80, placement="wide"): """Synthetic contract fixture only; never a physical result or run receipt.""" manifest = ((historical_manifest("MRT", "window") if jet_scenario == "none" else r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)) if observation_mode == "window" else json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())) selection = dict(case_selection=case_selection, uniform_diameter=uniform_diameter, placement=placement) (path / "plan.json").write_text( json.dumps( dict( source_manifest=manifest, **selection, collision="MRT", jet_scenario=jet_scenario, observation_mode=observation_mode, bounds_root_edge=list(r.PLACEMENTS[placement]), runner_sha256=r.runner_hash(), ) ) ) for case, diameter in r.selected_cases(case_selection, uniform_diameter): 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, **selection, collision="MRT", jet_scenario=jet_scenario, 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, **selection, collision="MRT", jet_scenario=jet_scenario, 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 historical_manifest("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 historical_manifest(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 def test_public_jet_configuration_and_geometry(): for diameter, center, radius in ((40, (80.5, 81), 20), (80, (160, 160.5), 40)): cfg, _, _ = r.config(h, diameter, "MRT", "fixed") assert not cfg["method"]["ddf_shifting"] and not cfg["method"]["les"]["enabled"] jets = r.public_jets(center, radius, "fixed") assert [j.amplitude for j in jets] == [.001, -.001] assert all(j.field.inward is False and j.profile == "parabolic" for j in jets) assert all(j.region.r_max == radius * 1.005 for j in jets) assert all(j.amplitude == 0 for j in r.public_jets(center, radius, "zero")) cfg, _, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed") assert not cfg.ddf_shifting and len(host.boundary_cells) > 0 assert r.public_jets((0, 0), 1, "none") == () with unittest.TestCase().assertRaises(ValueError): r.config(h, 40, "SRT", "fixed") def test_public_jet_reversal_boundary_and_windows(): from types import SimpleNamespace calls = [] handle = SimpleNamespace(set_jet_amplitude=lambda i, a: calls.append((i, a))) for diameter in (40, 80): boundary = round(250 * diameter / r.SPEED) assert r.jet_interval(handle, "fixed", boundary - 3, 80, diameter) == (3, None) switch = round(300 * diameter / r.SPEED) assert r.jet_interval(handle, "reverse", switch - 3, 80, diameter) == (3, None) assert not calls n, event = r.jet_interval(handle, "reverse", switch, 80, diameter) assert n == 80 and event["step"] == switch and event["tstar"] == 300 assert calls == [(0, -.001), (1, .001)] calls.clear() assert r.jet_interval(handle, "reverse", switch + 80, 80, diameter) == (80, None) assert r.jet_interval(handle, "fixed", switch, 80, diameter) == (80, None) assert not calls rows = [dict(tstar=t, tstar_force=t - .5, cd=2.0, cl=-.1) for t in range(251, 306)] windows = r.jet_diagnostics(rows, "reverse") assert len(windows) == 6 and all(w["complete"] for w in windows) assert all(w["cd"]["mean"] == 2 and np.isclose(w["cl"]["mean"], -.1) for w in windows) assert windows[-1]["covered_tstar"] == 5 with unittest.TestCase().assertRaisesRegex(ValueError, "crosses"): r.jet_diagnostics([dict(tstar=250.5, tstar_force=250, cd=1, cl=0)], "fixed") assert r.plan(h, "MRT", "window", "reverse")["modes"]["full_tstar"] == 305 def test_public_jet_source_and_pilot_separation(): from tempfile import TemporaryDirectory with TemporaryDirectory() as tmp: path = Path(tmp) manifest = pilot_fixture(path, jet_scenario="fixed") assert manifest["source_capture"].startswith("current checkout") assert len(manifest["checkout_revision"]) == 40 assert r.verify_pilot(path, manifest, "MRT", "window", "fixed") == 3606.0 for scenario in ("none", "zero", "reverse"): with unittest.TestCase().assertRaisesRegex(ValueError, "Jet scenario"): r.verify_pilot(path, manifest, "MRT", "window", scenario) for relative in ("uniform/summary.json", "refined/ready.json"): receipt = path / relative record = json.loads(receipt.read_text()) receipt.write_text(json.dumps(dict(record, jet_scenario="zero"))) with unittest.TestCase().assertRaises(ValueError): r.verify_pilot(path, manifest, "MRT", "window", "fixed") receipt.write_text(json.dumps(record)) with unittest.TestCase().assertRaises(ValueError): r.verify_source(r.DEFAULT_SOURCE, "MRT", "last_base_step", "fixed") def test_cv_snapshot_common_units_and_window_endpoints(): records = [] for refined, shape, origin, diameter, impulse in ( (True, (240, 568), (38., 22.), 40, [4., -8.]), (False, (330, 1761), (.25, .5), 80, [16., -32.]), (False, (658, 3521), (.375, .75), 160, [64., -128.]), ): rho, ux, uy, fluid = np.ones(shape), np.full(shape, .01), np.zeros(shape), np.ones(shape, bool) begin, n = round(290*diameter/r.SPEED), round(.02*diameter/r.SPEED) record = r.cv_snapshot(rho, ux, uy, fluid, origin, refined, begin, n, diameter, impulse) records.append(record) np.testing.assert_allclose(record["outward_flux"], 0, atol=1e-12) assert r.cv_selected(begin, diameter, "full") and not r.cv_selected(begin-n, diameter, "full") end = round(300*diameter/r.SPEED) assert r.cv_selected(end, diameter, "full") and not r.cv_selected(end+n, diameter, "full") assert r.cv_selected(n, diameter, "pilot") assert sum(r.cv_selected(step, diameter, "full") for step in range(begin-n, end+n+1, n)) == 501 for key in ("time_root", "interval_start_root", "interval_dt_root", "interval_impulse_root", "inventory"): for record in records[1:]: np.testing.assert_allclose(records[0][key], record[key], rtol=0, atol=1e-10) assert records[0]["interval_dt_root"] == 80 and np.isclose(records[0]["tstar"], 290.02) np.testing.assert_array_equal(records[0]["interval_impulse_root"], [4, -8]) def test_cv_pilot_mode_and_helper_provenance(): from tempfile import TemporaryDirectory manifest = r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", "fixed", True) assert "tests/validation/run_jet_circle_cv.py" in manifest["sha256"] with TemporaryDirectory() as tmp: path = Path(tmp) old = pilot_fixture(path, jet_scenario="fixed") with unittest.TestCase().assertRaisesRegex(ValueError, "CV observation mode"): r.verify_pilot(path, old, "MRT", "window", "fixed", True) for relative in ("plan.json", "uniform/summary.json", "refined/summary.json", "uniform/ready.json", "refined/ready.json"): file = path / relative record = json.loads(file.read_text()) record.update(observe_cv=True, cv_samples=250) if relative == "plan.json": record["source_manifest"] = manifest file.write_text(json.dumps(record)) assert r.verify_pilot(path, manifest, "MRT", "window", "fixed", True) == 3606 def test_compact_geometry_and_cv_stencils(): from CelerisLab.lbm.refinement import ACTIVE from CelerisLab.lbm.descriptors import SOLID _, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed", "compact") assert pair.fine_shape == (168, 168) and tuple(pair.fine_origin) == (46, 46) fluid = (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0) rho, ux, uy = np.ones(pair.fine_shape), np.full(pair.fine_shape, .01), np.zeros(pair.fine_shape) record = r.cv_snapshot(rho, ux, uy, fluid, pair.fine_origin, True, 1160000, 80, 40, [1, 2], "compact") np.testing.assert_allclose(record["outward_flux"], 0, atol=1e-12) stencil = np.zeros(pair.fine_shape, bool) for rect in r.cv_contours(True, 40, "compact"): x0, y0, x1, y1 = np.rint((rect - np.tile(pair.fine_origin, 2))/.5).astype(int) stencil[y0-2:y1+2, x0-2:x0+2] = stencil[y0-2:y1+2, x1-2:x1+2] = True stencil[y0-2:y0+2, x0-2:x1+2] = stencil[y1-2:y1+2, x0-2:x1+2] = True assert fluid[stencil].all() and not stencil.ravel()[host.transfer_fine_cells].any() def test_d160_config_and_single_case_receipts(): from tempfile import TemporaryDirectory from unittest.mock import patch cfg, _, p = r.config(h, 160, "MRT", "fixed") assert (p["nx"], p["ny"], p["center"], p["radius"]) == (3521, 658, [320, 320.5], 80) assert p["viscosity"] == .016 and p["mean_speed"] == .01 and not cfg["method"]["ddf_shifting"] assert r.uniform_origin(160) == (.375, .75) np.testing.assert_array_equal(r.cv_contours(False, 160), r.CV_CONTOURS + [.125, 0, .125, 0]) for scenario, expected_exit in (("fixed", 0), ("zero", 2)): with TemporaryDirectory() as tmp: path = Path(tmp) manifest = pilot_fixture(path, jet_scenario=scenario, case_selection="uniform", uniform_diameter=160) assert r.verify_pilot(path, manifest, "MRT", "window", scenario, False, "uniform", 160) == 1803 for selection, diameter, placement in (("both", 160, "wide"), ("uniform", 80, "wide"), ("uniform", 160, "compact")): with unittest.TestCase().assertRaises(ValueError): r.verify_pilot(path, manifest, "MRT", "window", scenario, False, selection, diameter, placement) # Synthetic dispatch/result check: no CUDA imports or execution. output = path / "full" argv = ["runner", "--collision", "MRT", "--exchange-window", "--jet-scenario", scenario, "--case", "uniform", "--uniform-diameter", "160", "--mode", "full", "--pilot-results", str(path), "--output", str(output), "--max-wall-seconds", "10", "--gpu-gates-approved"] result = dict(case="uniform", healthy=True, periodic=scenario == "zero", jet_diagnostic_windows=[], status="driven_diagnostics_complete" if scenario == "fixed" else "periodic", reference_comparison=dict(reference_agreement=False)) with patch.object(sys, "argv", argv), patch.dict(r.os.environ, CUDA_VISIBLE_DEVICES="mock-no-gpu"), \ patch.object(r, "run_case", return_value=result) as run: assert r.main() == expected_exit and run.call_count == 1 comparison = json.loads((output / "comparison.json").read_text()) assert comparison["matched_case_pass"] is False and "refined_vs_uniform" not in comparison assert comparison["status"].startswith("single_case_") 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())