AsyncAPI Code Generation with Corvus: Testing
At endjin, we maintain Corvus.JsonSchema, and in the previous post we looked at filtering large specs down to the channels you actually need.
Now let's look at testing. And for generated messaging code, the interesting question isn't really "how". It's "what should I actually be testing here?"
What to test (and what not to)
The generated code itself - including serialization, schema validation, and channel address construction - is tested extensively in the Corvus.JsonSchema repository. You don't need to verify that validation works, or that the producer serializes a payload correctly. That's the generator's job.
What you do need to test is everything you wrote yourself: your handler logic, your error policy choices, the way your producer integrates into your domain layer, and the end-to-end behaviour when producer and consumer talk to each other through your chosen contract.
This gives us three testing levels, each with a clear purpose.
Level 1: Unit testing your handlers
Your handler implements a generated interface. For a streetlights consumer, that might be IReceiveLightMeasurementHandler with a method that receives a typed LightMeasuredPayload. The handler is where your business logic lives, and it's the most important thing to test.
The good news is that handlers are trivial to test in isolation. You don't need a transport, a consumer, or any messaging infrastructure at all. Just build a payload with the property-parameter Build() factory and call the method (if you're not familiar with this way of constructing JSON values, the Corvus.JsonSchema documentation explains the mutable document model in detail):
[Test]
public async Task Handler_records_measurement_when_payload_is_valid()
{
// Arrange
LightMeasurementHandler handler = new();
LightMeasuredPayload payload = LightMeasuredPayload.Build(
lumens: 1024, sentAt: DateTimeOffset.UtcNow);
// Act
await handler.HandleLightMeasuredAsync(payload);
// Assert
Assert.AreEqual(1, handler.ReceivedCount);
Assert.AreEqual(1024, handler.LastLumens);
}
This runs in microseconds. There's no transport to set up, no broker to start, and no cleanup to worry about. The payload arrives already validated (as it would in production, because the consumer validates before calling your handler), so your tests focus purely on what your handler does with valid data.
You should also test edge cases at the handler level - boundary values, null optional fields, and the interactions your handler has with its own dependencies (repositories, external services, etc.).
Level 2: Testing producer output
The generated producer validates and serializes messages, then writes them to whatever transport you provide. The InMemoryMessageTransport from Corvus.Text.Json.AsyncApi.Testing captures those messages so you can inspect them:
dotnet add package Corvus.Text.Json.AsyncApi.Testing
[Test]
public async Task Producer_publishes_to_correct_channel_with_resolved_parameters()
{
// Arrange
await using InMemoryMessageTransport transport = new();
TurnOnProducer producer = new(transport, ValidationMode.Basic);
// Act
await producer.PublishTurnOnOffAsync(
payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
streetlightId: "lamp-042");
// Assert
Assert.AreEqual(1, transport.PublishedMessages.Count);
var message = transport.PublishedMessages[0];
Assert.That(message.Channel, Does.Contain("lamp-042"));
Assert.That(message.Channel, Does.Contain("turn.on"));
}
This tells you that your domain code is producing the right messages in the right channels. You can also verify that validation rejects bad payloads.
Level 3: End-to-end with InMemoryMessageTransport
The most valuable integration test verifies the full round-trip: a producer publishes, the in-memory transport delivers to a subscribed consumer, the consumer validates and dispatches to your handler, and your handler produces some observable effect.
[Test]
public async Task Published_message_flows_through_consumer_to_handler()
{
// Arrange
await using InMemoryMessageTransport transport = new();
LightMeasurementHandler handler = new();
ReceiveLightMeasurementConsumer consumer = new(
transport,
handler,
validationMode: ValidationMode.Basic);
await consumer.StartAsync();
TurnOnProducer producer = new(transport, ValidationMode.Basic);
// Act - publish a command (different channel from consumer)
await producer.PublishTurnOnOffAsync(
payload: TurnOnOffPayload.Build(command: "on"u8, sentAt: DateTimeOffset.UtcNow),
streetlightId: "lamp-001");
// Deliver a measurement to the consumer's channel
await transport.DeliverAsync<LightMeasuredPayload>(
"smartylighting.streetlights.1.0.action.{streetlightId}.lighting.measured",
"""{"lumens":512,"sentAt":"2026-01-15T10:30:00Z"}"""u8.ToArray());
// Assert
Assert.AreEqual(1, handler.ReceivedCount);
Assert.AreEqual(512, handler.LastLumens);
// Cleanup
await consumer.StopAsync();
}
Notice the use of DeliverAsync<T>(). This simulates raw bytes arriving from a broker on a subscribed channel. It exercises the consumer's deserialization, validation, and handler dispatch pipeline, which is exactly the path messages take in production.
Testing error policy behaviour
The in-memory transport also lets you verify your error policy choices. Deliver an invalid message and confirm it ends up in the dead-letter collection:
[Test]
public async Task Invalid_message_is_dead_lettered()
{
// Arrange
await using InMemoryMessageTransport transport = new();
LightMeasurementHandler handler = new();
IMessageErrorPolicy policy = new DefaultMessageErrorPolicy(
deserializationAction: MessageErrorAction.DeadLetter,
handlerAction: MessageErrorAction.DeadLetter,
transportAction: MessageErrorAction.Abort);
ReceiveLightMeasurementConsumer consumer = new(
transport,
handler,
errorPolicy: policy,
validationMode: ValidationMode.Basic);
await consumer.StartAsync();
// Act - deliver something that doesn't match the schema
await transport.DeliverAsync<LightMeasuredPayload>(
"smartylighting.streetlights.1.0.action.{streetlightId}.lighting.measured",
"""{"notAValidField":true}"""u8.ToArray());
// Assert
Assert.AreEqual(0, handler.ReceivedCount);
Assert.AreEqual(1, transport.DeadLetteredMessages.Count);
await consumer.StopAsync();
}
This is a test you absolutely want, because it verifies that malformed messages don't silently disappear and don't crash your consumer. The DeadLetteredMessages collection on the in-memory transport gives you full visibility into what was rejected and why.
When to use Testcontainers
The in-memory transport covers the vast majority of test scenarios. But there are some things it can't verify: transport-specific behaviour like Kafka partition assignment, NATS JetStream acknowledgement semantics, or AMQP connection recovery after a broker restart.
For those cases, the Corvus.JsonSchema repository itself uses Testcontainers to spin up real broker instances in Docker. You can follow the same approach for your own integration tests, but treat these as a small number of targeted tests rather than your primary test suite. They're slower to run, harder to debug, and test transport behaviour rather than your business logic.
The general principle is: test your logic with InMemoryMessageTransport (fast, deterministic, no dependencies), and test your infrastructure assumptions with Testcontainers (slow, realistic, requires Docker).
A practical test structure
For a typical AsyncAPI-based service, a sensible test organisation might look like:
- Handler unit tests - fast, focused on business logic, one per behaviour
- Producer output tests - verify channel routing and validation rejection
- End-to-end flow tests - confirm the full publish-validate-handle pipeline
- Error policy tests - verify dead-lettering, skip, and abort behaviour
- Testcontainers tests (optional) - a handful of smoke tests against a real broker
The first four categories all use InMemoryMessageTransport and run in milliseconds. They belong in your CI pipeline and should run on every commit. The Testcontainers tests, if you have them, typically run in a separate slower pipeline or on a less frequent schedule.
If you'd like to see all of these patterns in a working example, the AsyncAPI end-to-end recipe in the Corvus.JsonSchema documentation walks through a complete producer-consumer integration with the in-memory transport.
In the next post, we'll look at implementing your own transport - for when you need a broker that isn't covered by the built-in packages.
