Redis Pub/Sub Channel Subscribe Mock in Python

A lightweight in-memory mock of Redis pub/sub that lets you subscribe to channels, publish messages, and verify handler behavior in tests without a real Redis server.

Easy Python 3.9+ Aug 9, 2026 Streaming & messaging 11 views 0 copies

Python code

37 lines
Python 3.9+
class MockRedisPubSub:
    def __init__(self):
        self.channels = {}

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

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

    def unsubscribe(self, channel, handler=None):
        if channel in self.channels:
            if handler is None:
                self.channels[channel] = []
            else:
                self.channels[channel].remove(handler)


if __name__ == "__main__":
    mock = MockRedisPubSub()
    received = []

    mock.subscribe("news").append(lambda msg: received.append(f"news: {msg}"))
    mock.subscribe("sports").append(lambda msg: received.append(f"sports: {msg}"))

    mock.publish("news", "breaking story")
    mock.publish("sports", "game result")
    mock.publish("unknown", "nobody hears this")

    for item in received:
        print(item)

Output

stdout
news: breaking story
sports: game result

How it works

subscribe returns a list of handlers for a channel, allowing direct .append() of callbacks. publish iterates the channel's handlers synchronously and returns the number of receivers, mirroring Redis's return value. unsubscribe removes either all handlers or a specific one. The mock is synchronous, unlike real Redis's async message loop, which keeps test logic simple and deterministic. This avoids needing a live Redis instance in unit tests, cutting setup time and flakiness.

Common mistakes

  • Returning the channel name instead of the handler list in subscribe, breaking chained .append() calls.
  • Publishing to a channel before any subscriber exists, silently dropping the message (Redis drops by design).
  • Forgetting that publish returns the subscriber count, not success/failure.
  • Removing a handler that isn't subscribed, raising ValueError instead of ignoring.

Variations

  1. Use a dict of sets instead of lists to avoid duplicate handler registration.
  2. Add a threading.Event to simulate async delivery with a background loop.

Real-world use cases

  • Unit testing an event-driven microservice that publishes to Redis channels, without running Redis locally or in CI.
  • Verifying webhook dispatchers by subscribing mock handlers before triggering message flow.
  • Prototyping a chat or notification system's pub/sub behavior before wiring up the real Redis adapter.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Streaming & messaging

Related tutorials and quizzes for this topic.