test: exercise the timeout wiring end to end and stop leaking live statuses

This commit is contained in:
wyzula_j committed 2026-09-01 17:18:23 +02:00
1 parent 6432aa6d99
commit 00c14b28be
2 files changed
+48 -14

No files matched your search

+25 -9
View File
@@ -106,20 +106,36 @@ def test_status_completed_when_req_done_sub_runs(mock_psi_positioner: PSISimpleP
def test_positioner_timeout_init_arg_survives_positioner_base_init():
"""PSIDeviceBase timeout should remain the default move timeout."""
pos = PSISimplePositioner(name="positioner", prefix="SIM:MOTOR", deadband=0.0013, timeout=3)
with patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
pos = PSISimplePositioner(name="positioner", prefix="SIM:MOTOR", deadband=0.0013, timeout=3)
# The normalized float proves the PSI path owns the value; PositionerBase
# would have stored the raw int (or clobbered it to None).
assert pos._timeout == 3.0
assert isinstance(pos._timeout, float)
assert pos._timeout == 3
pos_no_timeout = PSISimplePositioner(
name="positioner2", prefix="SIM:MOTOR2", deadband=0.0013, timeout=-1
)
assert pos_no_timeout._timeout is None
def test_positioner_move_uses_timeout_init_arg(mock_psi_positioner: PSISimplePositioner):
"""Positioner move statuses should inherit the init timeout by default."""
mock_psi_positioner._timeout = 3
mock_psi_positioner.motor_done_move._read_pv.mock_data = 0
mock_psi_positioner._position = 0
def test_positioner_move_uses_timeout_init_arg():
"""Positioner move statuses should inherit the constructor timeout by default."""
with patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
pos = PSISimplePositioner(name="positioner", prefix="SIM:MOTOR", deadband=0.0013, timeout=3)
pos.wait_for_connection()
patch_dual_pvs(pos)
pos.motor_done_move._read_pv.mock_data = 0
pos._position = 0
st = mock_psi_positioner.move(1, wait=False)
st = pos.move(1, wait=False)
assert st.timeout == 3
assert st.timeout == 3
pos.stop()
def test_positioner_internal_move_status_not_bound_by_default_timeout(
+23 -5
View File
@@ -1188,17 +1188,35 @@ def test_patched_status_types_use_default_timeout_from_object():
assert [status.timeout for status in statuses] == [3, 3, 3, 4, 3, 3, 3, None, 3]
# Finish the statuses so their timeout threads do not fire mid-suite and
# call device.stop() on the test devices.
for status in statuses:
if not status.done:
status.set_finished()
dev.destroy()
pos.destroy()
def test_patched_status_explicit_timeout_overrides_object_default():
"""Explicit status timeouts should take precedence over object defaults."""
dev = TimeoutDevice(name="device", timeout=3)
pos = TimeoutPositioner(name="positioner", timeout=4)
assert StatusBase(obj=dev, timeout=7).timeout == 7
assert Status(obj=dev, timeout=7).timeout == 7
assert DeviceStatus(dev, timeout=7).timeout == 7
assert MoveStatus(pos, target=10, timeout=7).timeout == 7
assert CompareStatus(dev.sig, value=1, timeout=7, run=False).timeout == 7
statuses = [
StatusBase(obj=dev, timeout=7),
Status(obj=dev, timeout=7),
DeviceStatus(dev, timeout=7),
MoveStatus(pos, target=10, timeout=7),
CompareStatus(dev.sig, value=1, timeout=7, run=False),
]
assert [status.timeout for status in statuses] == [7, 7, 7, 7, 7]
for status in statuses:
if not status.done:
status.set_finished()
dev.destroy()
pos.destroy()
def test_patched_status_ignores_signal_internal_timeout():