Testing

RabbitMQ Integration Testing — Local and CI Strategies

HelpMeTest

HelpMeTest

— 8 min read
RabbitMQ Integration Testing — Local and CI Strategies

RabbitMQ integration tests need a real broker to catch routing failures, binding misconfiguration, and dead-letter queue behavior that mocks will never surface. This guide covers testing exchanges, queues, bindings, and DLQs with Testcontainers — for both local development and CI pipelines.

Key Takeaways

Test the routing topology, not just message delivery. A message reaching the right consumer tells you delivery works. Testing with the wrong routing key deliberately tells you your dead-letter configuration works.

Use the management API to inspect broker state. RabbitMQ's HTTP API lets you assert queue depth, binding existence, and message rates directly from tests — no consumer required.

Dead-letter queues need dedicated tests. DLQ routing is configured at queue creation time. A queue that silently drops rejected messages instead of routing to a DLQ is a production incident waiting to happen.

One @Container per test class saves startup cost. RabbitMQ initializes faster than Kafka, but reusing the container across a test class still saves 3–5 seconds per test.

Test nack behavior explicitly. Consumers that nack with requeue=false should route to the DLQ. Consumers that nack with requeue=true re-enqueue. If you never test nack, you don't know which path your code actually takes.

RabbitMQ's flexibility is also its test surface. Unlike Kafka's append-only log model, RabbitMQ routes messages through exchanges via binding rules. A direct exchange, a topic exchange, and a fanout exchange all behave differently. Headers exchanges are their own universe. The routing topology is configuration — and configuration bugs are silent until they're in production.

The goal of integration testing with RabbitMQ is to run your application's channel, exchange, queue, and binding setup against a real broker, publish messages, consume them, and assert on outcomes. Testcontainers makes this self-contained and CI-friendly.

Dependencies

For Java with Spring AMQP:

<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>rabbitmq</artifactId>
  <version>1.19.7</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit-test</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.20.0</version>
</dependency>

For Python with pika:

testcontainers[rabbitmq]==3.7.6
pika==1.3.2
requests==2.31.0
pytest==8.1.0

Container Setup

Testcontainers ships a RabbitMQContainer that wraps the official rabbitmq:management image. The management image exposes port 15672 for the HTTP API — useful for assertions.

@Testcontainers
class RabbitMQIntegrationTest {

    @Container
    static final RabbitMQContainer rabbit = new RabbitMQContainer(
        DockerImageName.parse("rabbitmq:3.13-management")
    ).withExposedPorts(5672, 15672);

    private Connection connection;
    private Channel channel;

    @BeforeEach
    void setUp() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(rabbit.getHost());
        factory.setPort(rabbit.getMappedPort(5672));
        factory.setUsername(rabbit.getAdminUsername());
        factory.setPassword(rabbit.getAdminPassword());
        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    @AfterEach
    void tearDown() throws Exception {
        channel.close();
        connection.close();
    }
}

Testing Exchange and Queue Binding

A direct exchange routes messages to queues whose binding key exactly matches the routing key. Let's test that messages go where they're supposed to:

@Test
void directExchangeRoutesMessageToCorrectQueue() throws Exception {
    String exchange = "orders";
    String routingKey = "order.created";
    String queue = "order-processing";

    // Declare topology
    channel.exchangeDeclare(exchange, "direct", true);
    channel.queueDeclare(queue, true, false, false, null);
    channel.queueBind(queue, exchange, routingKey);

    // Publish
    String message = "{\"orderId\":\"42\",\"status\":\"CREATED\"}";
    channel.basicPublish(exchange, routingKey, null, message.getBytes());

    // Consume and assert
    List<String> received = new ArrayList<>();
    channel.basicConsume(queue, true, (tag, delivery) ->
        received.add(new String(delivery.getBody())),
        tag -> {}
    );

    await().atMost(5, SECONDS).until(() -> !received.isEmpty());
    assertThat(received).hasSize(1);
    assertThat(received.get(0)).contains("order-42");
}

Now test the negative case — a message with the wrong routing key should not reach the queue:

@Test
void directExchangeDoesNotRouteToUnmatchedQueue() throws Exception {
    String exchange = "orders";
    channel.exchangeDeclare(exchange, "direct", true);
    channel.queueDeclare("order-processing", true, false, false, null);
    channel.queueBind("order-processing", exchange, "order.created");

    // Publish with wrong routing key
    channel.basicPublish(exchange, "order.deleted", null,
        "{\"orderId\":\"99\"}".getBytes());

    // Queue should remain empty
    Thread.sleep(500); // Brief wait — we're asserting absence, not presence
    long messageCount = channel.messageCount("order-processing");
    assertThat(messageCount).isZero();
}

For testing absence, a short sleep is acceptable. You cannot poll for the absence of something — the absence is the assertion after a reasonable wait.

Topic Exchange Testing

