Testing Message Queue Code: Testcontainers, the MassTransit Test Harness and Contract Tests

Version note: The code in this post was built and run with .NET 10, xUnit v3 4.0.2 (Microsoft.Testing.Platform), Testcontainers.RabbitMq 4.16.0, RabbitMQ 4.4, RabbitMQ.Client 7.2.2 and MassTransit 8.5.12.

Guess First

An integration test publishes OrderPlaced to a real RabbitMQ and, as soon as BasicPublishAsync returns, asserts on the next line that the invoice has been created. Does the test pass? If it does, does it pass every time on the CI (Continuous Integration) server too? Does putting await Task.Delay(2000) in between solve the problem, or does it just make it slower and rarer?

The difficulty of testing messaging code comes down to two places: when business logic is buried in broker code, you need a broker to test it, and once the broker is involved, the test has to answer "when is it done?" in an asynchronous world. In this post we separate the layers: unit tests without a broker, integration tests against a real RabbitMQ with Testcontainers, MassTransit's test harness, and separate tests for message contracts. The integration test infrastructure on the ASP.NET Core side is in the "Integration Test" post of the Web API (Application Programming Interface) series; the end-to-end test strategy is in the upcoming Integration and E2E (End-to-End) Testing series.

Separate the Logic First: Testing the Handler Without a Broker

Testability starts in the shape of the code, before you write a single test. Billing's logic lives in a service that knows nothing about the broker:

public sealed class InvoiceService(IInvoiceStore store)
{
    private static readonly HashSet<string> SupportedCurrencies = ["EUR", "USD", "TRY"];

    public async Task<Invoice?> CreateInvoiceAsync(OrderPlaced order, CancellationToken ct = default)
    {
        if (!SupportedCurrencies.Contains(order.Currency))
            throw new PermanentMessageException($"Unsupported currency {order.Currency}");
        if (await store.ExistsForOrderAsync(order.OrderId, ct))
            return null;

        decimal amount = order.Items.Sum(i => i.Quantity * i.UnitPrice);
        var invoice = new Invoice(Guid.CreateVersion7(), order.OrderId, amount, order.Currency);
        await store.SaveAsync(invoice, ct);
        return invoice;
    }
}

// thin broker-bound adapter: the logic lives in InvoiceService, only the ack/reject decision is here
public sealed class OrderPlacedRabbitHandler(InvoiceService service)
{
    public async Task HandleAsync(IChannel channel, BasicDeliverEventArgs ea, CancellationToken ct = default)
    {
        try
        {
            OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, JsonSerializerOptions.Web)
                ?? throw new PermanentMessageException("Empty message body");
            await service.CreateInvoiceAsync(order, ct);
            await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, ct);
        }
        catch (Exception ex) when (ex is PermanentMessageException or JsonException)
        {
            await channel.BasicRejectAsync(ea.DeliveryTag, requeue: false, ct);
        }
    }
}

The MassTransit consumer uses the same InvoiceService; two adapters, one piece of logic. The unit tests never see the broker:

public sealed class InvoiceServiceTests
{
    private readonly InMemoryInvoiceStore _store = new();

    [Fact]
    public async Task Second_delivery_of_same_order_does_not_create_invoice()
    {
        OrderPlaced order = Orders.Sample();
        var service = new InvoiceService(_store);

        await service.CreateInvoiceAsync(order, TestContext.Current.CancellationToken);
        Invoice? second = await service.CreateInvoiceAsync(order, TestContext.Current.CancellationToken);

        Assert.Null(second);
    }

    [Fact]
    public async Task Unsupported_currency_is_a_permanent_failure()
    {
        var service = new InvoiceService(_store);

        await Assert.ThrowsAsync<PermanentMessageException>(() =>
            service.CreateInvoiceAsync(Orders.Sample("XXX"), TestContext.Current.CancellationToken));
    }
}

These tests finish in milliseconds and lock in two critical decisions from earlier posts in the series: a second delivery of the same order does not issue an invoice (the tenth post), and an unsupported currency is a permanent failure that is not retried (the eighth post). TestContext.Current.CancellationToken is xUnit v3's per-test cancellation token; xUnit's analyzer recommends passing it to calls that take a token. The fake store is deliberately a simple class; no mocking library needed:

public sealed class InMemoryInvoiceStore : IInvoiceStore
{
    private readonly ConcurrentDictionary<Guid, Invoice> _invoices = new();
    private readonly ConcurrentDictionary<Guid, TaskCompletionSource<Invoice>> _waiters = new();

