usov_i-org/bec
1
0
Fork 0
mirror of https://github.com/ivan-usov-org/bec.git synced 2025-04-21 10:10:02 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

tests(redis_connector): fixed test for messages that are not BECMessages

Browse Source
This commit is contained in:
wakonig_k 2024-12-13 11:16:23 +01:00
parent 64fcbe37b9
commit d42f26dad8
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
bec_lib/tests/test_redis_connector.py
View File

@ -236,19 +236,16 @@ def test_redis_connector_lrange(connector, topic, start, end, use_pipe):
)
def test_redis_connector_set_and_publish(connector, topic, msg, pipe, expire):
if not isinstance(msg, BECMessage):
with pytest.raises(TypeError):
connector.set_and_publish(topic, msg, pipe, expire)
msg_sent = msg
else:
connector.set_and_publish(topic, msg, pipe, expire)
msg_sent = MsgpackSerialization.dumps(msg)
connector._redis_conn.pipeline().publish.assert_called_once_with(
topic, MsgpackSerialization.dumps(msg)
)
connector._redis_conn.pipeline().set.assert_called_once_with(
topic, MsgpackSerialization.dumps(msg), ex=expire
)
if not pipe:
connector._redis_conn.pipeline().execute.assert_called_once()
connector.set_and_publish(topic, msg, pipe, expire)
connector._redis_conn.pipeline().publish.assert_called_once_with(topic, msg_sent)
connector._redis_conn.pipeline().set.assert_called_once_with(topic, msg_sent, ex=expire)
if not pipe:
connector._redis_conn.pipeline().execute.assert_called_once()
@pytest.mark.parametrize("topic, msg, expire", [["topic1", "msg1", None], ["topic2", "msg2", 400]])