Topic exchanges route by pattern matching. order.* matches order.created and order.deleted. order.# matches order.created.us and order.created.eu.west. Bugs here are easy to introduce and hard to spot in code review.

@Test
void topicExchangeRoutesWildcardPatterns() throws Exception {
    String exchange = "events";
    channel.exchangeDeclare(exchange, "topic", true);

    // Two queues with different binding patterns
    channel.queueDeclare("us-orders", true, false, false, null);
    channel.queueBind("us-orders", exchange, "order.*.us");

    channel.queueDeclare("all-orders", true, false, false, null);
    channel.queueBind("all-orders", exchange, "order.#");

    // Publish
    channel.basicPublish(exchange, "order.created.us", null,
        "msg1".getBytes());
    channel.basicPublish(exchange, "order.created.eu", null,
        "msg2".getBytes());

    List<String> usMessages = consumeAll("us-orders", 2000);
    List<String> allMessages = consumeAll("all-orders", 2000);

    // "order.created.us" matches both patterns
    // "order.created.eu" matches only "order.#"
    assertThat(usMessages).hasSize(1);
    assertThat(allMessages).hasSize(2);
}

private List<String> consumeAll(String queue, long waitMillis) throws Exception {
    List<String> messages = new ArrayList<>();
    channel.basicConsume(queue, true,
        (tag, delivery) -> messages.add(new String(delivery.getBody())),
        tag -> {});
    Thread.sleep(waitMillis);
    return messages;
}

Dead-Letter Queue Testing

Dead-letter queues are the part of RabbitMQ that most teams configure once and never test. The DLQ is configured via queue arguments at declaration time. If you get those arguments wrong, rejected messages disappear silently.

@Test
void rejectedMessageRoutesToDeadLetterQueue() throws Exception {
    String dlxExchange = "orders.dlx";
    String dlq = "orders.dead";
    String mainQueue = "orders.main";
    String exchange = "orders";

    // Set up DLX infrastructure
    channel.exchangeDeclare(dlxExchange, "fanout", true);
    channel.queueDeclare(dlq, true, false, false, null);
    channel.queueBind(dlq, dlxExchange, "");

    // Declare main queue with DLX configured
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", dlxExchange);
    channel.queueDeclare(mainQueue, true, false, false, args);
    channel.exchangeDeclare(exchange, "direct", true);
    channel.queueBind(mainQueue, exchange, "order");

    // Publish a message
    channel.basicPublish(exchange, "order", null,
        "{\"orderId\":\"bad-order\"}".getBytes());

    // Consumer nacks with requeue=false (simulating processing failure)
    channel.basicConsume(mainQueue, false, (tag, delivery) -> {
        // Simulate a processing error
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(),
            false, false); // requeue=false → goes to DLX
    }, tag -> {});

    // Assert message arrives in DLQ
    List<String> deadMessages = new ArrayList<>();
    channel.basicConsume(dlq, true,
        (tag, delivery) -> deadMessages.add(new String(delivery.getBody())),
        tag -> {});

    await().atMost(5, SECONDS).until(() -> !deadMessages.isEmpty());
    assertThat(deadMessages).hasSize(1);
    assertThat(deadMessages.get(0)).contains("bad-order");
}

This test would fail if the DLX argument were missing or spelled incorrectly. Without it, basicNack with requeue=false drops the message entirely.

Using the Management API for State Assertions

RabbitMQ's HTTP management API is a powerful assertion tool that most developers overlook. You can query queue depth, check binding existence, and inspect exchange configuration — without consuming any messages.

private int getQueueDepth(String vhost, String queue) throws Exception {
    String url = "http://" + rabbit.getHost() + ":" +
        rabbit.getMappedPort(15672) +
        "/api/queues/" + URLEncoder.encode(vhost, "UTF-8") +
        "/" + URLEncoder.encode(queue, "UTF-8");

    HttpClient client = HttpClient.newHttpClient();
    HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create(url))
        .header("Authorization", basicAuth(
            rabbit.getAdminUsername(),
            rabbit.getAdminPassword()))
        .GET()
        .build();

    HttpResponse<String> response =
        client.send(request, HttpResponse.BodyHandlers.ofString());

    JsonNode node = new ObjectMapper().readTree(response.body());
    return node.get("messages").asInt();
}

Use this to assert that a consumer has drained a queue, or that published messages are accumulating as expected before a consumer starts:

@Test
void publisherEnqueuesMessagesBeforeConsumerStarts() throws Exception {
    channel.queueDeclare("batch-jobs", true, false, false, null);

    for (int i = 0; i < 10; i++) {
        channel.basicPublish("", "batch-jobs", null,
            ("job-" + i).getBytes());
    }

    // Verify queue depth via management API
    await().atMost(3, SECONDS).untilAsserted(() ->
        assertThat(getQueueDepth("/", "batch-jobs")).isEqualTo(10)
    );
}

Python Integration Tests

Using pika with pytest and the same Testcontainers setup:

