test.cs

// See https://aka.ms/new-console-template for more information
using MassTransit;
using MassTransit.Testing;
using Microsoft.Extensions.DependencyInjection;
using System.Diagnostics;

namespace MTTest;

public class Program
{
    public static async Task Main(string[] args)
    {
        var services = new ServiceCollection();
        services.AddMassTransitTestHarness(x =>
        {
            x.AddConfigureEndpointsCallback((ctx, name, cfg) =>
            {
                cfg.UseMessageRetry(r => r.Intervals(TimeSpan.FromSeconds(1)));

                cfg.UseInMemoryOutbox(ctx, x => x.ConcurrentMessageDelivery = true);
            });

            x.AddConsumer<TestConsumer>();
        });

        await using var sp = services.BuildServiceProvider();
        var harness = sp.GetRequiredService<ITestHarness>();
        await harness.Start();

        var msg = new TestMessage();
        await sp.GetRequiredService<IBus>().Publish(msg);

        var consumed = await harness.Consumed.SelectAsync<TestMessage>().FirstOrDefault();
        Debug.Assert(consumed.Exception is not null);

        var evt = await harness.Published.SelectAsync<TestEvent>().FirstOrDefault();
        Debug.Assert(evt is null); // fails as event is published
    }
}

public class TestConsumer
    : IConsumer<Batch<TestMessage>>
{
    public async Task Consume(ConsumeContext<Batch<TestMessage>> context)
    {
        foreach (var msg in context.Message)
        {
            await msg.Publish(new TestEvent
            {
                Source = msg.Message.Id,
            });

            throw new Exception("Test");
        }
    }
}

public class TestMessage
{
    public Guid Id { get; } = Guid.CreateVersion7();
}

public class TestEvent
{
    public Guid Id { get; } = Guid.CreateVersion7();

    public required Guid Source { get; init; }
}