    public Task<bool> ExistsForOrderAsync(Guid orderId, CancellationToken ct) =>
        Task.FromResult(_invoices.ContainsKey(orderId));

    public Task SaveAsync(Invoice invoice, CancellationToken ct)
    {
        _invoices[invoice.OrderId] = invoice;
        Waiter(invoice.OrderId).TrySetResult(invoice);
        return Task.CompletedTask;
    }

    // a signal instead of Thread.Sleep: the test continues the moment the invoice is written
    public Task<Invoice> WaitForInvoiceAsync(Guid orderId, TimeSpan timeout) =>
        Waiter(orderId).Task.WaitAsync(timeout);

    private TaskCompletionSource<Invoice> Waiter(Guid orderId) =>
        _waiters.GetOrAdd(orderId, _ => new TaskCompletionSource<Invoice>(TaskCreationOptions.RunContinuationsAsynchronously));
}

A Real RabbitMQ with Testcontainers

A unit test does not tell you whether the message really goes to the queue, whether the DLX (Dead Letter Exchange) is bound correctly, or whether ack is called correctly. For that you need a real broker, and Testcontainers starts one as a Docker container when the tests begin and removes it at the end:

public sealed class RabbitMqFixture : IAsyncLifetime
{
    private readonly RabbitMqContainer _container = new RabbitMqBuilder("rabbitmq:4.4-management").Build();

    public IConnection Connection { get; private set; } = null!;

    public async ValueTask InitializeAsync()
    {
        await _container.StartAsync();
        var factory = new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) };
        Connection = await factory.CreateConnectionAsync();
    }

    public async ValueTask DisposeAsync()
    {
        await Connection.DisposeAsync();
        await _container.DisposeAsync();
    }
}

Give the image explicitly, with its version: the test environment should run the same broker version as production, since behaviors such as the quorum queue delivery limit or 4.3's delayed retry depend on the version. Testcontainers picks a random host port by default and GetConnectionString() returns the right address; parallel test processes on the same machine do not collide. (In the environment where I ran the experiments for this post, container names and ports were restricted, so I added WithName and WithPortBinding to the fixture and turned off the Ryuk cleanup container; you do not need that in your own project.) With IClassFixture the container is started once for all tests in the class; each test isolates itself by using its own queue name.

The Waiting Problem: Signals Instead of Sleep

The answer to the opening question: an assert right after the publish usually fails, because BasicPublishAsync returns when the broker accepts the message; the consumer receiving and processing it happens on another thread a few milliseconds later. Task.Delay(2000) makes the test pass but creates two problems: every test loses 2 seconds, and on the day 2 seconds are not enough on a slow CI machine, the test breaks "sometimes". The right way is to turn the very thing the test is waiting for into a signal:

public sealed class BillingRabbitMqTests(RabbitMqFixture rabbit) : IClassFixture<RabbitMqFixture>
{
    private readonly InMemoryInvoiceStore _store = new();

    [Fact]
    public async Task Published_order_produces_an_invoice()
    {
        CancellationToken ct = TestContext.Current.CancellationToken;
        await using IChannel channel = await StartBillingAsync("billing-it-1", ct);
        OrderPlaced order = Orders.Sample();

        await channel.BasicPublishAsync("bookstore.ordering", "order.placed", mandatory: true,
            new BasicProperties { MessageId = Guid.NewGuid().ToString(), Persistent = true },
            JsonSerializer.SerializeToUtf8Bytes(order, JsonSerializerOptions.Web), ct);

        Invoice invoice = await _store.WaitForInvoiceAsync(order.OrderId, TimeSpan.FromSeconds(10));
        Assert.Equal(50.00m, invoice.Amount);
    }

    [Fact]
    public async Task Malformed_message_is_dead_lettered()
    {
        CancellationToken ct = TestContext.Current.CancellationToken;
        await using IChannel channel = await StartBillingAsync("billing-it-2", ct);

        await channel.BasicPublishAsync("bookstore.ordering", "order.placed", mandatory: true,
            new BasicProperties { MessageId = "poison-1", Persistent = true },
            Encoding.UTF8.GetBytes("{ not json"), ct);

        BasicGetResult dead = await Eventually.GetAsync(
            () => channel.BasicGetAsync("billing-it-2.dlq", autoAck: true, ct), TimeSpan.FromSeconds(10));
        Assert.Equal("poison-1", dead.BasicProperties.MessageId);
    }
}

