Reference library

Streaming & messaging

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

6 matches

At Most Once Fire-and-Forget Mock in Python

A Python mock that enforces send() is called at most once and records the arguments for verification.

class FireForgetMock:
    def __init__(self):
        self._calls = 0
        self._last_args = None
        self._last_kwargs = None

    def send(self, *args, **kwargs):
        if self._calls > 0:
            raise RuntimeError("send() called more than once")
        self._calls += 1
        self._last_args = args
…

16 0 Open

Streaming & messaging easy

How to Build a Mock Change Data Capture Event Stream in Python

Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.

cdc mock event-stream

Python

from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta

seed(42)  # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)

def mock_cdc_event(stream_index: int) -> dict:
…

12 0 Open

Streaming & messaging easy

How to Mock MQTT Topic Subscriptions with QoS in Python

Build a lightweight MQTT client mock that tracks topic subscriptions with QoS levels and simulates wildcard message delivery.

mqtt mock qos

Python

import time
from collections import defaultdict

class MockMQTTClient:
    def __init__(self):
        self.subscriptions = defaultdict(list)
        self.messages = []
    
    def subscribe(self, topic, qos=0):
        self.subscriptions[topic].append(qos)
        print(f"Subscribed to '{topic}' with QoS {qos}")
   …

14 0 Open

Streaming & messaging easy

How to Mock RabbitMQ Ack Nack Requeue in Python

A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.

rabbitmq testing mock

Python

import json
from collections import deque


class MockChannel:
    def __init__(self):
        self.acked = []
        self.nacked = []
        self.requeued = []

    def basic_ack(self, delivery_tag):
        self.acked.append(delivery_tag)

    def basic_nack(self, delivery_tag, requeue=False):
        self.nacked.…

15 0 Open

Streaming & messaging easy

How to Simulate a Micro-Batch Interval Trigger in Python

A dataclass-based mock that emits batch numbers at fixed intervals, mimicking a micro-batch streaming scheduler for testing and development.

streaming mock dataclass

Python

import time
from dataclasses import dataclass, field
from typing import List, Callable


@dataclass
class MicroBatchTriggerMock:
    batch_interval_seconds: float = 0.5
    max_batches: int = 5
    _batches_emitted: int = 0
    _next_emit_time: float = field(init=False, default=0)

    def start(self, on_batch: Callab…

13 0 Open

Streaming & messaging easy

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.

redis pubsub testing

Python

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:
            …

11 0 Open

Browse by section

Each section groups closely related Python snippets.