使用 InMemoryTestFixture 测试 MassTransit 消费者

Question

想要围绕 MassTransit 消费者设计我的测试，我可以在其中向消费者发送包含各种内容的消息。基于消息的内容，消费者将“工作”并转发消息。

我遇到的问题是，在单独的测试夹具中运行其中两个测试时，似乎有一些东西干扰了第二个测试。但是单独运行每个测试都会成功运行。

查看完 MassTransit 测试项目后，我想出了一些示例测试代码来演示我遇到的问题。

[TestFixture]
public class PingPongMessageTestFixture : InMemoryTestFixture
{
    private PongConsumer _pongConsumer;
    protected override void ConfigureInMemoryReceiveEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
    {
        _received = Handled<IPongMessage>(configurator);
    }

    protected override void PreCreateBus(IInMemoryBusFactoryConfigurator configurator)
    {
        var _pingConsumer = new PingConsumer();
        _pongConsumer = new PongConsumer();
        configurator.ReceiveEndpoint("test_ping_queue", e =>
        {
            e.Consumer(() => _pingConsumer);
        });

        configurator.ReceiveEndpoint("test_pong_queue", e =>
        {
            e.Consumer(() => _pongConsumer);
        });
    }

    Task<ConsumeContext<IPongMessage>> _received;

    [Test]
    public async Task test_how_to_test_consumers()
    {
        await Bus.Publish<IPingMessage>(new { MessageId = 100 });
        await _received;

        Assert.IsTrue(_pongConsumer.hitme);
        Assert.AreEqual(100, _pongConsumer.pongMessage.MessageId);
    }

    public class PingConsumer : IConsumer<IPingMessage>
    {
        public Task Consume(ConsumeContext<IPingMessage> context)
        {
            context.Publish<IPongMessage>(new { context.Message.MessageId });
            return Task.CompletedTask;
        }
    }

    public class PongConsumer : IConsumer<IPongMessage>
    {
        internal bool hitme;
        internal IPongMessage pongMessage;
        public Task Consume(ConsumeContext<IPongMessage> context)
        {
            hitme = true;
            pongMessage = context.Message;
            return Task.CompletedTask;
        }
    }

    public interface IPingMessage
    {
        int MessageId { get; set; }
    }

    public interface IPongMessage
    {
        int MessageId { get; set; }
    }
}

此测试将向 ping 消费者发送一条消息，而 ping 消费者本身将向 pong 消费者发送一条消息。

这本身就可以工作并测试 ping 消费者是否会发送 pong 消息。在现实生活场景中，“ping”消费者将更新消息发送到另一个服务，而 pong 消费者只是与测试一起使用的测试消费者。

如果我有第二个测试夹具，对于这个问题非常相似，当两个测试一起运行时它将失败。虽然单独它会通过。

测试做同样的事情

[TestFixture]
public class DingDongMessageTestFixture : InMemoryTestFixture
{
    private DongConsumer _pongConsumer;
    protected override void ConfigureInMemoryReceiveEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
    {
        _received = Handled<IDongMessage>(configurator);
    }

    protected override void PreCreateBus(IInMemoryBusFactoryConfigurator configurator)
    {
        var _dingConsumer = new DingConsumer();
        _dongConsumer = new DongConsumer();
        configurator.ReceiveEndpoint("test_ding_queue", e =>
        {
            e.Consumer(() => _dingConsumer);
        });

        configurator.ReceiveEndpoint("test_dong_queue", e =>
        {
            e.Consumer(() => _dongConsumer);
        });
    }

    Task<ConsumeContext<IDongMessage>> _received;

    [Test]
    public async Task test_how_to_test_consumers()
    {
        await Bus.Publish<IDingMessage>(new { MessageId = 100 });
        await _received;

        Assert.IsTrue(_pongConsumer.hitme);
        Assert.AreEqual(100, _pongConsumer.pongMessage.MessageId);
    }

    public class DingConsumer : IConsumer<IDingMessage>
    {
        public Task Consume(ConsumeContext<IDingMessage> context)
        {
            context.Publish<IDongMessage>(new { context.Message.MessageId });
            return Task.CompletedTask;
        }
    }

    public class DongConsumer : IConsumer<IDongMessage>
    {
        internal bool hitme;
        internal IDongMessage pongMessage;
        public Task Consume(ConsumeContext<IDongMessage> context)
        {
            hitme = true;
            pongMessage = context.Message;
            return Task.CompletedTask;
        }
    }

    public interface IDingMessage
    {
        int MessageId { get; set; }
    }

    public interface IDongMessage
    {
        int MessageId { get; set; }
    }
}

这是测试 Masstransit 消费者的好方法吗？

如果是这样，我是否需要以某种方式重置每个测试夹具的 InMemoryTestFixture？

Answer 1

在您的测试装置中，我认为不应该有任何冲突，但由于与 NUnit 的交互，由于正在使用基类继承，我可能不知道其中的某些内容。

如果您直接使用InMemoryTestHarness（与文本装置相同的功能，但没有任何测试框架依赖），我希望您不会遇到两个同时执行的测试之间的任何交互。

您的方法应该是这样，但我还是建议使用InMemoryTestHarness 而不是夹具。

链接了一个示例测试：https://github.com/MassTransit/MassTransit/blob/master/src/MassTransit.Tests/Testing/ConsumerTest_Specs.cs

Answer 2

这种行为的关键在于source code 对应InMemoryTestFixture。

public class InMemoryTestFixture : BusTestFixture
{

    ...

    [OneTimeSetUp]
    public Task SetupInMemoryTestFixture()
    {
        return InMemoryTestHarness.Start();
    }

    [OneTimeTearDown]
    public async Task TearDownInMemoryTestFixture()
    {
        await InMemoryTestHarness.Stop().ConfigureAwait(false);

        InMemoryTestHarness.Dispose();
    }

    ...

}

从这个 sn-p 可以看出，测试工具在 [OneTimeSetUp] 和 [OneTimeTearDown] 标记中启动和停止，即在运行 [TestFixture] 中的任何测试之前以及在夹具中的所有测试完成之后- 不是每个测试用例。

我的解决方案是每次都创建一个新的测试夹具。我相信这是MassTransit.TestFramework 的作者的意图，因为这是他们在Common_SagaStateMachine example 中所做的。