bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-07-10 10:41:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Revert "fix(deprecation): remove all remaining .dumps(), .loads() and producer->connector""

Browse Source 
This reverts commit b12292246f
This commit is contained in:
wakonig_k
2024-02-17 17:56:44 +01:00
committed by Mathias Guijarro
parent 12fda019df
commit f1e9d1ceaa
20 changed files with 90 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
tests/test_eiger9m_csaxs.py
View File

@ -27,7 +27,7 @@ def mock_det():
prefix = "X12SA-ES-EIGER9M:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "producer"):
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
@ -50,7 +50,7 @@ def test_init():
prefix = "X12SA-ES-EIGER9M:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "producer"):
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
@ -428,24 +428,24 @@ def test_publish_file_location(mock_det, scaninfo):
done=scaninfo["done"], successful=scaninfo["successful"]
)
if scaninfo["successful"] is None:
msg = messages.FileMessage(file_path=scaninfo["filepath"], done=scaninfo["done"]).dumps()
msg = messages.FileMessage(file_path=scaninfo["filepath"], done=scaninfo["done"])
else:
msg = messages.FileMessage(
file_path=scaninfo["filepath"], done=scaninfo["done"], successful=scaninfo["successful"]
).dumps()
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
msg,
pipe=mock_det.producer.pipeline.return_value,
pipe=mock_det.connector.pipeline.return_value,
),
mock.call(
MessageEndpoints.file_event(mock_det.name),
msg,
pipe=mock_det.producer.pipeline.return_value,
pipe=mock_det.connector.pipeline.return_value,
),
]
assert mock_det.producer.set_and_publish.call_args_list == expected_calls
assert mock_det.connector.set_and_publish.call_args_list == expected_calls
def test_stop(mock_det):