import pytest
import pika
import json
import time
from testcontainers.rabbitmq import RabbitMqContainer

@pytest.fixture(scope="module")
def rabbit():
    with RabbitMqContainer("rabbitmq:3.13-management") as container:
        yield container

@pytest.fixture
def channel(rabbit):
    credentials = pika.PlainCredentials(
        rabbit.RABBITMQ_DEFAULT_USER,
        rabbit.RABBITMQ_DEFAULT_PASS
    )
    params = pika.ConnectionParameters(
        host=rabbit.get_container_host_ip(),
        port=rabbit.get_exposed_port(5672),
        credentials=credentials
    )
    connection = pika.BlockingConnection(params)
    ch = connection.channel()
    yield ch
    ch.close()
    connection.close()

def test_fanout_exchange_broadcasts_to_all_queues(channel):
    channel.exchange_declare("notifications", exchange_type="fanout")
    channel.queue_declare("email-notifs")
    channel.queue_declare("sms-notifs")
    channel.queue_bind("email-notifs", "notifications", "")
    channel.queue_bind("sms-notifs", "notifications", "")

    channel.basic_publish(
        exchange="notifications",
        routing_key="",
        body=json.dumps({"type": "alert", "text": "Server down"})
    )

    time.sleep(0.5)

    _, _, email_body = channel.basic_get("email-notifs", auto_ack=True)
    _, _, sms_body = channel.basic_get("sms-notifs", auto_ack=True)

    assert email_body is not None
    assert sms_body is not None
    assert json.loads(email_body)["type"] == "alert"

CI Configuration

RabbitMQ starts faster than Kafka, so the CI overhead is lower. Still, warm the image before tests:

name: RabbitMQ Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with:
          java-version: '21'
          distribution: 'temurin'
      - name: Pull RabbitMQ image
        run: docker pull rabbitmq:3.13-management
      - name: Run tests
        run: mvn test -pl :rabbitmq-integration-tests

If you prefer running a real RabbitMQ service rather than Testcontainers in CI (some teams find this simpler for GitLab pipelines), use the service container:

rabbitmq-tests:
  image: maven:3.9-eclipse-temurin-21
  services:
    - name: rabbitmq:3.13-management
      alias: rabbitmq
  variables:
    RABBITMQ_HOST: rabbitmq
    RABBITMQ_PORT: "5672"
  script:
    - mvn test -Drabbitmq.host=$RABBITMQ_HOST -Drabbitmq.port=$RABBITMQ_PORT

The trade-off: service containers run for the full job lifetime and don't support the dynamic port mapping that Testcontainers provides. If you have multiple jobs, each gets its own RabbitMQ instance, which is actually a benefit for isolation.

Testing With Spring AMQP

Spring Boot apps using spring-boot-starter-amqp can use @SpringBootTest with a dynamic bootstrap configuration:

@SpringBootTest
@Testcontainers
class OrderListenerIntegrationTest {

    @Container
    static final RabbitMQContainer rabbit =
        new RabbitMQContainer(DockerImageName.parse("rabbitmq:3.13-management"));

    @DynamicPropertySource
    static void rabbitProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbit::getHost);
        registry.add("spring.rabbitmq.port", rabbit::getAmqpPort);
        registry.add("spring.rabbitmq.username", rabbit::getAdminUsername);
        registry.add("spring.rabbitmq.password", rabbit::getAdminPassword);
    }

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    OrderRepository orderRepository;

    @Test
    void listenerPersistsOrderOnMessage() {
        rabbitTemplate.convertAndSend("orders.exchange", "order.created",
            new OrderCreatedEvent("order-789", 150.00));

        await().atMost(10, SECONDS).untilAsserted(() ->
            assertThat(orderRepository.findById("order-789")).isPresent()
        );
    }
}

@DynamicPropertySource injects the container's host and port into the Spring application context before the context starts. This gives you a fully wired Spring application talking to a real RabbitMQ — no mocking, no in-memory stubs.

Common Pitfalls

Vhost isolation. By default, all queues and exchanges share the / vhost. In tests, this is fine. In production, different services often use separate vhosts for isolation. If your tests use / but production uses /payments, you may have untested vhost-level permission issues. Add a vhost to your container setup to match production.

Message TTL and queue TTL. If your queues have x-message-ttl or x-expires configured, messages can expire before your test consumer reads them. Always set timeouts shorter than your test TTLs.

Connection reset between tests. Creating a new connection per test is slow but guarantees clean channel state. Reusing a connection across tests is faster but requires careful cleanup — especially if a test leaves unacked messages or unconsumed queues.

Prefetch count. Default channel prefetch is unlimited. In production, your consumers likely set basic.qos to limit in-flight messages. Test with the same prefetch count to catch backpressure behavior.

RabbitMQ integration tests pay their startup cost once and give you confidence in the entire routing path — from exchange declaration through binding rules to dead-letter handling. That's coverage that no unit test of your consumer logic can provide.

Read more

Start now free