<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>

@SpringBootTest
@SpringRabbitTest
class OrderConsumerSpyTest {

    @Autowired
    private RabbitListenerTestHarness harness;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void shouldInvokeListener() throws Exception {
        OrderConsumer spy = harness.getSpy("orderListener");
        assertThat(spy).isNotNull();

        LatchCountDownAndCallRealMethodAnswer answer =
            harness.getLatchAnswerFor("orderListener", 1);

        rabbitTemplate.convertAndSend("orders.exchange", "orders.created",
            new OrderEvent("123", "CREATED"));

        assertThat(answer.await(10)).isTrue();
        verify(spy).handleOrder(argThat(e -> e.getOrderId().equals("123")));
    }
}

@SpringBootTest
@SpringRabbitTest
class OrderConsumerCaptureTest {

    @Autowired
    private RabbitListenerTestHarness harness;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void shouldCaptureInvocationData() throws Exception {
        rabbitTemplate.convertAndSend("orders.exchange", "orders.created",
            new OrderEvent("456", "PAID"));

        InvocationData data = harness.getNextInvocationDataFor(
            "orderListener", 10, TimeUnit.SECONDS);

        assertThat(data).isNotNull();
        OrderEvent captured = (OrderEvent) data.getArguments()[0];
        assertThat(captured.getOrderId()).isEqualTo("456");
        assertThat(captured.getStatus()).isEqualTo("PAID");
    }
}

@SpringBootTest
@SpringRabbitTest
class OrderServiceReplyTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void shouldReturnOrderResponse() {
        OrderRequest request = new OrderRequest("item-1", 2);

        OrderResponse response = (OrderResponse) rabbitTemplate.convertSendAndReceive(
            "orders.exchange", "orders.create", request);

        assertThat(response).isNotNull();
        assertThat(response.getStatus()).isEqualTo("CREATED");
    }
}

@SpringBootTest
@SpringRabbitTest
class NoBrokerTest {

    @Autowired
    private TestRabbitTemplate testRabbitTemplate;

    @Test
    void shouldSendWithoutBroker() {
        testRabbitTemplate.convertAndSend("orders.exchange", "orders.created",
            new OrderEvent("789", "CREATED"));

        // TestRabbitTemplate routes directly to @RabbitListener methods
        // Verify side effects (database writes, service calls, etc.)
    }
}

@SpringBootTest
@Testcontainers
class RabbitIntegrationTest {

    @Container
    @ServiceConnection
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.13-management");

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Test
    void shouldProduceAndConsumeOrder() {
        rabbitTemplate.convertAndSend("orders.exchange", "orders.created",
            new OrderEvent("123", "CREATED"));

        await().atMost(Duration.ofSeconds(10))
            .untilAsserted(() -> {
                Optional<Order> order = orderRepository.findById("123");
                assertThat(order).isPresent();
                assertThat(order.get().getStatus()).isEqualTo("CREATED");
            });
    }
}

static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.13-management")
    .withExchange("orders.exchange", "direct")
    .withQueue("orders.queue")
    .withBinding("orders.exchange", "orders.queue",
        Map.of(), "orders.created", "queue");

@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);
}

import { RabbitMQContainer } from "@testcontainers/rabbitmq";
import amqp from "amqplib";

describe("RabbitMQ Integration", () => {
  let container: StartedTestContainer;
  let connection: amqp.Connection;

  beforeAll(async () => {
    container = await new RabbitMQContainer("rabbitmq:3.13-management").start();
    connection = await amqp.connect(container.getAmqpUrl());
  }, 60_000);

  afterAll(async () => {
    await connection.close();
    await container.stop();
  });

  it("should produce and consume messages", async () => {
    const channel = await connection.createChannel();
    const queue = "test-queue";
    await channel.assertQueue(queue, { durable: false });

    const message = { orderId: "123", status: "CREATED" };
    channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)));

    const received = await new Promise<any>((resolve) => {
      channel.consume(queue, (msg) => {
        if (msg) resolve(JSON.parse(msg.content.toString()));
      });
    });

    expect(received.orderId).toBe("123");
    await channel.close();
  });
});

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

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

def test_produce_and_consume(rabbitmq):
    params = pika.ConnectionParameters(
        host=rabbitmq.get_container_host_ip(),
        port=rabbitmq.get_exposed_port(5672),
        credentials=pika.PlainCredentials("guest", "guest"),
    )
    connection = pika.BlockingConnection(params)
    channel = connection.channel()
    channel.queue_declare(queue="test-queue")

    message = {"orderId": "123", "status": "CREATED"}
    channel.basic_publish(exchange="", routing_key="test-queue",
                          body=json.dumps(message))

    method, props, body = channel.basic_get(queue="test-queue", auto_ack=True)
    assert method is not None
    assert json.loads(body)["orderId"] == "123"

    connection.close()