Testy kolejek z MassTransit

Jakiś czas temu pisałem o testach w kontekście API (tutaj). W tym poście postaram się opisać jak napisać podobne testy, ale gdy nasza aplikacja komunikuje się poprzez kolejkę. Pokaże to z wykorzystaniem biblioteki MassTransit.

Przykładowy konsument

Załóżmy, że mamy następującego konsumenta (consumer) – przyjmuje on jakieś zależności w konstruktorze, a w środku metody Consume publikuje jakąś wiadomość:

public class CustomMessageConsumer : IConsumer<CustomMessage>
{
    private readonly ICustomClass _customClass;
 
    public CustomMessageConsumer(ICustomClass customClass)
    {
        _customClass = customClass;
    }
 
    public async Task Consume(ConsumeContext<CustomMessage> context)
    {
        Console.WriteLine($"Processing {context.Message.Text}");
 
        await context.Publish(
            new OtherMessage { Text = $"{_customClass.DoSomething(context.Message.Text)}" }
        );
    }
}

Nasze wiadomości wyglądają następująco:

public class CustomMessage
{
    public string Text { get; set; }
}
 
public class OtherMessage
{
    public string Text { get; set; }
}

Przykładowy test

Test dla naszego konsumenta może wyglądać następująco:

public async Task CustomMessageConsumerTest()
{
    var messageBus = new InMemoryTestHarness();
    var customMessageConsumer = GetCustomMessageConsumer(messageBus);
 
    await messageBus.Start();
    try
    {
        await messageBus.InputQueueSendEndpoint.Send(new CustomMessage
        {
            Text = "test"
        });
 
        (await messageBus.Consumed.Any<CustomMessage>()).Should().BeTrue();
        (await customMessageConsumer.Consumed.Any<CustomMessage>()).Should().BeTrue();
        (await messageBus.Published.Any<OtherMessage>()).Should().BeTrue();
        (await messageBus.Published.Any<Fault<CustomMessage>>()).Should().BeFalse();
    }
    finally
    {
        await messageBus.Stop();
    }
}

Na początku tworzymy obiekt typu InMemoryTestHarness. Jest to klasa, która zachowuje się jak kolejka, ale działa w pamięci i służy do testów. Udostępnia ona metody pozwalające sprawdzić jakie wiadomości zostały opublikowane i skonsumowane. Następnie tworzymy naszego konsumenta (o tym więcej poniżej) i startujemy kolejkę. To tyle, jeśli chodzi o sekcję „przygotowanie” (arrange).

Jako „akcja” (act) w naszym przypadku będzie wysłanie wiadomości CustomMessage.

W sekcji „sprawdzenie” (assert) dokonujemy następujących sprawdzeń:

wiadomość CustomMessage została skonsumowana przez naszą kolejkę,
wiadomość CustomMessage została skonsumowana przez nasz CustomMessageConsumer,
wiadomość OtherMessage została opublikowana,
wiadomość Fault<CustomMessage> nie została opublikowana – czyli wiadomość CustomMessage została przetworzona poprawnie i nie pojawiły się żadne błędy.

Innymi rzeczami, jakie możemy sprawdzać w sekcji „sprawdzenie”, jest np. czy odpowiedni mock został wywołany (w przypadku testów jednostkowych) albo czy aplikacja ma odpowiedni stan (w przypadku testów e2e).

GetCustomMessageConsumer

W zależności od tego, czy chcemy przetestować naszego konsumenta przy pomocy testów jednostkowych, czy testów e2e, to ta metoda będzie wyglądała inaczej.

W przypadku testów jednostkowych może ona wyglądać tak:

private ConsumerTestHarness<CustomMessageConsumer> GetCustomMessageConsumer(
    InMemoryTestHarness messageBus
)
{
    var customClass = new Mock<ICustomClass>();
    customClass.Setup(s => s.DoSomething(It.IsAny<string>())).Returns("test2");
 
    return messageBus.Consumer(() => new CustomMessageConsumer(customClass.Object));
}

A w przypadku testów e2e może ona wyglądać tak:

private ConsumerTestHarness<CustomMessageConsumer> GetCustomMessageConsumer(
    InMemoryTestHarness messageBus
)
{
    var serviceProvider = GetServiceProvider();
    return messageBus.Consumer(() => serviceProvider.GetService<CustomMessageConsumer>());
}

W metodzie GetServiceProvider tworzylibyśmy ServiceProvider według naszych potrzeb (np. z wykorzystaniem testowego Startup). Taki ServiceProvider może istnieć w kontekście:

