Test harness is missing feature to wait for all messages being consumed.

[pacyfist]

Contact Details

No response

Version

8.x

On which operating system(s) are you experiencing the issue?

No response

Using which broker(s) did you encounter the issue?

No response

What are the steps required to reproduce the issue?

As an example I prepared this xunit test that fails proving that the `.Any<>` will unlock the thread after first message is processed. After long search I was unable to substitute the `await harness.Consumed.Any<UpdateMessage>();` line with anything that would make this test pass.


public class State
{
    public int Count { get; set; } = 0;
    public int LastMessageId { get; set; } = 0;
}

public class UpdateMessage
{
    required public int Id { get; set; }
    required public int Delay { get; set; }
}

public class UpdateMessageConsumer : IConsumer<UpdateMessage>
{
    private readonly State _state;

    public UpdateMessageConsumer(State state)
    {
        _state = state;
    }

    public async Task Consume(ConsumeContext<UpdateMessage> context)
    {
        await Task.Delay(context.Message.Delay);

        lock (_state)
        {
            _state.Count++;
            _state.LastMessageId = context.Message.Id;
        }
    }
}


[Fact]
public async Task WaitForAllMessagesSent()
{
    var provider = new ServiceCollection()
        .AddSingleton<State>()
        .AddMassTransitTestHarness(x => x.AddConsumer<UpdateMessageConsumer>())
        .BuildServiceProvider(true);

    var harness = provider.GetRequiredService<ITestHarness>();
    var state = provider.GetRequiredService<State>();

    try
    {
        await harness.Start();

        await harness.Bus.Publish(new UpdateMessage()
        {
            Id = 1,
            Delay = 200,
        });

        await harness.Bus.Publish(new UpdateMessage()
        {
            Id = 2,
            Delay = 100,
        });

        await harness.Consumed.Any<UpdateMessage>();

        Assert.Equal(2, state.Count);
    }
    finally
    {
        await harness.Stop();
    }
}

What is the expected behavior?

I'm using end to end testing with single mass transit test harness injected into many (WebApplicationFactory) microservices to allow them to communicate between one another. Unfortunately I'm unable to ensure that all messages were consumed before asserting the system state.

Documentation seems to show only a single message example, and completely omits possibility that there can be a hidden second message published by one of the tested microservices that the test should not directly interface with.

Is there a possibility to create .All<> and .All (second one guarantees that all messages of all types published in the harness are consumed?

What actually happened?

I was unable to find any info how to wait for all messages in a test harness to be consumed

Related log output, including any exceptions

No response

Link to repository that demonstrates/reproduces the issue

No response

[phatboyg]

You can wait for the bus to be idle using:

await harness.InactivityTask;

You can adjust the test inactivity timeout using:

x.SetTestTimeouts(testInactivityTimeout: TimeSpan.FromSeconds(10));

[pacyfist]

I guess this approach indeed does achieve this effect. The problem is that it increases the length of every tests by 10 seconds (or however many seconds I setup in the delay) which is not optimal when there are hundreds of tests.

[phatboyg]

The default timeout is typically enough (.6 or 1.2 seconds, I can't recall), you really only need to increase if it your tests include code that isn't sending/consuming messages.

[darooman]

It would be useful if this was documented on the MassTransit testing page https://masstransit.io/documentation/concepts/testing

[phatboyg]

PR's are accepted.

[htuomola]

Hi,

I have somewhat of a similar issue. Saga is created by publishing one message and then later it is updated by publishing second message. await harness.InactivityTask; completes before the Saga has processed the published message.

What's weird is I tried to create a simple repro but couldn't reproduce it yet. My proper Saga is only transitioning to new state when receiving the event. I managed to hook up Open telemetry Activities to XUnit logging. Based on those logs it seems that the Saga would start and complete processing of the event message only after the InactivityTask was done.

Using Task.Delay(100) (or 10) instead of harness.InactivityTask works but it fails sometimes in CI build.

Test code

_provider = new ServiceCollection()
	.AddMassTransitTestHarness(cfg =>
	{
		cfg.AddSagaStateMachine<DemoSaga, DemoState>();
		// Configures handler but this is not used in ResultsApproved phase
		cfg.AddHandler<JobExecutionRequest>(ctx => ctx.RespondAsync(new JobExecutionResponse()
		{
			Id = Guid.NewGuid(),
			JobLogicalName = ctx.Message.JobLogicalName,
			Success = true
		}));
	})
	.BuildServiceProvider(true);
_harness = _provider.GetRequiredService<ITestHarness>();
_sagaHarness = _harness.GetSagaStateMachineHarness<DemoSaga, DemoState>();

// Send first message and create saga, all goes well

// Publish second message
_testOutputHelper.WriteLine("Publising ResultsApproved");
await _harness.Bus.Publish(new ResultsApproved()
{
	CorrelationId = sagaId
});
_testOutputHelper.WriteLine("Published ResultsApproved");

await _harness.InactivityTask;
_testOutputHelper.WriteLine("Inactivity Task done");

var hasMessages = await _harness.Consumed.Any<ResultsApproved>();
_testOutputHelper.WriteLine("Got consumed ResultsApproved: " + hasMessages);
// this fails
hasMessages.Should().BeTrue();

Logs

Publising ResultsApproved
08.32.32.065109: START  urn:message:MyApp.Events:ResultsApproved send, MessageType=urn:message:MyApp.Events:ResultsApproved
08.32.32.066459: START  DemoState receive, 
08.32.32.066969: END  urn:message:MyApp.Events:ResultsApproved send, Duration=00:00:00.0018605
Published ResultsApproved
Inactivity Task done
Got consumed ResultsApproved: False
08.32.32.069253: START [DemoSaga (id=47e8242e-e5fd-49ae-bdc2-f9af374d7066)] DemoState process, MessageType=urn:message:MyApp.Events:ResultsApproved
08.32.32.069967: END [DemoSaga (id=47e8242e-e5fd-49ae-bdc2-f9af374d7066)] DemoState process, Duration=00:00:00.0007140
08.32.32.071152: END  DemoState receive, Duration=00:00:00.0046926

[phatboyg]

x.SetTestTimeouts(...), increase the inactivity timeout to suit your scenario.

[htuomola]

Yes I tried that as the first attempt but it has no effect on my test. Adding cfg.SetTestTimeouts(testInactivityTimeout: TimeSpan.FromSeconds(10), testTimeout: TimeSpan.FromSeconds(10)) changes nothing and after Publish was called await harness.InactivityTask completes almost instantly.

It bugs me why the simple failed to reproduce it. My proper saga has 2 requests, ~12 states and 7 events setup.