StartBillingAsync is a 30-line helper that declares the queue together with its DLX, wires OrderPlacedRabbitHandler to an AsyncEventingBasicConsumer and returns the channel; it uses the same arguments as the production topology code. The first test waits on the TaskCompletionSource in the fake store: the test continues the moment the invoice is written, and 10 seconds is only an upper bound. In the second test there is no signal we can wait on (the application has no callback for a message landing in the DLQ, the Dead-Letter Queue), so we use bounded polling with short intervals:

public static class Eventually
{
    // bounded polling with short intervals instead of a fixed Task.Delay(5000)
    public static async Task<T> GetAsync<T>(Func<Task<T?>> probe, TimeSpan timeout) where T : class
    {
        using var cts = new CancellationTokenSource(timeout);
        while (true)
        {
            if (await probe() is { } value)
                return value;
            await Task.Delay(100, cts.Token);
        }
    }
}

If the condition holds within 100 ms, the test takes 100 ms; if it never holds, the test fails meaningfully with a TaskCanceledException after 10 seconds. A fixed Delay spends the full time in both cases.

The MassTransit Test Harness

For a MassTransit consumer you do not need to start a broker in a container: the test harness runs the whole bus on the in-memory transport and records the messages sent, published and consumed:

public sealed class OrderPlacedConsumerTests
{
    [Fact]
    public async Task Consumer_publishes_InvoiceCreated()
    {
        await using ServiceProvider provider = BuildProvider();
        ITestHarness harness = provider.GetRequiredService<ITestHarness>();
        await harness.Start();
        CancellationToken ct = TestContext.Current.CancellationToken;
        OrderPlaced order = Orders.Sample();

        await harness.Bus.Publish(order, ct);

        // Any waits until the message shows up or the bus has been idle for a while; no polling to write
        IConsumerTestHarness<OrderPlacedConsumer> consumer = harness.GetConsumerHarness<OrderPlacedConsumer>();
        Assert.True(await consumer.Consumed.Any<OrderPlaced>(x => x.Context.Message.OrderId == order.OrderId, ct));
        Assert.True(await harness.Published.Any<InvoiceCreated>(x => x.Context.Message.OrderId == order.OrderId, ct));
    }

    [Fact]
    public async Task Unsupported_currency_publishes_a_fault()
    {
        await using ServiceProvider provider = BuildProvider();
        ITestHarness harness = provider.GetRequiredService<ITestHarness>();
        await harness.Start();
        CancellationToken ct = TestContext.Current.CancellationToken;

        await harness.Bus.Publish(Orders.Sample("XXX"), ct);

        Assert.True(await harness.Published.Any<Fault<OrderPlaced>>(ct));
        Assert.False(await harness.Published.Any<InvoiceCreated>(ct));
    }

    private static ServiceProvider BuildProvider() => new ServiceCollection()
        .AddSingleton<IInvoiceStore, InMemoryInvoiceStore>()
        .AddScoped<InvoiceService>()
        .AddMassTransitTestHarness(x => x.AddConsumer<OrderPlacedConsumer>())
        .BuildServiceProvider(validateScopes: true);
}

Consumed.Any and Published.Any wait until the expected message is recorded or the bus has been idle for a while; the waiting problem is solved inside the harness. That has a side effect: a negative assert saying "this message was not published" waits for the inactivity period to pass, to be sure the message will not come. According to the MassTransit documentation that period (TestInactivityTimeout) is 1.2 seconds by default, which explains why the fault test in the output below takes 1.4 seconds. According to the documentation, harness.Act(...), introduced in v9.1, removes these waits. The second test also locks in the fault path from the eighth and ninth posts: an unsupported currency produces a Fault<OrderPlaced> and no invoice event is published.

Contract Tests: Can v2 Read a v1 Message?

In the sixth post of the series we moved OrderPlaced to v2: an optional CustomerEmail was appended. A test should prove that decision has not been broken; the JSON (JavaScript Object Notation) produced by the v1 producer in production goes into the test as a pinned sample:

// v2: an optional field appended at the end (additive change)
public sealed record OrderPlacedV2(
    Guid OrderId, Guid CustomerId, decimal Total, string Currency, DateTimeOffset PlacedAt,
    IReadOnlyList<OrderPlacedItem> Items, string? CustomerEmail = null);

