如何通过InMemoryTestHarness驱动MassTransitStateMachine及编写单元测试

问题1：如何通过InMemoryTestHarness成功驱动MassTransitStateMachine？ #

InMemoryTestHarness是MassTransit专为本地测试打造的工具，完全在内存中运行，不需要依赖外部消息队列，是状态机单元测试的绝佳选择。核心操作步骤如下：

• 1. 引入依赖：先安装MassTransit.Testing NuGet包，这是包含InMemoryTestHarness的核心库。
• 2. 初始化测试上下文：在测试类里声明InMemoryTestHarness实例、你的状态机对象，以及对应的内存版Saga仓库InMemorySagaRepository<T>。
• 3. 关联Harness与状态机：在测试初始化方法中，创建harness实例，通过StateMachineSaga方法把状态机和Saga仓库注册到harness的配置里。
• 4. 启动测试环境：每个测试执行前启动harness，确保消息处理环境就绪。
• 5. 触发状态机流转：用harness的InputQueueSendEndpoint发送事件/命令，或者直接发布事件，驱动状态机执行状态切换逻辑。
• 6. 验证结果：利用harness提供的断言方法（比如Consumed.Any<T>()）确认消息被处理，再通过内存Saga仓库检查状态实例的状态值、数据是否符合预期。
• 7. 清理资源：测试结束后停止harness，释放内存资源，避免影响后续测试。

问题2：完整的MS Test单元测试示例（针对ProcedureStateMachine） #

假设你的ProcedureStateMachine是一个包含启动、完成流程的简单状态机，下面是完整的可运行测试类示例（包含状态机的基础定义，方便你直接参考）：

using MassTransit;
using MassTransit.Testing;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Linq;
using System.Threading.Tasks;

// ------------------- 状态机相关基础定义（示例） -------------------
public enum ProcedureStates
{
    Initial,
    Started,
    Completed
}

public class ProcedureState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    // 可添加其他业务字段，比如流程名称、创建时间等
}

public record StartProcedure(Guid ProcedureId);
public record CompleteProcedure(Guid ProcedureId);

public class ProcedureStateMachine : MassTransitStateMachine<ProcedureState>
{
    public ProcedureStateMachine()
    {
        InstanceState(x => x.CurrentState);

        // 关联事件与Saga实例ID
        Event(() => Start, x => x.CorrelateById(m => m.Message.ProcedureId));
        Event(() => Complete, x => x.CorrelateById(m => m.Message.ProcedureId));

        // 初始状态逻辑：收到Start事件后切换到Started状态
        Initially(
            When(Start)
                .TransitionTo(Started)
                .Then(context =>
                {
                    // 这里可以添加启动流程的业务逻辑，比如记录日志
                }));

        // Started状态逻辑：收到Complete事件后切换到Completed状态
        During(Started,
            When(Complete)
                .TransitionTo(Completed));
    }

    public State Started { get; private set; }
    public State Completed { get; private set; }

    public Event<StartProcedure> Start { get; private set; }
    public Event<CompleteProcedure> Complete { get; private set; }
}

// ------------------- MS Test测试类 -------------------
[TestClass]
public class ProcedureStateMachineTests
{
    private InMemoryTestHarness _harness;
    private ProcedureStateMachine _machine;
    private InMemorySagaRepository<ProcedureState> _sagaRepository;

    [TestInitialize]
    public async Task SetupTestEnvironment()
    {
        // 初始化状态机和内存Saga仓库
        _machine = new ProcedureStateMachine();
        _sagaRepository = new InMemorySagaRepository<ProcedureState>();

        // 创建并配置内存测试Harness
        _harness = new InMemoryTestHarness();
        _harness.StateMachineSaga(_machine, _sagaRepository);

        // 启动Harness，准备接收消息
        await _harness.Start();
    }

    [TestCleanup]
    public async Task CleanupTestEnvironment()
    {
        // 停止Harness，释放所有资源
        await _harness.Stop();
    }

    [TestMethod]
    public async Task Should_Switch_To_Started_State_After_Receiving_StartProcedure()
    {
        // 准备测试用的流程ID
        var testProcedureId = Guid.NewGuid();

        // 发送启动流程的事件
        await _harness.InputQueueSendEndpoint.Send(new StartProcedure(testProcedureId));

        // 等待Harness处理完该事件
        await _harness.Consumed.Any<StartProcedure>();

        // 验证Saga实例存在且状态为Started
        var sagaInstance = _sagaRepository.Sagas.FirstOrDefault(s => s.CorrelationId == testProcedureId);
        Assert.IsNotNull(sagaInstance, "Saga实例未被创建");
        Assert.AreEqual(_machine.Started.Name, sagaInstance.CurrentState, "状态未切换到Started");
    }

    [TestMethod]
    public async Task Should_Switch_To_Completed_State_After_Receiving_CompleteProcedure()
    {
        var testProcedureId = Guid.NewGuid();

        // 先发送启动事件，让状态机进入Started状态
        await _harness.InputQueueSendEndpoint.Send(new StartProcedure(testProcedureId));
        await _harness.Consumed.Any<StartProcedure>();

        // 发送完成流程的事件
        await _harness.InputQueueSendEndpoint.Send(new CompleteProcedure(testProcedureId));
        await _harness.Consumed.Any<CompleteProcedure>();

        // 验证状态流转到Completed
        var sagaInstance = _sagaRepository.Sagas.FirstOrDefault(s => s.CorrelationId == testProcedureId);
        Assert.IsNotNull(sagaInstance);
        Assert.AreEqual(_machine.Completed.Name, sagaInstance.CurrentState, "状态未切换到Completed");
    }
}

测试关键细节说明： #

• TestInitialize/TestCleanup：确保每个测试都在独立的干净环境中运行，避免测试间的状态污染。
• StateMachineSaga注册：让Harness托管状态机的消息处理逻辑，自动路由事件到状态机。
• Consumed.Any()：等待指定消息被消费完成，保证状态机有足够时间处理消息后再执行断言，避免异步时序问题。
• 直接访问内存Saga仓库：可以直观检查Saga实例的状态和数据，验证业务逻辑是否正确执行。

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