Redis

Redis Pub/Sub and Streams Testing Patterns

HelpMeTest

03 Sep 2026 — 6 min read

Redis Pub/Sub and Redis Streams are fundamentally different messaging primitives, and each requires distinct testing approaches. Pub/Sub is ephemeral—messages are lost if no subscriber is connected. Streams are persistent—messages are stored until explicitly deleted. This guide shows how to test both reliably.

Redis Pub/Sub Testing

The Challenge

Pub/Sub involves concurrent producers and subscribers. A test that publishes a message must wait for the subscriber to receive it without using sleep. This requires synchronization primitives—events, promises, or queues.

Basic Pub/Sub Test Setup

# notification_service.py
import json
import redis
import threading

class NotificationService:
    def __init__(self, redis_url: str):
        self.publisher = redis.Redis.from_url(redis_url, decode_responses=True)
        self.subscriber = redis.Redis.from_url(redis_url, decode_responses=True)
        self._handlers = {}
        self._pubsub = None
        self._thread = None
    
    def publish(self, channel: str, event_type: str, data: dict):
        message = json.dumps({'type': event_type, 'data': data})
        subscriber_count = self.publisher.publish(channel, message)
        return subscriber_count
    
    def subscribe(self, channel: str, handler):
        self._handlers[channel] = handler
        self._pubsub = self.subscriber.pubsub()
        self._pubsub.subscribe(channel)
        self._thread = threading.Thread(target=self._listen, daemon=True)
        self._thread.start()
    
    def _listen(self):
        for message in self._pubsub.listen():
            if message['type'] == 'message':
                channel = message['channel']
                data = json.loads(message['data'])
                handler = self._handlers.get(channel)
                if handler:
                    handler(data)
    
    def unsubscribe(self):
        if self._pubsub:
            self._pubsub.unsubscribe()
            self._pubsub.close()

# test_notification_service.py
import threading
import pytest
from testcontainers.redis import RedisContainer
from notification_service import NotificationService

@pytest.fixture(scope='module')
def redis_url():
    with RedisContainer('redis:7-alpine') as container:
        host = container.get_container_host_ip()
        port = container.get_exposed_port(6379)
        yield f'redis://{host}:{port}'

def test_subscriber_receives_published_message(redis_url):
    service = NotificationService(redis_url)
    
    received = []
    received_event = threading.Event()
    
    def handler(message):
        received.append(message)
        received_event.set()
    
    service.subscribe('notifications', handler)
    
    # Give subscriber thread time to connect
    import time
    time.sleep(0.1)
    
    service.publish('notifications', 'user.created', {'id': 1, 'name': 'Alice'})
    
    # Wait for message (timeout = test failure)
    assert received_event.wait(timeout=2), "Message not received within 2 seconds"
    
    assert len(received) == 1
    assert received[0]['type'] == 'user.created'
    assert received[0]['data']['name'] == 'Alice'
    
    service.unsubscribe()

def test_multiple_subscribers_all_receive(redis_url):
    service1 = NotificationService(redis_url)
    service2 = NotificationService(redis_url)
    
    events1 = threading.Event()
    events2 = threading.Event()
    messages1 = []
    messages2 = []
    
    service1.subscribe('alerts', lambda m: (messages1.append(m), events1.set()))
    service2.subscribe('alerts', lambda m: (messages2.append(m), events2.set()))
    
    import time
    time.sleep(0.1)
    
    service1.publish('alerts', 'system.down', {'service': 'payment'})
    
    assert events1.wait(timeout=2)
    assert events2.wait(timeout=2)
    
    assert messages1[0]['data']['service'] == 'payment'
    assert messages2[0]['data']['service'] == 'payment'
    
    service1.unsubscribe()
    service2.unsubscribe()

def test_unsubscribed_client_misses_messages(redis_url):
    """
    Pub/Sub is ephemeral — messages published before subscription are lost.
    This test documents that behavior.
    """
    service = NotificationService(redis_url)
    received = []
    
    # Publish BEFORE subscribing
    service.publish('ephemeral', 'event', {'data': 'lost'})
    
    # Now subscribe
    service.subscribe('ephemeral', lambda m: received.append(m))
    
    import time
    time.sleep(0.2)  # Wait to confirm nothing arrives
    
    assert len(received) == 0, "Expected no messages — pub/sub is ephemeral"
    service.unsubscribe()

Testing Pattern-Based Subscriptions

def test_pattern_subscribe_matches_multiple_channels(redis_url):
    import redis as redis_lib
    
    pub = redis_lib.Redis.from_url(redis_url, decode_responses=True)
    sub = redis_lib.Redis.from_url(redis_url, decode_responses=True)
    
    received = []
    done = threading.Event()
    
    def listen():
        pubsub = sub.pubsub()
        pubsub.psubscribe('user:*')
        
        for message in pubsub.listen():
            if message['type'] == 'pmessage':
                received.append(message)
                if len(received) >= 2:
                    done.set()
                    break
    
    t = threading.Thread(target=listen, daemon=True)
    t.start()
    
    import time
    time.sleep(0.1)
    
    pub.publish('user:created', 'alice')
    pub.publish('user:updated', 'alice:email')
    pub.publish('order:created', 'should_not_match')
    
    assert done.wait(timeout=2), "Expected 2 messages"
    
    channels = [m['channel'] for m in received]
    assert 'user:created' in channels
    assert 'user:updated' in channels
    assert 'order:created' not in channels