// breaking change: Total -> TotalAmount
public sealed record OrderPlacedRenamed(
    Guid OrderId, Guid CustomerId, decimal TotalAmount, string Currency, DateTimeOffset PlacedAt,
    IReadOnlyList<OrderPlacedItem> Items);

public sealed class ContractTests
{
    // JSON actually produced by the v1 producer, pinned as a sample in the file
    private const string V1Json = """
        {"orderId":"6c1f0e7a-2b8e-4a53-9d1e-3f8a7b6c5d4e","customerId":"0d9b2c1a-7e6f-4a5b-8c3d-2e1f0a9b8c7d",
         "total":50.00,"currency":"EUR","placedAt":"2026-11-08T10:15:00+00:00",
         "items":[{"bookId":"9a8b7c6d-5e4f-4a3b-2c1d-0e9f8a7b6c5d","quantity":2,"unitPrice":25.00}]}
        """;

    private static readonly JsonSerializerOptions Strict = new(JsonSerializerDefaults.Web)
    {
        RespectRequiredConstructorParameters = true
    };

    [Fact]
    public void V2_consumer_reads_v1_message()
    {
        OrderPlacedV2 message = JsonSerializer.Deserialize<OrderPlacedV2>(V1Json, Strict)!;

        Assert.Equal(50.00m, message.Total);
        Assert.Null(message.CustomerEmail);
    }

    [Fact]
    public void Rename_is_silent_with_default_options()
    {
        OrderPlacedRenamed message = JsonSerializer.Deserialize<OrderPlacedRenamed>(V1Json, JsonSerializerOptions.Web)!;

        Assert.Equal(0m, message.TotalAmount);
    }

    [Fact]
    public void Rename_is_caught_with_strict_options()
    {
        JsonException ex = Assert.Throws<JsonException>(() => JsonSerializer.Deserialize<OrderPlacedRenamed>(V1Json, Strict));
        Assert.Contains("totalAmount", ex.Message);
    }
}

Three tests lock in three different things. The v2 consumer can read a v1 message and the new field comes back as null: the additive change is safe. The reverse is tested too (V1_consumer_reads_v2_message, not shown above): System.Text.Json ignores unknown properties by default, so the old consumer does not choke on the new message. The real lesson is in the third: the breaking change that turns Total into TotalAmount raises no error with default options, and TotalAmount silently becomes 0. An invoicing service issuing invoices for 0 is a far more expensive failure than an exception. With RespectRequiredConstructorParameters, introduced in .NET 9, turned on, deserialization fails when a constructor parameter without a default value is missing from the JSON. The exception message from a small console run with the same JSON and options:

JSON deserialization for type 'OrderPlacedRenamed' was missing required properties including: 'totalAmount'.

Using this setting in the consumer's own deserialization too turns a breaking change from silent data corruption into a failure that lands in the DLQ and fires an alert. Feeding contract tests with real samples from production is the safest: storing a message produced by the latest version of the producer as a file and reading it in the consumer tests closes the gap between "compatible on paper" and "compatible in reality". On the Kafka side, Schema Registry's compatibility check does the same job at the schema level; that topic is in the sixth post of the series.

Running the Tests

The test project uses xUnit v3 and builds as an executable:

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <IsPackable>false</IsPackable>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.11.0" />
    <PackageReference Include="xunit.v3" Version="4.0.2" />
    <PackageReference Include="xunit.runner.visualstudio" Version="4.0.1" />
    <PackageReference Include="Testcontainers.RabbitMq" Version="4.16.0" />
    <PackageReference Include="MassTransit" Version="8.5.12" />
  </ItemGroup>
  <ItemGroup>
    <Using Include="Xunit" />
  </ItemGroup>
  <ItemGroup>
    <ProjectReference Include="../Bookstore.Billing/Bookstore.Billing.csproj" />
  </ItemGroup>
</Project>

The first time I ran dotnet test on the .NET 10 SDK (Software Development Kit), I got this error:

error : Testing with VSTest target is no longer supported by Microsoft.Testing.Platform on .NET 10 SDK and later. If you use dotnet test, you should opt-in to the new dotnet test experience. For more information, see https://aka.ms/dotnet-test-mtp-error

The 4.x versions of xUnit v3 run on Microsoft.Testing.Platform, and on .NET 10 you have to state explicitly in global.json that dotnet test should use that platform:

