5fce576e33
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
redis-py raises redis.exceptions.TimeoutError when a blocking XREADGROUP returns no data within block_ms. On idle streams (US market closed → no market:bars / signals:generated / trades:executed) every blocking read times out; the unhandled exception tore down each worker's asyncio.TaskGroup and exited the process, putting signal-generator, trade-executor and learning-engine into CrashLoopBackOff. Catch it and keep polling. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
"""Tests for the Redis Streams publish/consume helpers."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from shared.redis_streams import StreamConsumer, StreamPublisher
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publisher_publishes_json() -> None:
|
|
"""StreamPublisher should XADD a JSON-serialised payload."""
|
|
redis = AsyncMock()
|
|
redis.xadd = AsyncMock(return_value=b"1-0")
|
|
|
|
pub = StreamPublisher(redis, "test:stream")
|
|
msg_id = await pub.publish({"ticker": "AAPL", "score": 0.8})
|
|
|
|
redis.xadd.assert_called_once_with(
|
|
"test:stream",
|
|
{"data": json.dumps({"ticker": "AAPL", "score": 0.8})},
|
|
)
|
|
assert msg_id == b"1-0"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publisher_returns_message_id() -> None:
|
|
"""The returned value must be the raw message ID from Redis."""
|
|
redis = AsyncMock()
|
|
redis.xadd = AsyncMock(return_value=b"99999-42")
|
|
|
|
pub = StreamPublisher(redis, "events")
|
|
result = await pub.publish({"key": "value"})
|
|
|
|
assert result == b"99999-42"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumer_ensure_group_creates_group() -> None:
|
|
"""ensure_group should call XGROUP CREATE with mkstream=True."""
|
|
redis = AsyncMock()
|
|
redis.xgroup_create = AsyncMock()
|
|
|
|
consumer = StreamConsumer(redis, "test:stream", "my-group", "worker-1")
|
|
await consumer.ensure_group()
|
|
|
|
redis.xgroup_create.assert_called_once_with(
|
|
"test:stream", "my-group", id="0", mkstream=True
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumer_ensure_group_ignores_existing() -> None:
|
|
"""If the group already exists the exception should be swallowed."""
|
|
redis = AsyncMock()
|
|
redis.xgroup_create = AsyncMock(side_effect=Exception("BUSYGROUP"))
|
|
|
|
consumer = StreamConsumer(redis, "test:stream", "my-group", "worker-1")
|
|
# Should not raise
|
|
await consumer.ensure_group()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumer_consume_yields_and_acks() -> None:
|
|
"""consume() should yield deserialised data and ACK each message."""
|
|
redis = AsyncMock()
|
|
redis.xgroup_create = AsyncMock()
|
|
|
|
payload = {"ticker": "TSLA", "direction": "LONG"}
|
|
# xreadgroup returns list of (stream, [(msg_id, fields), ...])
|
|
redis.xreadgroup = AsyncMock(
|
|
side_effect=[
|
|
[
|
|
(
|
|
b"test:stream",
|
|
[(b"1-0", {b"data": json.dumps(payload).encode()})],
|
|
)
|
|
],
|
|
# Second call returns empty to break the loop in the test
|
|
KeyboardInterrupt,
|
|
]
|
|
)
|
|
redis.xack = AsyncMock()
|
|
|
|
consumer = StreamConsumer(redis, "test:stream", "grp", "c1")
|
|
|
|
results: list[tuple[str, dict]] = []
|
|
try:
|
|
async for msg_id, data in consumer.consume(batch_size=1, block_ms=100):
|
|
results.append((msg_id, data))
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
assert len(results) == 1
|
|
assert results[0] == (b"1-0", payload)
|
|
redis.xack.assert_called_once_with("test:stream", "grp", b"1-0")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consume_survives_idle_blocking_timeout() -> None:
|
|
"""A blocking-read TimeoutError on an idle stream must not kill the loop.
|
|
|
|
redis-py raises ``redis.exceptions.TimeoutError`` when a blocking
|
|
``XREADGROUP`` returns no data within ``block_ms`` (idle stream). This is
|
|
an expected idle condition, not a fatal error — ``consume`` must swallow it
|
|
and keep polling.
|
|
"""
|
|
from redis.exceptions import TimeoutError as RedisTimeoutError
|
|
|
|
redis = AsyncMock()
|
|
redis.xgroup_create = AsyncMock()
|
|
|
|
payload = {"ticker": "NVDA", "direction": "LONG"}
|
|
redis.xreadgroup = AsyncMock(
|
|
side_effect=[
|
|
RedisTimeoutError("Timeout reading from redis"), # idle stream
|
|
[(b"test:stream", [(b"5-0", {b"data": json.dumps(payload).encode()})])],
|
|
KeyboardInterrupt, # break the loop in the test
|
|
]
|
|
)
|
|
redis.xack = AsyncMock()
|
|
|
|
consumer = StreamConsumer(redis, "test:stream", "grp", "c1")
|
|
|
|
results: list[tuple[str, dict]] = []
|
|
try:
|
|
async for msg_id, data in consumer.consume(batch_size=1, block_ms=100):
|
|
results.append((msg_id, data))
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
# The idle timeout was swallowed; the next real message still came through.
|
|
assert results == [(b"5-0", payload)]
|
|
assert redis.xreadgroup.call_count == 3
|