danej metody – dla każdego testu tworzymy nowy,
danej klasy – tworzymy jeden na początku i wykorzystujemy go we wszystkich testach w danej klasie,
kilku klas (np. przy użyciu Fixture, gdy używamy xUnit) – tworzymy jeden i wykorzystujemy go w wielu klasach.

Inne podejście do testów

Poprzedni test był z wykorzystaniem InMemoryTestHarness. Alternatywą do tego rozwiązania jest skonfigurowanie naszego MassTransit tak, aby zamiast zwykłej kolejki używał kolejki w pamięci (InMemory).

Czym to się różni od stworzenia InMemoryTestHarness? Główna różnica jest taka, że nie musimy wtedy tworzyć ręcznie naszego konsumenta i może on wykorzystać konfigurację taką samą, jaka jest na produkcji (np. retry policy).

Test może wtedy wyglądać tak:

public async Task CustomMessageConsumerTest()
{
    var customClass = new Mock<ICustomClass>();
    customClass
        .Setup(x => x.DoSomething(It.IsAny<string>()))
        .Throws<Exception>();
 
    var serviceProvider = ConfigureServiceProvider(customClass.Object);
 
    var messageBus = serviceProvider.GetService<IBusControl>();
 
    await messageBus.StartAsync();
    try
    {
        await messageBus.Publish(new CustomMessage
        {
            Text = "test"
        });
 
 
        await Task.Delay(TimeSpan.FromMilliseconds(100)).ConfigureAwait(false);
 
        customClass.Verify(
            x => x.DoSomething(It.IsAny<string>()),
            Times.Exactly(6)
        );
    }
    finally
    {
        await messageBus.StopAsync();
    }
}

W tym teście, zamiast tworzyć messageBus, jest on pobierany z serviceProvider. Tutaj dodatkowo zarejestrowaliśmy interface ICustomClass jako mock, aby można było zrobić na nim sprawdzenia. Równie dobrze można ten krok pominąć i skorzystać z oryginalnych (produkcyjnych) klas.

Ciało metody ConfigureServiceProvider:

private static ServiceProvider ConfigureServiceProvider(ICustomClass customClass)
{
    var startup = new TestStartup();
 
    var serviceCollection = new ServiceCollection();
    serviceCollection.AddSingleton(customClass);
 
    startup.ConfigureMassTransit(serviceCollection);
 
    var serviceProvider = serviceCollection.BuildServiceProvider();
 
    return serviceProvider;
}

Tutaj podobnie samo jak w poprzednim przypadku – możemy taki serviceProvider stworzyć dla danego testu, dla danej klasy albo dla wielu klas z wykorzystaniem Fixture.

Jeśli chodzi o metodę ConfigureMassTransit, w klasie Startup wygląda ona następująco:

public IServiceCollection ConfigureMassTransit(IServiceCollection services)
{
    services.AddMassTransit(serviceCollectionConfigurator =>
    {
        serviceCollectionConfigurator.AddConsumer<CustomMessageConsumer>(
            typeof(CustomMessageConsumerDefinition)
        );
        serviceCollectionConfigurator.AddConsumer<OtherMessageConsumer>(
            typeof(OtherMessageConsumerDefinition)
        );
 
 
        ConfigureQueue(serviceCollectionConfigurator);
    });
 
    return services;
}

Metoda ConfigureQueue (w klasie Startup) wygląda tak:

protected virtual void ConfigureQueue(
    IServiceCollectionBusConfigurator serviceCollectionConfigurator
)
{
    serviceCollectionConfigurator.UsingRabbitMq((context, busCfg) =>
    {
        busCfg.ConfigureEndpoints(context);
 
        busCfg.Host("rabbitmq://localhost", hostCfg =>
        {
            hostCfg.Username("guest");
            hostCfg.Password("guest");
        });
    });
}

a w klasie TestStartup ta metoda wygląda tak:

protected override void ConfigureQueue(
    IServiceCollectionBusConfigurator serviceCollectionConfigurator
)
{
    serviceCollectionConfigurator.UsingInMemory((context, busCfg) =>
    {
        busCfg.ConfigureEndpoints(context);
    });
}

Podsumowanie

Jeśli mamy aplikację (albo chcemy stworzyć), która komunikuję się poprzez kolejkę, nie powinniśmy mieć większych trudności ze stworzeniem odpowiednich testów. Dość łatwo powinniśmy być w stanie stworzyć testy jednostkowy czy e2e.