bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

test: added test_zmq_consumer for eiger_plot.py

Browse Source
This commit is contained in:
wyzula-jan
2023-09-27 12:06:53 +02:00
committed by wakonig_k
parent 0a0d51d278
commit c827a25dab
1 changed files with 37 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
tests/test_eiger_plot.py
View File

@ -80,32 +80,40 @@ def test_start_zmq_consumer(eiger_plot_instance):
# TODO fix this test
# def test_zmq_consumer(eiger_plot_instance):
# fake_meta = json.dumps({"type": "int32", "shape": (2, 2)}).encode("utf-8")
# fake_data = np.array([[1, 2], [3, 4]], dtype="int32").tobytes()
#
# with patch("zmq.Context") as MockContext:
# # Mocking zmq socket and its methods
# mock_socket = MagicMock()
# MockContext().socket.return_value = mock_socket
# mock_socket.recv_multipart.return_value = [fake_meta, fake_data]
#
# # Mocking the update_signal to check if it gets emitted
# eiger_plot_instance.update_signal = MagicMock()
#
# # Run the method under test
# eiger_plot_instance.zmq_consumer()
#
# # Check if zmq methods are called
# MockContext.assert_called_once()
# mock_socket.connect.assert_called_with("tcp://129.129.95.38:20000")
# mock_socket.setsockopt_string.assert_called_with(zmq.SUBSCRIBE, "")
# mock_socket.recv_multipart.assert_called_once()
#
# # Check if update_signal was emitted
# eiger_plot_instance.update_signal.emit.assert_called_once()
#
# # Validate the image data
# np.testing.assert_array_equal(
# eiger_plot_instance.image, np.array([[1, 2], [3, 4]], dtype="int32")
# )
def test_zmq_consumer(eiger_plot_instance, qtbot):
fake_meta = json.dumps({"type": "int32", "shape": (2, 2)}).encode("utf-8")
fake_data = np.array([[1, 2], [3, 4]], dtype="int32").tobytes()
with patch("zmq.Context") as MockContext:
MockContext.reset_mock() # Reset the mock here
# Mocking zmq socket and its methods
mock_socket = MagicMock()
MockContext().socket.return_value = mock_socket
mock_socket.recv_multipart.side_effect = [[fake_meta, fake_data], Exception("Break loop")]
# Mocking the update_signal to check if it gets emitted
eiger_plot_instance.update_signal = MagicMock()
try:
# Run the method under test
eiger_plot_instance.zmq_consumer()
except Exception as e:
# Ensure the loop was broken by our mocked exception
assert str(e) == "Break loop"
# Check if zmq methods are called
# MockContext.assert_called_once()
assert MockContext.call_count == 2 # TODO why 2?
mock_socket.connect.assert_called_with("tcp://129.129.95.38:20000")
mock_socket.setsockopt_string.assert_called_with(zmq.SUBSCRIBE, "")
mock_socket.recv_multipart.assert_called()
# Check if update_signal was emitted
eiger_plot_instance.update_signal.emit.assert_called_once()
# Validate the image data
np.testing.assert_array_equal(
eiger_plot_instance.image, np.array([[1, 2], [3, 4]], dtype="int32")
)