bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-02-20 17:28:42 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(async signal tests): create messages with correct async update

Browse Source
This commit is contained in:
wakonig_k
2025-11-13 19:12:12 +01:00
committed by Klaus Wakonig
parent 17f8c826b2
commit 9a23557b9e
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/test_utils.py
View File

@@ -309,7 +309,7 @@ def test_utils_dynamic_signal():
with pytest.raises(ValueError): with pytest.raises(ValueError):
# Missing metadata # Missing metadata
signal.put(messages.DeviceMessage(signals=msg_dict)) signal.put(messages.DeviceMessage(signals=msg_dict))
metadata = {"async_update": {"type": "add"}} metadata = {"async_update": {"type": "add", "max_shape": [None, 1000]}}
msg = messages.DeviceMessage(signals=msg_dict, metadata=metadata) msg = messages.DeviceMessage(signals=msg_dict, metadata=metadata)
signal.put(msg) signal.put(msg)
reading = signal.read() reading = signal.read()
@@ -408,7 +408,9 @@ def test_utils_async_signal():
) )
val = np.random.random(1000) val = np.random.random(1000)
signal.put( signal.put(
val, async_update={"type": "add_slice", "max_shape": [None, 1000]}, acquisition_group="scan" val,
async_update={"type": "add_slice", "max_shape": [None, 1000], "index": 1},
acquisition_group="scan",
) )
reading = signal.read() reading = signal.read()
reading_value = reading[signal.name]["value"].model_dump(exclude={"timestamp"}) reading_value = reading[signal.name]["value"].model_dump(exclude={"timestamp"})
@@ -416,6 +418,7 @@ def test_utils_async_signal():
assert reading_value["metadata"]["async_update"] == { assert reading_value["metadata"]["async_update"] == {
"type": "add_slice", "type": "add_slice",
"max_shape": [None, 1000], "max_shape": [None, 1000],
"index": 1,
} }
assert reading_value["metadata"]["acquisition_group"] == "scan" assert reading_value["metadata"]["acquisition_group"] == "scan"