公共交通集成测试：TestHarness 不消费消息

Question

我已经设置了 MassTransit 集成测试，但我无法让它工作。我正在使用 Web 应用程序工厂来设置集成测试。我想消费一条消息，然后在数据库中创建一条记录。我能够验证测试工具是否正在发布消息，但工具从未使用过该消息。 “IsMessageConsumedByService”函数始终返回 false。当我调试/进入功能时，消费者工具始终显示 0 条消息已消耗。代码如下所示。任何帮助表示赞赏。

Web 应用工厂

public class CustomWebApplicationFactory
    : WebApplicationFactory<Program>, IAsyncLifetime
{
    private readonly MsSqlContainer _mssqlContainer = new MsSqlBuilder().Build();
    private readonly RabbitMqContainer _rabbitMqContainer = new RabbitMqBuilder().Build();
    protected override void ConfigureWebHost(IWebHostBuilder builder)
    {
        builder.ConfigureTestServices(services =>
        {
            services.RemoveAll<TenantStoreDbContext>();
            services.RemoveAll<DbConnection>();
            services.RemoveAll(typeof(DbContextOptions<TenantStoreDbContext>));

            services.AddDbContext<TenantStoreDbContext>((container, options) =>
            {
                options.UseSqlServer(_mssqlContainer.GetConnectionString());
            });

            services.AddMassTransitTestHarness(x =>
            {
                x.AddConsumer<StripeSubscriptionCreatedConsumer>();
                // ToDo: Test with RabbitMq container after in memory works
            });
        });

        builder.UseEnvironment("Local");
    }
    public async Task InitializeAsync() 
    {
        await _mssqlContainer.StartAsync();
        await _rabbitMqContainer.StartAsync();
    }

    public new async Task DisposeAsync() 
    {
        await _mssqlContainer.DisposeAsync();
        await _rabbitMqContainer.DisposeAsync();
    } 
}

测试班

 public class NewTenantTests : IClassFixture<CustomWebApplicationFactory>
    {
        private readonly CustomWebApplicationFactory _factory;

        public NewTenantTests(CustomWebApplicationFactory factory)
        {
            _factory = factory;
        }

        [Fact]
        public async void When_stripe_created_event_is_recieved_a_new_tenant_is_created()
        {
            // Arrange
            var serviceProvider = _factory.Services.GetRequiredService<IServiceProvider>();
            var testHarness = await StartTestHarness(serviceProvider);
            var subscriptionEvent = BuildStripeSubscriptionCreated();

            // Act
            await PublishMessage(subscriptionEvent, testHarness);

            // Assert
            (await IsMessageConsumedByService(testHarness, subscriptionEvent)).ShouldBe(true);
            (await IsMessageConsumedByConsumer(serviceProvider, subscriptionEvent)).ShouldBe(true);
            (await AssertNewTenantWasCreated(serviceProvider, subscriptionEvent.TenantIdentifier)).ShouldBe(true);

        }

        private static async Task<ITestHarness> StartTestHarness(IServiceProvider provider)
        {
            var harness = provider.GetTestHarness();
            await harness.Start();

            return harness;
        }

        private static StripeSubscriptionCreatedEvent BuildStripeSubscriptionCreated()
        {
            var subscriptionEvent = new StripeSubscriptionCreatedEvent()
            {
                SubscriptionId = "abc",
                CustomerId = "abc",
                Description = "description",
                CurrentPeriodStart = DateTime.Now,
                CurrentPeriodEnd = DateTime.Now.AddMonths(1),
                TenantIdentifier = Guid.NewGuid().ToString()
            };

            return subscriptionEvent;
        }

        private static async Task PublishMessage(StripeSubscriptionCreatedEvent subscriptionCreated, ITestHarness harness)
        {
            await harness.Bus.Publish(subscriptionCreated);
        }

        private static async Task<bool> IsMessageConsumedByConsumer(IServiceProvider serviceProvider,
            StripeSubscriptionCreatedEvent subscriptionCreated)
        {
            return await serviceProvider.
                GetRequiredService<IConsumerTestHarness<StripeSubscriptionCreatedConsumer>>().
                Consumed.Any<StripeSubscriptionCreatedEvent>(x => x.Context.Message.EventId == subscriptionCreated.EventId);
        }

        private static async Task<bool> IsMessageConsumedByService(ITestHarness harness,
            StripeSubscriptionCreatedEvent subscriptionCreated)
        {
            var consumerHarnes = harness.GetConsumerHarness<StripeSubscriptionCreatedConsumer>();
            return await consumerHarnes.Consumed.Any<StripeSubscriptionCreatedEvent>(x => x.Context.Message.EventId == subscriptionCreated.EventId);
        }

        private static async Task<bool> AssertNewTenantWasCreated(IServiceProvider serviceProvider, string tenantIdentifier)
        {
            var repository = serviceProvider.GetRequiredService<ITenantStoreRepository>();
            var identifier = TenantIdentifier.Create(tenantIdentifier).Value;
            var tenant = await repository.GetTenantByIdentifierAsync(identifier);

            return tenant != null;
        }
    }

Answer 1

首先，如果没有一个独立的项目重现这个问题，就不可能猜测。

我建议添加日志记录以查看发生了什么，在配置测试工具之前使用 XUnit 测试输出助手到您的容器。如果您的消费者从未收到消息，那么背后可能有无数理由。

有articles关于如何从

ILogger

兼容源获得测试输出。