如何在ASP.NET环境下编写含异步投影的Marten（基于PostgreSQL）集成测试？

我需要验证基于Marten实现的功能集成效果，非单元测试，重点验证以下流程：

• 控制器动作接收DDD风格命令
• 命令转换为领域事件并写入PostgreSQL
• 异步投影正确生成并写入PostgreSQL

Marten官方文档无具体示例，其代码库中的单元测试耦合度高，无法直接参考。以下是现有业务代码结构：

命令定义 #

public record StartQuest(
    Guid Id, 
    string Name, 
    int Day, 
    string Location, 
    params string[] Members) : IRequest;

聚合、投影与领域事件 #

public class Quest
{
    public Guid Id { get; set; }
}

// 异步投影类，基于事件流生成QuestParty视图
public class QuestPartyProjection : SingleStreamAggregation<QuestParty>
{
    // 从QuestStarted事件初始化视图
    public QuestParty Create(QuestStarted @event) => new() { Name = @event.Name };
    // 用MembersJoined事件更新视图成员列表
    public void Apply(QuestParty view, MembersJoined @event) => view.Members.Fill(@event.Members);
}

// 投影视图模型
public class QuestParty
{
    public Guid Id { get; set; }
    public List<string> Members { get; set; } = new();
    public string Name { get; set; }
}

// 领域事件：成员加入
public record MembersJoined(int Day, string Location, string[] Members);

// 领域事件：任务启动
public record QuestStarted(string Name);

MediatR命令处理流程 #

API控制器 #

[ApiController]
[Route("[controller]")]
public class QuestController : ControllerBase
{
    private readonly IMediator _mediator;

    public QuestController(IMediator mediator)
    {
        _mediator = mediator;
    }

    [HttpPost("[Action]")]
    public async Task<IActionResult> Start([FromBody] StartQuest command, CancellationToken cancellationToken)
    {
        await _mediator.Send(command, cancellationToken);
        return Ok();
    }
}

命令处理器 #

internal sealed class StartHandler : IRequestHandler<StartQuest>
{
    private readonly IDocumentSession _session;

    public StartHandler(IDocumentSession session)
    {
        _session = session;
    }

    public async Task<Unit> Handle(StartQuest command, CancellationToken cancellationToken)
    {
        // 将命令转换为领域事件
        var started = new QuestStarted(command.Name);
        var joined1 = new MembersJoined(command.Day, command.Location, command.Members);

        // 向Quest聚合的事件流写入事件
        _session.Events.StartStream(typeof(Quest), command.Id, started, joined1);
        await _session.SaveChangesAsync(cancellationToken);
        return Unit.Value;
    }
}

应用配置 #

builder.Services.AddMarten(x =>
{
    x.Connection(builder.Configuration.GetConnectionString("Marten")!);
    // 注册异步投影
    x.Projections.Add<QuestPartyProjection>(ProjectionLifecycle.Async);
})
.OptimizeArtifactWorkflow(TypeLoadMode.Static)
// 添加异步投影守护进程
.AddAsyncDaemon(DaemonMode.HotCold)
.UseLightweightSessions()
.InitializeWith();

builder.Services.AddMediatR(typeof(StartHandler));

集成测试实现方案 #

依赖准备 #

测试项目需安装以下NuGet包：

• xunit
• xunit.runner.visualstudio
• Microsoft.AspNetCore.Mvc.Testing
• Marten
• Testcontainers.PostgreSql
• FluentAssertions（可选，用于更简洁的断言）

测试类实现 #

using Marten;
using Marten.Events;
using Testcontainers.PostgreSql;
using Xunit;
using Microsoft.AspNetCore.Mvc.Testing;
using System.Net.Http.Json;

public class QuestIntegrationTests : IAsyncLifetime
{
    private readonly PostgreSqlContainer _postgresContainer;
    private readonly WebApplicationFactory<Program> _webAppFactory;
    private HttpClient _httpClient;
    private IDocumentStore _documentStore;

