tests: recompile installs a guarded stepper; refused close keeps failed state

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-10-30 06:55:26 +00:00
co-authored by Claude Opus 5.5
parent 36eac713f4
commit 71c12aedce
+32 -1
View File
@@ -232,6 +232,32 @@ def test_add_body_and_set_body_route_targets(monkeypatch):
**kwargs) **kwargs)
def test_recompile_installs_a_guarded_stepper(monkeypatch):
from CelerisLab.lbm.stepper import LBMStepper
sim = routed_simulation(monkeypatch)
sim.add_body("circle", (50., 40.), 10., translation=Translation((0., 0., -1., 1.), .1))
launches = []
kernel_module = SimpleNamespace(get_function=lambda name: lambda *a, **k: launches.append(name))
monkeypatch.setattr(module.compiler, "build_module", lambda *a, **k: kernel_module)
sim.lbm_cfg = SimpleNamespace(validate_runtime=lambda: None, viscosity=.01, omega=1.,
streaming="double_buffer", threads_per_block=128,
is_d2q9=True, nx=120, ny=80, nz=1)
sim._compiled_viscosity = .01
sim._assert_allocation_contract = lambda: None
sim._resolve_compile_arch = lambda: "sm_70"
sim._assert_object_count_contract = lambda **k: None
sim.field = SimpleNamespace(module=None, invalidate_module_constant_cache=lambda: None,
_upload_params=lambda: None, _runtime_params={},
ddf_gpu="ddf", temp_gpu="temp", n_curved=0)
sim.stepper = SimpleNamespace(_step_count=7, blocked=None)
sim._recompile()
assert isinstance(sim.stepper, LBMStepper) and sim.stepper.step_count == 7
with pytest.raises(RuntimeError, match="moving kernels"):
sim.stepper.step(1, action_gpu=None, obs_gpu=None)
assert launches == [] and (sim.field.ddf_gpu, sim.field.temp_gpu) == ("ddf", "temp")
assert sim.stepper.step_count == 7
def test_stepper_latch_follows_moving_bodies(monkeypatch): def test_stepper_latch_follows_moving_bodies(monkeypatch):
sim = routed_simulation(monkeypatch) sim = routed_simulation(monkeypatch)
still = sim.add_body("circle", (100., 40.), 5.) still = sim.add_body("circle", (100., 40.), 5.)
@@ -315,6 +341,11 @@ def test_failed_moving_step_marks_simulation_failed(monkeypatch):
call() call()
with pytest.raises(RuntimeError, match="unknown"): # no retry on the failed circle with pytest.raises(RuntimeError, match="unknown"): # no retry on the failed circle
motion.advance(lambda *_: None, lambda: None, lambda: None) motion.advance(lambda *_: None, lambda: None, lambda: None)
sim.close() # close still works with sim.ctx.activate(): # a refused close keeps the failed state and resources
with pytest.raises(RuntimeError, match="active CUDA"):
sim.close()
assert sim._failed is not None and sim.bodies is not None and not sim._closed
sim.close() # retry outside the activation succeeds
assert sim._closed and sim._failed is not None
with pytest.raises(RuntimeError, match="Simulation failed"): # no re-entry after close with pytest.raises(RuntimeError, match="Simulation failed"): # no re-entry after close
sim.run(1) sim.run(1)