simulation: a failed moving step also blocks direct stepper calls
Tests cover failures at the first upload, the step-group sync, the no_donor check, the hold-record upload and its asynchronous sync: before the commit point the target stays pending, after it applied matches the moved device; in every case run and direct stepper reuse are refused. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
c21a365eca
commit
6f9e40caa3
@@ -295,18 +295,28 @@ def test_run_one_advances_and_commits_after_sync(monkeypatch):
|
||||
assert ("step", 3, False) in writes and motion.applied == (0., .25)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", ("upload", "no_donor", "hold_upload"))
|
||||
@pytest.mark.parametrize("failure", ("upload", "first_sync", "no_donor", "hold_upload",
|
||||
"hold_sync"))
|
||||
def test_failed_advance_fails_simulation(monkeypatch, failure):
|
||||
kwargs = dict(no_donor=1.) if failure == "no_donor" else dict(fail=failure)
|
||||
sim, writes, motion = advancing_simulation(monkeypatch, **kwargs)
|
||||
motion._on_failure = sim._fail_moving
|
||||
motion.set_target((0., .25), (0., .05))
|
||||
syncs = []
|
||||
|
||||
def synchronize(): # stands in for an asynchronous failure surfacing at sync
|
||||
syncs.append(1)
|
||||
if (failure, len(syncs)) in (("first_sync", 1), ("hold_sync", 2)):
|
||||
raise RuntimeError("sync failed")
|
||||
with pytest.raises(RuntimeError):
|
||||
sim.run(1, stream=STREAM)
|
||||
# Before the step-group sync nothing is committed; a failed hold-record upload
|
||||
# happens after the commit (applied already matches the moved device geometry).
|
||||
assert motion.applied == ((0., .25) if failure == "hold_upload" else (0., 0.))
|
||||
sim.run(1, stream=SimpleNamespace(synchronize=synchronize))
|
||||
# Before the step-group sync and no_donor check nothing is committed; the hold
|
||||
# record is staged after the commit (applied already matches the moved device).
|
||||
committed = failure in ("hold_upload", "hold_sync")
|
||||
assert motion.applied == ((0., .25) if committed else (0., 0.))
|
||||
assert (motion.pending is None) == committed
|
||||
assert motion.failed and sim._failed is not None
|
||||
assert sim.stepper.blocked == sim._failed # direct stepper reuse refused too
|
||||
with pytest.raises(RuntimeError, match="Simulation failed"):
|
||||
sim.run(1, stream=STREAM)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user