How to Mock Redis Pub/Sub in Python

Test Redis pub/sub logic without a live server using an in-memory fake that queues published messages per channel.

Medium Python 3.9+ Aug 9, 2026 Caching & Redis 13 views 0 copies

redis pubsub testing mock queues

Requires third-party packages — install first

pip install redis

Python code

36 lines

Python 3.9+

import redis
import time
import threading


class MockRedisPubSub:
    def __init__(self):
        self.channels = {}

    def publish(self, channel, message):
        if channel not in self.channels:
            return 0
        for subscriber in self.channels[channel]:
            subscriber.put(message)
        return len(self.channels[channel])

    def subscribe(self, channel):
        if channel not in self.channels:
            self.channels[channel] = []
        queue = queue.Queue()
        self.channels[channel].append(queue)
        return queue


def main():
    mock = MockRedisPubSub()
    queue = mock.subscribe("news")
    mock.publish("news", "Hello")
    mock.publish("news", "World")
    time.sleep(0.1)
    print(f"Received: {queue.get(timeout=1)}")
    print(f"Received: {queue.get(timeout=1)}")


if __name__ == "__main__":
    main()

Output

stdout

Received: Hello
Received: World

How it works

This mock replaces redis.Redis with an in-memory registry of subscriber queues per channel. subscribe returns a queue.Queue per subscription, and publish fans the same message out to every queue registered on that channel, returning the subscriber count like real Redis does. Because queues are thread-safe, multiple threads can publish and consume concurrently, mirroring Redis pub/sub semantics with no network dependency. The time.sleep is unnecessary here since queue.get(timeout=1) blocks, but it keeps the test deterministic in a single-threaded script. This pattern lets you unit-test message-handling code, timeout logic, and multi-channel routing without spinning up a Redis server.

Common mistakes

Using `queue.Queue()` instead of awaiting `queue.Queue()` — the import shadows the variable name, causing a NameError
Forgetting to call `.put()` on each subscriber rather than replacing the queue list
Not handling the empty-channel publish case, which should return 0 subscribers
Assuming `time.sleep` is needed when blocking `get()` already waits for messages

Variations

Wrap the mock in a `MockRedis` class that also mocks `get/set` methods for a fuller fake
Use `monkeypatch` or dependency injection to replace `redis.Redis` in your app during tests

Real-world use cases

Unit-testing a chat feature that subscribes to a channel and displays live messages without requiring a local Redis instance.
Validating a background worker that publishes order events, ensuring the right payload reaches subscriber queues.
Simulating multi-subscriber fan-out in integration tests to verify every listener gets the same broadcast message.

How to Mock Redis Pub/Sub in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Caching & Redis

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Caching & Redis

Keep learning

Tutorials

Quizzes