Reference library

Streaming & messaging

Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.

4 matches

How to Mock NATS Subject Hierarchies with Wildcards in Python

Build a lightweight NATS-style pub/sub mock that matches subject hierarchies with '*' and '>' wildcards for tests or prototypes.

nats pubsub wildcards

Python

# Mock a simplified NATS subject hierarchy with wildcard matching
# Supports: exact match, '*' (single token), '>' (tail wildcard)

class NATSSubjectMock:
    def __init__(self):
        self.subscriptions = {}  # subject -> list of callbacks

    def subscribe(self, subject, callback):
        self.subscriptions.setd…

13 0 Open

Streaming & messaging medium

How to Mock Offset Commit Auto vs Manual in Python

Demonstrates a Kafka-style offset commit function with auto/manual modes and tests it using unittest.mock.patch.

unittest mocking kafka

Python

from unittest.mock import Mock, patch

def commit_offsets(topic_partition_offsets, auto_commit=False):
    """Manually commit offsets or simulate auto-commit."""
    if auto_commit:
        print(f"Auto-committing offsets: {topic_partition_offsets}")
        return {"status": "auto_committed"}
    
    print(f"Manuall…

15 0 Open

Streaming & messaging medium

How to Mock a Kafka Rebalance Listener in Python

Simulate Kafka consumer rebalance callbacks (on_partitions_revoked and on_partitions_assigned) with a mock consumer to test listener logic.

kafka rebalance mocking

Python

import time
from collections import defaultdict


class MockKafkaConsumer:
    def __init__(self):
        self.assignments = defaultdict(list)
        self.rebalances = 0

    def assign(self, partitions):
        self.rebalances += 1
        self.assignments.clear()
        for partition in partitions:
            s…

15 0 Open

Streaming & messaging medium

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.

redis streams mocking

Python

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])}"
…

13 0 Open

Browse by section

Each section groups closely related Python snippets.