bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-01-22 09:32:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

test: fix flaky test for positioner is moving

Browse Source
This commit is contained in:
appel_c
2025-02-10 10:48:53 +01:00
parent d7a251dcb1
commit 1cc8a67ded
2 changed files with 34 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ophyd_devices/sim/sim_positioner.py
View File

@@ -190,7 +190,7 @@ class SimPositioner(Device, PositionerBase):
for status in self._status_list: for status in self._status_list:
status.set_exception(exc=exc) status.set_exception(exc=exc)
finally: finally:
self._set_sim_state(self.motor_is_moving.name, 0) self.motor_is_moving.put(0)
if not self._stopped: if not self._stopped:
self._update_state(self.readback.get()) self._update_state(self.readback.get())
self._status_list = [] self._status_list = []
@@ -199,8 +199,8 @@ class SimPositioner(Device, PositionerBase):
"""Change the setpoint of the simulated device, and simultaneously initiate a motion.""" """Change the setpoint of the simulated device, and simultaneously initiate a motion."""
self._stopped = False self._stopped = False
self.check_value(value) self.check_value(value)
self._set_sim_state(self.motor_is_moving.name, 1) self.motor_is_moving.put(1)
self._set_sim_state(self.setpoint.name, value) self.setpoint.put(value)
st = DeviceStatus(device=self) st = DeviceStatus(device=self)
self._status_list.append(st) self._status_list.append(st)
@@ -210,7 +210,7 @@ class SimPositioner(Device, PositionerBase):
self.move_thread.start() self.move_thread.start()
else: else:
self._done_moving() self._done_moving()
self._set_sim_state(self.motor_is_moving.name, 0) self.motor_is_moving.put(0)
self._update_state(value) self._update_state(value)
st.set_finished() st.set_finished()
return st return st

42
tests/test_simulation.py
View File

@@ -268,22 +268,40 @@ def test_positioner_move(positioner):
) )
@pytest.mark.timeout(30)
def test_positioner_motor_is_moving_signal(positioner): def test_positioner_motor_is_moving_signal(positioner):
"""Test that motor is moving is 0 and 1 while (not) moving""" """Test that motor is moving is 0 and 1 while (not) moving"""
done = threading.Event()
def cb_motor_done(): recorded_data = []
done.set() cid = None
positioner.move(0).wait() init_velocity = positioner.velocity.get()
positioner.velocity.set(3)
assert positioner.motor_is_moving.get() == 0 def motor_is_moving_cb(value: any, obj, **kwargs):
status = positioner.move(5) data = obj.read()[f"{obj.name}_motor_is_moving"]["value"]
status.add_callback(cb_motor_done) recorded_data.append(data)
assert positioner.motor_is_moving.get() == 1
status.wait() # Wait will not block until callbacks are executed try:
done.wait(5) # Wait for the additional callback to be executed cid = positioner.subscribe(motor_is_moving_cb, event_type="readback", run=False)
assert positioner.motor_is_moving.get() == 0
status = positioner.move(-20)
status.wait()
positioner.velocity.set(10)
status = positioner.move(20)
status.wait()
# Check that motor was moving and motor_is_moving switched to "1"
assert any(recorded_data)
# Check that motor is not moving and motor_is_moving switched back to "0"
assert recorded_data[-1] == 0
finally:
# Restore initial velocity, remove subscription
positioner.velocity.set(init_velocity)
if cid is not None:
positioner.unsubscribe(cid)
@pytest.mark.parametrize( @pytest.mark.parametrize(