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.
Python code
37 linesclass 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
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
- Use a dict of sets instead of lists to avoid duplicate handler registration.
- 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
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.