Mock Redis Streams XADD and XREAD in Python

A pure-Python mock of Redis streams that implements basic XADD, XREAD, and XLEN behavior for local testing without a real Redis server.

Medium Python 3.9+ Aug 9, 2026 Streaming & messaging 13 views 0 copies

redis streams mocking testing messaging

Requires third-party packages — install first

pip install redis

Python code

43 lines

Python 3.9+

import redis
import time
import threading


class MockRedisStreams:
    def __init__(self):
        self.streams = {}

    def xadd(self, stream_name, fields):
        if stream_name not in self.streams:
            self.streams[stream_name] = []
        entry_id = f"{time.time_ns()}-{len(self.streams[stream_name])}"
        self.streams[stream_name].append((entry_id, fields))
        return entry_id

    def xread(self, streams, block_ms=0):
        results = []
        if block_ms:
            time.sleep(block_ms / 1000)
        for stream_name, last_id in streams:
            if stream_name not in self.streams:
                continue
            for entry_id, fields in self.streams[stream_name]:
                if entry_id > last_id:
                    results.append((stream_name, [(entry_id, fields)]))
        if not results and str(streams[0][0]) not in self.streams:
            return None
        return results

    def xlen(self, stream_name):
        return len(self.streams.get(stream_name, []))


if __name__ == "__main__":
    mock = MockRedisStreams()
    id1 = mock.xadd("events", {"user": "alice", "action": "login"})
    id2 = mock.xadd("events", {"user": "bob", "action": "purchase"})

    print(f"Added IDs: {id1}, {id2}")
    print(f"Stream length: {mock.xlen('events')}")
    print(f"Read all: {mock.xread([('events', '0-0')])}")
    print(f"Read after first: {mock.xread([('events', id1)])}")

Output

stdout

Added IDs: 1742189373382704000-0, 1742189373382704000-1
Stream length: 2
Read all: [('events', [('1742189373382704000-0', {'user': 'alice', 'action': 'login'}), ('1742189373382704000-1', {'user': 'bob', 'action': 'purchase'})])]
Read after first: [('events', [('1742189373382704000-1', {'user': 'bob', 'action': 'purchase'})])]

How it works

The mock implements a minimal Redis Streams-compatible API by storing entries as a list of tuples in a dictionary keyed by stream name. The xadd method generates unique IDs using nanosecond timestamps plus a sequence number, mimicking Redis's ID format. The xread method filters entries by comparing the last-read ID, with optional blocking behavior simulated via time.sleep. This approach keeps the logic simple while preserving the core semantics of stream reads.

Common mistakes

Forgetting to handle the last_id comparison correctly in xread when a real Redis ID like '0-0' is passed
Not accounting for the block_ms parameter being in milliseconds rather than seconds
Assuming the mock is thread-safe when multiple threads write concurrently
Returning None instead of an empty list when streams exist but have no new entries

Variations

Use fakeredis to get a more complete, battle-tested in-memory Redis mock
Implement the mock as a context manager to support async/await patterns

Real-world use cases

Testing event-driven services locally without spinning up a Redis container in CI pipelines
Writing unit tests for stream consumers that need deterministic behavior and fast execution
Prototyping messaging logic during development before integrating with a production Redis cluster

Mock Redis Streams XADD and XREAD in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes