martinholovsky/claude-skills-generator/rabbitmq-expert

# tests/test_message_queue.py
import pytest
import pika
import json
import time
from unittest.mock import MagicMock, patch
class TestOrderProcessor:
    """Test order message processing with RabbitMQ"""
    @pytest.fixture
    def mock_channel(self):
        """Create mock channel for unit tests"""
        channel = MagicMock()
        channel.basic_qos = MagicMock()
        channel.basic_consume = MagicMock()
        channel.basic_ack = MagicMock()
        channel.basic_nack = MagicMock()
        return channel
    @pytest.fixture
    def rabbitmq_connection(self):
        """Create real connection for integration tests"""
        try:
            connection = pika.BlockingConnection(
                pika.ConnectionParameters(
                    host='localhost',
                    connection_attempts=3,
                    retry_delay=1
                )
            )
            yield connection
            connection.close()
        except pika.exceptions.AMQPConnectionError:
            pytest.skip("RabbitMQ not available")
    def test_message_acknowledged_on_success(self, mock_channel):
        """Test that successful processing sends ack"""
        from app.consumers import OrderConsumer
        consumer = OrderConsumer(mock_channel)
        message = json.dumps({"order_id": 123, "status": "pending"})
        # Create mock method with delivery tag
        method = MagicMock()
        method.delivery_tag = 1
        # Process message
        consumer.process_message(mock_channel, method, None, message.encode())
        # Verify ack was called
        mock_channel.basic_ack.assert_called_once_with(delivery_tag=1)
        mock_channel.basic_nack.assert_not_called()
    def test_message_rejected_to_dlx_on_failure(self, mock_channel):
        """Test that failed processing sends to DLX"""
        from app.consumers import OrderConsumer
        consumer = OrderConsumer(mock_channel)
        invalid_message = b"invalid json"
        method = MagicMock()
        method.delivery_tag = 2
        # Process invalid message
        consumer.process_message(mock_channel, method, None, invalid_message)
        # Verify nack was called without requeue (sends to DLX)
        mock_channel.basic_nack.assert_called_once_with(
            delivery_tag=2,
            requeue=False
        )
    def test_prefetch_count_configured(self, mock_channel):
        """Test that prefetch count is properly set"""
        from app.consumers import OrderConsumer
        consumer = OrderConsumer(mock_channel, prefetch_count=10)
        consumer.setup()
        mock_channel.basic_qos.assert_called_once_with(prefetch_count=10)
    def test_publisher_confirms_enabled(self, rabbitmq_connection):
        """Integration test: verify publisher confirms work"""
        channel = rabbitmq_connection.channel()
        channel.confirm_delivery()
        # Declare test queue
        channel.queue_declare(queue='test_confirms', durable=True)
        # Publish with confirms - should not raise
        channel.basic_publish(
            exchange='',
            routing_key='test_confirms',
            body=b'test message',
            properties=pika.BasicProperties(delivery_mode=2)
        )
        # Cleanup
        channel.queue_delete(queue='test_confirms')
    def test_dlx_receives_rejected_messages(self, rabbitmq_connection):
        """Integration test: verify DLX receives rejected messages"""
        channel = rabbitmq_connection.channel()
        # Setup DLX
        channel.exchange_declare(exchange='test_dlx', exchange_type='fanout')
        channel.queue_declare(queue='test_dead_letters')
        channel.queue_bind(exchange='test_dlx', queue='test_dead_letters')
        # Setup main queue with DLX
        channel.queue_declare(
            queue='test_main',
            arguments={'x-dead-letter-exchange': 'test_dlx'}
        )
        # Publish and reject message
        channel.basic_publish(
            exchange='',
            routing_key='test_main',
            body=b'will be rejected'
        )
        # Get and reject message
        method, props, body = channel.basic_get('test_main')
        if method:
            channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        # Wait for DLX delivery
        time.sleep(0.1)
        # Verify message arrived in DLX queue
        method, props, body = channel.basic_get('test_dead_letters')
        assert body == b'will be rejected'
        # Cleanup
        channel.queue_delete(queue='test_main')
        channel.queue_delete(queue='test_dead_letters')
        channel.exchange_delete(exchange='test_dlx')

# app/consumers.py
import json
import logging
logger = logging.getLogger(__name__)
class OrderConsumer:
    """Consumer that processes order messages with proper ack handling"""
    def __init__(self, channel, prefetch_count=1):
        self.channel = channel
        self.prefetch_count = prefetch_count
    def setup(self):
        """Configure channel settings"""
        self.channel.basic_qos(prefetch_count=self.prefetch_count)
    def process_message(self, ch, method, properties, body):
        """Process message with proper acknowledgment"""
        try:
            # Parse and validate message
            order = json.loads(body)
            # Process the order
            self._handle_order(order)
            # Acknowledge success
            ch.basic_ack(delivery_tag=method.delivery_tag)
            logger.info(f"Processed order: {order.get('order_id')}")
        except json.JSONDecodeError as e:
            logger.error(f"Invalid JSON: {e}")
            # Send to DLX, don't requeue
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        except Exception as e:
            logger.error(f"Processing failed: {e}")
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    def _handle_order(self, order):
        """Business logic for order processing"""
        # Implementation here
        pass

# Run unit tests
pytest tests/test_message_queue.py -v
 
# Run with coverage
pytest tests/ --cov=app --cov-report=term-missing
 
# Run integration tests (requires RabbitMQ)
pytest tests/ -m integration -v
 
# Verify message flow end-to-end
python -m pytest tests/e2e/ -v

# BAD: Unlimited prefetch - consumer gets overwhelmed
channel.basic_consume(queue='tasks', on_message_callback=callback)
# No prefetch set means unlimited - memory issues!
# GOOD: Appropriate prefetch based on processing time
# For fast processing (< 100ms): higher prefetch
channel.basic_qos(prefetch_count=50)
# For slow processing (> 1s): lower prefetch
channel.basic_qos(prefetch_count=1)
# For balanced workloads
channel.basic_qos(prefetch_count=10)

# BAD: Publishing one message at a time with confirms
for order in orders:
    channel.basic_publish(
        exchange='orders',
        routing_key='order.created',
        body=json.dumps(order),
        properties=pika.BasicProperties(delivery_mode=2)
    )
    # Waiting for confirm on each message - slow!
# GOOD: Batch publishing with bulk confirms
channel.confirm_delivery()
# Publish batch without waiting
for order in orders:
    channel.basic_publish(
        exchange='orders',
        routing_key='order.created',
        body=json.dumps(order),
        properties=pika.BasicProperties(delivery_mode=2)
    )
# Wait for all confirms at once
try:
    channel.get_waiting_message_count()  # Forces confirm flush
except pika.exceptions.NackError as e:
    # Handle rejected messages
    logger.error(f"Messages rejected: {e.messages}")

# BAD: Creating new connection for each operation
def send_message(message):
    connection = pika.BlockingConnection(params)  # Expensive!
    channel = connection.channel()
    channel.basic_publish(...)
    connection.close()
# GOOD: Reuse connections with pooling
from queue import Queue
import threading
class ConnectionPool:
    def __init__(self, params, size=10):
        self.pool = Queue(maxsize=size)
        self.params = params
        for _ in range(size):
            conn = pika.BlockingConnection(params)
            self.pool.put(conn)
    def get_connection(self):
        return self.pool.get()
    def return_connection(self, conn):
        if conn.is_open:
            self.pool.put(conn)
        else:
            # Replace dead connection
            self.pool.put(pika.BlockingConnection(self.params))
    def publish(self, exchange, routing_key, body):
        conn = self.get_connection()
        try:
            channel = conn.channel()
            channel.basic_publish(
                exchange=exchange,
                routing_key=routing_key,
                body=body,
                properties=pika.BasicProperties(delivery_mode=2)
            )
        finally:
            self.return_connection(conn)

# BAD: Classic queue with large backlog - memory pressure
channel.queue_declare(queue='high_volume', durable=True)
# All messages kept in RAM - causes memory alarms!
# GOOD: Lazy queue moves messages to disk
channel.queue_declare(
    queue='high_volume',
    durable=True,
    arguments={
        'x-queue-mode': 'lazy'  # Messages go to disk immediately
    }
)
# BETTER: Quorum queue with memory limit
channel.queue_declare(
    queue='high_volume',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        'x-max-in-memory-length': 1000  # Only 1000 msgs in RAM
    }
)

# BAD: Synchronous confirms - blocking on each message
channel.confirm_delivery()
for msg in messages:
    try:
        channel.basic_publish(...)  # Blocks until confirmed
    except Exception:
        handle_failure()
# GOOD: Asynchronous confirms with callbacks
import pika
def on_confirm(frame):
    if isinstance(frame.method, pika.spec.Basic.Ack):
        logger.debug(f"Message {frame.method.delivery_tag} confirmed")
    else:
        logger.error(f"Message {frame.method.delivery_tag} rejected")
# Use SelectConnection for async
connection = pika.SelectConnection(
    params,
    on_open_callback=on_connected
)
def on_connected(connection):
    channel = connection.channel(on_open_callback=on_channel_open)