Redis Streams Testing

Streams vs Pub/Sub

Redis Streams persist messages. Consumers can read messages from any point in the stream, replay history, and use consumer groups for at-least-once delivery. This is closer to Kafka than to a traditional message queue.

Basic Stream Producer/Consumer

# order_stream.py
import json
import redis

r = redis.Redis.from_url('redis://localhost', decode_responses=True)

STREAM_KEY = 'orders:stream'
GROUP_NAME = 'order_processors'
CONSUMER_NAME = 'processor-1'

def publish_order(order_id: str, data: dict) -> str:
    """Publish to stream. Returns message ID."""
    return r.xadd(STREAM_KEY, {
        'order_id': order_id,
        'data': json.dumps(data)
    })

def process_orders(batch_size: int = 10, block_ms: int = 1000) -> list:
    """Read and process pending messages from consumer group."""
    try:
        r.xgroup_create(STREAM_KEY, GROUP_NAME, id='0', mkstream=True)
    except redis.exceptions.ResponseError:
        pass  # Group already exists
    
    messages = r.xreadgroup(
        GROUP_NAME,
        CONSUMER_NAME,
        {STREAM_KEY: '>'},  # '>' = new messages only
        count=batch_size,
        block=block_ms
    )
    
    processed = []
    if messages:
        for stream, entries in messages:
            for msg_id, fields in entries:
                order = {
                    'msg_id': msg_id,
                    'order_id': fields['order_id'],
                    'data': json.loads(fields['data'])
                }
                processed.append(order)
                # Acknowledge processing
                r.xack(STREAM_KEY, GROUP_NAME, msg_id)
    
    return processed

Testing Streams with Consumer Groups

# test_order_stream.py
import pytest
import redis as redis_lib
from testcontainers.redis import RedisContainer
import order_stream

@pytest.fixture(scope='module')
def redis_container():
    with RedisContainer('redis:7-alpine') as container:
        yield container

@pytest.fixture
def stream_client(redis_container, monkeypatch):
    client = redis_lib.Redis(
        host=redis_container.get_container_host_ip(),
        port=redis_container.get_exposed_port(6379),
        decode_responses=True
    )
    monkeypatch.setattr(order_stream, 'r', client)
    yield client
    client.delete(order_stream.STREAM_KEY)

def test_published_message_readable_from_stream(stream_client):
    msg_id = order_stream.publish_order('ord-001', {'items': ['a', 'b'], 'total': 99.99})
    
    # Verify message is in the stream
    messages = stream_client.xrange(order_stream.STREAM_KEY, '-', '+')
    assert len(messages) == 1
    assert messages[0][0] == msg_id
    assert messages[0][1]['order_id'] == 'ord-001'

def test_consumer_group_processes_messages(stream_client):
    order_stream.publish_order('ord-001', {'total': 10})
    order_stream.publish_order('ord-002', {'total': 20})
    order_stream.publish_order('ord-003', {'total': 30})
    
    processed = order_stream.process_orders(batch_size=10, block_ms=100)
    
    assert len(processed) == 3
    order_ids = [p['order_id'] for p in processed]
    assert 'ord-001' in order_ids
    assert 'ord-002' in order_ids
    assert 'ord-003' in order_ids

def test_acknowledged_messages_not_redelivered(stream_client):
    order_stream.publish_order('ord-001', {'total': 10})
    
    # Process and acknowledge
    first_batch = order_stream.process_orders(batch_size=10, block_ms=100)
    assert len(first_batch) == 1
    
    # Second read should return no messages (already acknowledged)
    second_batch = order_stream.process_orders(batch_size=10, block_ms=100)
    assert len(second_batch) == 0

def test_pending_messages_redeliverable(stream_client):
    """Messages not acknowledged remain in PEL (Pending Entry List)."""
    r = stream_client
    
    # Publish a message
    msg_id = order_stream.publish_order('ord-001', {'total': 10})
    
    # Read but DON'T acknowledge
    r.xreadgroup(
        order_stream.GROUP_NAME,
        order_stream.CONSUMER_NAME,
        {order_stream.STREAM_KEY: '>'},
        count=1,
        block=100
    )
    
    # Check PEL — should have 1 pending entry
    pending = r.xpending(order_stream.STREAM_KEY, order_stream.GROUP_NAME)
    assert pending['pending'] == 1
    
    # Claim and reprocess (simulating failure recovery)
    pending_detail = r.xpending_range(
        order_stream.STREAM_KEY, 
        order_stream.GROUP_NAME,
        min='-', max='+', count=10
    )
    
    for entry in pending_detail:
        r.xack(order_stream.STREAM_KEY, order_stream.GROUP_NAME, entry['message_id'])
    
    # PEL should be empty now
    pending_after = r.xpending(order_stream.STREAM_KEY, order_stream.GROUP_NAME)
    assert pending_after['pending'] == 0

