How to Mock Azure Service Bus Queue in Python

A lightweight in-memory mock of the Azure Service Bus queue API for local testing without cloud dependencies.

Easy Python 3.9+ Aug 9, 2026 Cloud + Python 14 views 0 copies

azure service-bus mock queue testing

Python code

82 lines

Python 3.9+

import json
import time
from collections import deque

class ServiceBusQueueMock:
    def __init__(self, queue_name):
        self.queue_name = queue_name
        self._messages = deque()
        self._dead_letter_queue = deque()
        self._message_counter = 0

    def send_message(self, body, message_id=None, properties=None):
        self._message_counter += 1
        message = {
            "message_id": message_id or f"msg-{self._message_counter}",
            "body": body,
            "properties": properties or {},
            "enqueued_time": time.time(),
            "delivery_count": 0,
            "locked_until": None
        }
        self._messages.appendleft(message)
        return message["message_id"]

    def peek_message(self):
        if self._messages:
            return dict(self._messages[-1])
        return None

    def receive_message(self, lock_timeout=30):
        if self._messages:
            message = self._messages.pop()
            message["delivery_count"] += 1
            message["locked_until"] = time.time() + lock_timeout
            return message
        return None

    def complete_message(self, message_id):
        # Messages already removed on receive; this confirms processing
        print(f"Completing message: {message_id}")

    def abandon_message(self, message_id, reason="abandoned"):
        # For demonstration, simulate re-queuing
        print(f"Abandoning message {message_id}: {reason}")

    def dead_letter_message(self, message_id, reason="dead-lettered"):
        message = self.receive_message()
        if message and message["message_id"] == message_id:
            message["dead_letter_reason"] = reason
            self._dead_letter_queue.append(message)
            print(f"Dead-lettered: {message_id} ({reason})")
        return None

    def get_queue_length(self):
        return len(self._messages)

    def get_dead_letter_length(self):
        return len(self._dead_letter_queue)


if __name__ == "__main__":
    queue = ServiceBusQueueMock("orders")

    # Send messages
    msg1 = queue.send_message({"order_id": 1, "item": "laptop"})
    msg2 = queue.send_message({"order_id": 2, "item": "mouse"}, properties={"priority": "high"})

    # Peek (without removing)
    peeked = queue.peek_message()
    print(f"Peeked message: {json.dumps(peeked['body'])}")

    # Receive and complete
    received = queue.receive_message()
    print(f"Received: {json.dumps(received['body'])}")
    queue.complete_message(received["message_id"])

    # Dead-letter
    queue.dead_letter_message(msg2, reason="invalid payload")

    # Check counters
    print(f"Queue length: {queue.get_queue_length()}")
    print(f"Dead-letter length: {queue.get_dead_letter_length()}")

Output

stdout

Peeked message: {"order_id": 2, "item": "mouse"}
Received: {"order_id": 2, "item": "mouse"}
Completing message: msg-2
Dead-lettered: msg-2 (invalid payload)
Queue length: 1
Dead-letter length: 1

How it works

The mock uses deque to simulate the FIFO queue semantics of Azure Service Bus; appendleft and pop preserve order. peek_message returns a shallow copy so callers can inspect without mutating state, while receive_message pops and increments the delivery count. The dead-letter queue is a separate deque that holds poisoned messages for later inspection. This lets you test send/receive/complete/dead-letter flows without network calls or an Azure subscription.

Common mistakes

Popping from the wrong deque end—use `appendleft` and `pop` for FIFO order.
Returning the internal dict directly; callers could mutate the queue state.
Forgetting to increment `delivery_count` on receive, which changes dead-letter behavior.
Not simulating lock expiry—the mock only tracks `locked_until`, it never enforces it.

Variations

Use `unittest.mock` to patch the real `azure.servicebus` client in unit tests.
Wrap the mock with a context manager for automatic cleanup of pending messages.

Real-world use cases

Running integration tests for order processing pipelines in CI without hitting Azure.
Developing an event-driven microservice locally with zero network dependencies.
Simulating dead-letter flows to test error-handling logic before deployment to Azure.

How to Mock Azure Service Bus Queue in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Cloud + Python

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Cloud + Python

Keep learning

Tutorials

Quizzes