{
  "sdk": { "version": "10.0.402", "rollForward": "latestFeature" },
  "test": { "runner": "Microsoft.Testing.Platform" }
}

After that, dotnet test --project Bookstore.Billing.Tests --output Detailed (I kept the test lines and part of the summary):

passed Bookstore.Billing.Tests.InvoiceServiceTests.Unsupported_currency_is_a_permanent_failure (8ms)
passed Bookstore.Billing.Tests.InvoiceServiceTests.Second_delivery_of_same_order_does_not_create_invoice (1ms)
passed Bookstore.Billing.Tests.InvoiceServiceTests.Creates_invoice_with_item_total (2ms)
passed Bookstore.Billing.Tests.ContractTests.Rename_is_silent_with_default_options (18ms)
passed Bookstore.Billing.Tests.ContractTests.V2_consumer_reads_v1_message (0ms)
passed Bookstore.Billing.Tests.ContractTests.V1_consumer_reads_v2_message (2ms)
passed Bookstore.Billing.Tests.ContractTests.Rename_is_caught_with_strict_options (0ms)
passed Bookstore.Billing.Tests.OrderPlacedConsumerTests.Unsupported_currency_publishes_a_fault (1s 381ms)
passed Bookstore.Billing.Tests.OrderPlacedConsumerTests.Consumer_publishes_InvoiceCreated (12ms)
passed Bookstore.Billing.Tests.BillingRabbitMqTests.Published_order_produces_an_invoice (128ms)
passed Bookstore.Billing.Tests.BillingRabbitMqTests.Malformed_message_is_dead_lettered (81ms)
Bookstore.Billing.Tests.dll (net10.0|arm64) passed (3s 243ms)
Test run summary: Passed!
  total: 11
  duration: 3s 428ms

The two tests that talk to a real RabbitMQ took 128 and 81 ms; thanks to the signal and the short polling there is no waiting. Most of the total time is spent starting the container (in the fixture, not counted in the test times). The slowest test is the harness test that does not use a broker at all: nearly all of its 1.4 seconds is the inactivity wait of the negative assert.

Production Notes

CI needs Docker. Testcontainers expects a Docker daemon (or a compatible runtime). GitHub Actions Linux runners have one out of the box; some corporate agents do not, and the tests fail with "could not connect". Pulling and caching images in advance keeps the first test from waiting for minutes.

Share the container, not the queue. Starting a new container per test is expensive; one container per class or collection is enough. But every test should use its own queue and exchange names; two tests listening to the same queue consume each other's messages and break at random.

Pin the version. A test written against rabbitmq:latest can change behavior the day the image is updated. Keep the broker version in tests the same as in production and upgrade them together.

Tests document the topology. The dead-letter test proves on every build that the DLX arguments are right. Calling the same declare code from production and tests, instead of copying the topology between them, keeps the two from drifting apart.

Common Pitfalls

Waiting with Thread.Sleep or a fixed Task.Delay. Both slow and flaky; turn the expected result itself into a signal, or use bounded polling.

Mocking IChannel. A test that passes with a mocked channel also passes when the DLX argument is misspelled. If you want to test broker behavior, use a real broker; if you want to test logic, leave the broker out entirely.

Asserting queue depth right after publishing. The message may still be on its way, or the consumer may already have taken it; the number is misleading either way.

Ignoring the cost of negative asserts. Every harness assert that says "not published" waits for the inactivity period; many negative asserts add seconds to the test run.

Trusting JSON's default tolerance. Default options silently turn a renamed field into its default value; write contract tests with strict options.

A Small Experiment

★☆☆ Add a test to BillingRabbitMqTests that publishes the same OrderPlaced twice and verifies that only one invoice is created. Is the fake store's ExistsForOrderAsync open to a race? Run the test 50 times and interpret the result.

★★☆ Write an integration test for the TTL (Time-To-Live) + DLX retry topology from the eighth post: verify with Eventually that a message failing twice with a transient error succeeds on the third attempt and that the DLQ stays empty. Pass the levels in as parameters to shorten the waits in the test.

★★★ Test the BillingRetryRouter from the eighth post with Testcontainers.Kafka: with a handler that always returns 503, verify that the message lands in the DLT (dead-letter topic) with retry-attempt: 3 and the right original-topic-partition-offset headers.

With tests in place, the next step stops being scary: upgrading versions. In the last post of the series we cover the move from RabbitMQ.Client 6 to 7 and from MassTransit 8 to 9, with real compiler errors and runtime surprises.