MassTransit Consumer Test Passing But Confusing Error Thrown

Question

I'm trying to Unit test a MassTransit Consumer using the MassTransit.Testing Framework and the InMemoryTestHarness.

I'm able to successfully test that a message is sent for two separate consumers so far.

One of the consumers is also successfully consumed, but I get an error message as follows:

R-FAULT loopback://localhost/vhost/input_queue 49820000-5689-0050-3b5c-08d5ecc4708c Acme.Company.Messages.Commands.ISomeCommand Acme.Company.SomeService.Consumers.SomeCommandConsumer(00:00:00.2328493) Failure: The payload was not found: MassTransit.RabbitMqTransport.ModelContext, StackTrace: at GreenPipes.PipeExtensions.GetPayload[TPayload](PipeContext context) at MassTransit.DeferExtensions.Defer[T](ConsumeContext1 context, TimeSpan delay, Action2 callback)

The code at this point is attempting to defer the message for one minute so I wonder whether that is the reason for the missing payload???

The code is as follows:

[TestFixture]
public class SomeCommandConsumerTests
{
    private InMemoryTestHarness _harness;

    private Mock<ISomeRepository> _SomeRepositoryMock;
    private Mock<IAnotherRepository> _AnotherRepositoryMock;

    [OneTimeSetUp]
    public async Task OneTimeInit()
    {
        _harness = new InMemoryTestHarness("vhost");
        _harness.Consumer(() => new SomeCommandConsumer(_SomeRepositoryMock.Object, _AnotherRepositoryMock.Object));
        await _harness.Start();
    }

    [SetUp]
    public void Init()
    {
        _SomeRepositoryMock = new Mock<ISomeRepository>();
        _AnotherRepositoryMock = new Mock<IAnotherRepository>();
        _SomeRepositoryMock.Setup(x => x.UpdateSomeId(It.IsAny<SomeEnum>(), It.IsAny<int>()))
            .Returns(Task.Factory.StartNew(() => { }));
        _SomeRepositoryMock.Setup(x => x.UpdateProcMessage(It.IsAny<string>(), It.IsAny<int>()))
            .Returns(Task.Factory.StartNew(() => { }));
        _SomeRepositoryMock.Setup(
                x => x.UpdateSomeProcStartTime(It.IsAny<int>()))
            .Returns(Task.Factory.StartNew(() => { }));
        _SomeRepositoryMock.Setup(
                x => x.UpdateSomeProcEndTime(It.IsAny<int>()))
            .Returns(Task.Factory.StartNew(() => { }));
    }

    [Test]
    public async Task ProcessMessage_MethodCalledWithSomeCondition_MessageSent()
    {
        //Arrange
        _SomeRepositoryMock.Setup(x => x.GetAsync(It.IsAny<int>())).ReturnsAsync(new Entity
        {
            Property1 = true,
            SomeID = 12345
        });

        await _harness.InputQueueSendEndpoint.Send(new SomeCommand
        {
            MessageType = MessageTypeEnum.SomeMessgae,
            SomeID = 12345
        });

        //Assert
        _harness.Sent.Select<ISomeCommand>().Any().Should().BeTrue();
    }

    [Test]
    public async Task ProcessMessage_MethodCalledWithSomeCondition_CorrectNextStepReturned()
    {
        //Arrange
        _SomeRepositoryMock.Setup(x => x.GetAsync(It.IsAny<int>())).ReturnsAsync(new Control()
        {
            Property1 = true,
            SomeID = 12345
        });

        await _harness.InputQueueSendEndpoint.Send(new SomeCommand
        {
            MessageType = MessageTypeEnum.SomeMessgae,
            SomeID = 12345
        });

        //Assert
        _harness.Consumed.Select<ISomeCommand>().Any().Should().BeTrue();
        _harness.Consumed
            .Select<ISomeCommand>()
            .First()
            .Context
            .Message
            .SomeID
            .Should()
            .Be(12345);
        _harness.Consumed
            .Select<ISomeCommand>()
            .First()
            .Context
            .Message
            .MessageProcessingResult
            .Should()
            .Be(MessageProcessingResult.DeferProcessing);
    }

    [OneTimeTearDown] 
    public async Task Teardown()
    { 
        await _harness.Stop(); 
    }
}

And the code that is being hit in the consumer is:

await context.Defer(TimeSpan.FromMinutes(1));

Basically, what am I missing, is it even an issue?

Answer 1

This is happening because you are using the in-memory test harness with a feature (Defer) that is supported by RabbitMQ. Defer tries to use the RabbitMQ model from the consumer to defer the message, and it isn't there, because in-memory doesn't know anything about it.

If you want to use a more generic solution, use Redeliver instead. You'll need to use the QuartzIntegration library with the in-memory test harness, but it does in-memory message redelivery using that scheduler.

You'll also need to update your RabbitMQ bus configuration to include the cfg.UseDelayedExchangeMessageScheduler(); so that RabbitMQ is used for message scheduling.