Testing Stream Trimming and Retention

def test_stream_maxlen_trims_old_messages(stream_client):
    # Publish more than maxlen messages
    for i in range(20):
        stream_client.xadd(
            order_stream.STREAM_KEY,
            {'order_id': f'ord-{i}'},
            maxlen=10,
            approximate=False
        )
    
    # Stream should have at most 10 messages
    length = stream_client.xlen(order_stream.STREAM_KEY)
    assert length <= 10

def test_xrange_reads_historical_messages(stream_client):
    ids = []
    for i in range(5):
        msg_id = stream_client.xadd(
            order_stream.STREAM_KEY,
            {'seq': str(i)}
        )
        ids.append(msg_id)
    
    # Read range between first and last ID
    messages = stream_client.xrange(
        order_stream.STREAM_KEY,
        ids[1],  # Start from second message
        ids[3]   # End at fourth message
    )
    
    assert len(messages) == 3  # msgs 1, 2, 3 (inclusive)
    seqs = [int(m[1]['seq']) for m in messages]
    assert seqs == [1, 2, 3]

Testing Multiple Consumer Groups

Real-world streams often have multiple consumer groups (e.g., an analytics group and a billing group that both need to process every order).

def test_independent_consumer_groups(stream_client):
    ANALYTICS_GROUP = 'analytics'
    BILLING_GROUP = 'billing'
    
    # Create both groups
    try:
        stream_client.xgroup_create(order_stream.STREAM_KEY, ANALYTICS_GROUP, id='0', mkstream=True)
    except redis_lib.exceptions.ResponseError:
        pass
    
    try:
        stream_client.xgroup_create(order_stream.STREAM_KEY, BILLING_GROUP, id='0', mkstream=True)
    except redis_lib.exceptions.ResponseError:
        pass
    
    # Publish one order
    stream_client.xadd(order_stream.STREAM_KEY, {'order_id': 'ord-999'})
    
    # Both groups can read the same message independently
    analytics_msgs = stream_client.xreadgroup(
        ANALYTICS_GROUP, 'analytics-consumer',
        {order_stream.STREAM_KEY: '>'}, count=10, block=100
    )
    
    billing_msgs = stream_client.xreadgroup(
        BILLING_GROUP, 'billing-consumer',
        {order_stream.STREAM_KEY: '>'}, count=10, block=100
    )
    
    assert len(analytics_msgs[0][1]) == 1
    assert len(billing_msgs[0][1]) == 1
    
    analytics_order_id = analytics_msgs[0][1][0][1]['order_id']
    billing_order_id = billing_msgs[0][1][0][1]['order_id']
    
    assert analytics_order_id == billing_order_id == 'ord-999'

Testing with fakeredis2 for Streams

For unit tests that don't need Docker, fakeredis2 supports Streams:

import fakeredis
import pytest

def test_stream_with_fakeredis():
    r = fakeredis.FakeRedis(decode_responses=True)
    
    # Add to stream
    r.xadd('test:stream', {'field': 'value1'})
    r.xadd('test:stream', {'field': 'value2'})
    
    # Create consumer group
    r.xgroup_create('test:stream', 'test-group', id='0')
    
    # Read as consumer group
    messages = r.xreadgroup(
        'test-group', 'consumer-1',
        {'test:stream': '>'}, count=10
    )
    
    assert len(messages[0][1]) == 2
    assert messages[0][1][0][1]['field'] == 'value1'
    
    # Acknowledge
    msg_id = messages[0][1][0][0]
    r.xack('test:stream', 'test-group', msg_id)
    
    # Verify acknowledged
    pending = r.xpending('test:stream', 'test-group')
    assert pending['pending'] == 1  # Only one pending (second not acked)

Common Mistakes

Mistake 1: Not waiting for subscriber thread to connect

After starting a subscribe thread, publish immediately without a small delay. The subscription may not be active yet:

# Bad
service.subscribe('channel', handler)
service.publish('channel', 'event', {})  # May miss

# Good
service.subscribe('channel', handler)
time.sleep(0.05)  # Let subscription establish
service.publish('channel', 'event', {})

Mistake 2: Not cleaning up streams between tests

Streams accumulate messages across tests. Use DELETE on the stream key or use a unique key per test:

@pytest.fixture(autouse=True)
def clean_stream(stream_client):
    stream_client.delete(order_stream.STREAM_KEY)
    yield
    stream_client.delete(order_stream.STREAM_KEY)

Mistake 3: Ignoring XGROUP_CREATE errors

Consumer group creation fails if the group already exists. Always wrap in try/except or use mkstream=True only on first creation.

Conclusion

Redis Pub/Sub tests need synchronization primitives (events, queues) to avoid flaky timing. Streams tests need Testcontainers or fakeredis2 for full consumer group support. The key properties to verify are: message delivery, consumer group isolation, acknowledgment semantics, and pending entry list behavior for failure recovery.