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

fix: update redis mock for changes in bec

Browse Source
This commit is contained in:
perl_d
2025-02-21 09:07:31 +01:00
committed by wakonig_k
parent 95c931af0b
commit 6a43554f3b
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
tests/unit_tests/test_bec_dispatcher.py
View File

@ -4,7 +4,6 @@ import time
from unittest import mock from unittest import mock
import pytest import pytest
import redis
from bec_lib.messages import ScanMessage from bec_lib.messages import ScanMessage
from bec_lib.serialization import MsgpackSerialization from bec_lib.serialization import MsgpackSerialization
@ -21,13 +20,13 @@ def bec_dispatcher_w_connector(bec_dispatcher, topics_msg_list, send_msg_event):
time.sleep(0.2) time.sleep(0.2)
yield StopIteration yield StopIteration
with mock.patch("redis.Redis"): redis_class_mock = mock.MagicMock()
pubsub = redis.Redis().pubsub() pubsub = redis_class_mock().pubsub()
messages = pubsub_msg_generator() messages = pubsub_msg_generator()
pubsub.get_message.side_effect = lambda timeout: next(messages) pubsub.get_message.side_effect = lambda timeout: next(messages)
connector = QtRedisConnector("localhost:1") connector = QtRedisConnector("localhost:1", redis_class_mock)
bec_dispatcher.client.connector = connector bec_dispatcher.client.connector = connector
yield bec_dispatcher yield bec_dispatcher
dummy_msg = MsgpackSerialization.dumps(ScanMessage(point_id=0, scan_id="0", data={})) dummy_msg = MsgpackSerialization.dumps(ScanMessage(point_id=0, scan_id="0", data={}))