    public QuestIntegrationTests()
    {
        // 初始化PostgreSQL测试容器
        _postgresContainer = new PostgreSqlBuilder()
            .WithImage("postgres:15")
            .Build();

        // 配置测试Web应用，替换连接字符串为容器地址
        _webAppFactory = new WebApplicationFactory<Program>()
            .WithWebHostBuilder(builder =>
            {
                builder.ConfigureServices(services =>
                {
                    // 替换Marten的连接配置
                    services.AddMarten(x =>
                    {
                        x.Connection(_postgresContainer.GetConnectionString());
                        x.Projections.Add<QuestPartyProjection>(ProjectionLifecycle.Async);
                    })
                    .OptimizeArtifactWorkflow(TypeLoadMode.Static)
                    .AddAsyncDaemon(DaemonMode.HotCold)
                    .UseLightweightSessions()
                    .InitializeWith();
                });
            });
    }

    public async Task InitializeAsync()
    {
        await _postgresContainer.StartAsync();
        _httpClient = _webAppFactory.CreateClient();
        _documentStore = _webAppFactory.Services.GetRequiredService<IDocumentStore>();

        // 初始化数据库 schema
        await _documentStore.Advanced.Clean.DeleteAllDocumentsAsync();
        await _documentStore.Advanced.Clean.DeleteAllEventDataAsync();
        await _documentStore.Schema.ApplyAllConfiguredChangesToDatabaseAsync();
    }

    public async Task DisposeAsync()
    {
        await _httpClient.DisposeAsync();
        await _webAppFactory.DisposeAsync();
        await _postgresContainer.DisposeAsync();
        await _documentStore.DisposeAsync();
    }

    // 测试1：控制器接收命令并返回成功响应
    [Fact]
    public async Task StartQuest_ApiCall_ReturnsOk()
    {
        // Arrange
        var questId = Guid.NewGuid();
        var command = new StartQuest(
            questId,
            "Dragon Slayer",
            1,
            "Red Mountain",
            "Gandalf", "Aragorn");

        // Act
        var response = await _httpClient.PostAsJsonAsync("/Quest/Start", command);

        // Assert
        response.EnsureSuccessStatusCode();
    }

    // 测试2：领域事件正确写入PostgreSQL
    [Fact]
    public async Task StartQuest_Command_WritesEventsToDatabase()
    {
        // Arrange
        var questId = Guid.NewGuid();
        var command = new StartQuest(
            questId,
            "Dragon Slayer",
            1,
            "Red Mountain",
            "Gandalf", "Aragorn");

        // Act
        await _httpClient.PostAsJsonAsync("/Quest/Start", command);

        // Assert
        using var session = _documentStore.QuerySession();
        var events = await session.Events.FetchStreamAsync(questId);

        events.Should().HaveCount(2);
        events[0].Data.Should().BeOfType<QuestStarted>()
            .Which.Name.Should().Be("Dragon Slayer");
        events[1].Data.Should().BeOfType<MembersJoined>()
            .Which.Members.Should().Contain("Gandalf", "Aragorn");
    }

    // 测试3：异步投影正确生成视图
    [Fact]
    public async Task StartQuest_AsyncProjection_CreatesQuestPartyView()
    {
        // Arrange
        var questId = Guid.NewGuid();
        var command = new StartQuest(
            questId,
            "Dragon Slayer",
            1,
            "Red Mountain",
            "Gandalf", "Aragorn");

        // Act
        await _httpClient.PostAsJsonAsync("/Quest/Start", command);

        // 等待异步投影完成（使用Marten守护进程API）
        var daemon = _documentStore.AsyncDaemon();
        await daemon.WaitForNonStaleDataAsync(questId, CancellationToken.None);

        // Assert
        using var session = _documentStore.QuerySession();
        var questParty = await session.LoadAsync<QuestParty>(questId);

        questParty.Should().NotBeNull();
        questParty.Name.Should().Be("Dragon Slayer");
        questParty.Members.Should().Contain("Gandalf", "Aragorn");
    }
}

关键说明 #

1. Testcontainers：启动临时PostgreSQL容器，确保测试环境完全隔离，每次测试使用全新数据库实例
2. WebApplicationFactory：模拟ASP.NET应用运行环境，动态替换生产级连接字符串为测试容器地址
3. 异步投影等待：使用Marten内置的AsyncDaemon.WaitForNonStaleDataAsync方法，确保投影处理完成后再执行断言，避免异步时序问题
4. 数据库初始化：每次测试前清空所有数据并重建schema，保证测试用例的独立性和可重复性

内容的提问来源于stack exchange，提问作者annemartijn