2

跟进：如何编写 MassTransitStateMachine 单元测试？

这是一个简单的状态机的简单测试类（使用 MS 测试），称为ProcedureStateMachine（注意：这对我们来说不是一个真正的生产状态机......只是我曾经玩过MassTransitStateMachine一段时间的一个实验......看起来一个方便的自包含的地方来尝试进行单元测试）：

[TestClass]
public class ProcedureStateMachineTests
{
    private ProcedureStateMachine _machine;
    private InMemoryTestHarness _harness;
    private StateMachineSagaTestHarness<ProcedureContext, ProcedureStateMachine> _saga;

    [TestInitialize]
    public void SetUp()
    {
        _machine = new ProcedureStateMachine();
        _harness = new InMemoryTestHarness();
        _saga = _harness.StateMachineSaga<ProcedureContext, ProcedureStateMachine>(_machine);

        _harness.Start().Wait();
    }

    [TestCleanup]
    public void TearDown()
    {
        _harness.Stop().Wait();
    }

    [TestMethod]
    public async Task That_Can_Start()
    {
        // Arrange
        // Act
        await _harness.InputQueueSendEndpoint.Send(new BeginProcessing
        {
            ProcedureId = Guid.NewGuid(),
            Steps = new List<string> {"A", "B", "C" }
        });

        // Assert
        var sagaContext = _saga.Created.First();
        sagaContext.Saga.RemainingSteps.ShouldHaveCountOf(2);
    }
}

这是状态机类本身：

public class ProcedureStateMachine : MassTransitStateMachine<ProcedureContext>
{
    public State Processing { get; private set; }
    public State Cancelling { get; private set; }
    public State CompleteOk { get; private set; }
    public State CompleteError { get; private set; }
    public State CompleteCancelled { get; private set; }

    public Event<BeginProcessing> Begin { get; private set; }
    public Event<StepCompleted> StepDone { get; private set; }
    public Event<CancelProcessing> Cancel { get; private set; }
    public Event<FinalizeProcessing> Finalize { get; private set; }

    public ProcedureStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => Begin);
        Event(() => StepDone);
        Event(() => Cancel);
        Event(() => Finalize);

        BeforeEnterAny(binder => binder
            .ThenAsync(context => Console.Out.WriteLineAsync(
                $"ENTERING STATE [{context.Data.Name}]")));

        Initially(
            When(Begin)
                .Then(context =>
                {
                    context.Instance.RemainingSteps = new Queue<string>(context.Data.Steps);
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(Begin)}]: Procedure [{context.Data.ProcedureId}] Steps [{string.Join(",", context.Data.Steps)}]"))
                .Publish(context => new ExecuteStep
                {
                    ProcedureId = context.Instance.CorrelationId,
                    StepId = context.Instance.RemainingSteps.Dequeue()
                })
                .Publish(context => new SomeFunMessage
                {
                    CorrelationId = context.Data.CorrelationId,
                    TheMessage = $"Procedure [{context.Data.CorrelationId} has begun..."
                })
                .TransitionTo(Processing)
            );

        During(Processing,
            When(StepDone)
                .Then(context =>
                {
                    if (null == context.Instance.AccumulatedResults)
                    {
                        context.Instance.AccumulatedResults = new List<StepResult>();
                    }
                    context.Instance.AccumulatedResults.Add(
                        new StepResult
                        {
                            CorrelationId = context.Instance.CorrelationId,
                            StepId = context.Data.StepId,
                            WhatHappened = context.Data.WhatHappened
                        });
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(StepDone)}]: Procedure [{context.Data.ProcedureId}] Step [{context.Data.StepId}] Result [{context.Data.WhatHappened}] RemainingSteps [{string.Join(",", context.Instance.RemainingSteps)}]"))
                .If(context => !context.Instance.RemainingSteps.Any(),
                    binder => binder.TransitionTo(CompleteOk))
                .If(context => context.Instance.RemainingSteps.Any(),
                    binder => binder.Publish(context => new ExecuteStep
                    {
                        ProcedureId = context.Instance.CorrelationId,
                        StepId = context.Instance.RemainingSteps.Dequeue()
                    })),
            When(Cancel)
                .Then(context =>
                {
                    context.Instance.RemainingSteps.Clear();
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(Cancel)}]: Procedure [{context.Data.ProcedureId}] will be cancelled with following steps remaining [{string.Join(",", context.Instance.RemainingSteps)}]"))
                .TransitionTo(Cancelling)
            );

        During(Cancelling,
            When(StepDone)
                .Then(context =>
                {
                    context.Instance.SomeStringValue = "Booo... we cancelled...";
                })
                .ThenAsync(context => Console.Out.WriteLineAsync(
                    $"EVENT [{nameof(StepDone)}]: Procedure [{context.Data.ProcedureId}] Step [{context.Data.StepId}] completed while cancelling."))
                .TransitionTo(CompleteCancelled));

        During(CompleteOk, When(Finalize).Finalize());
        During(CompleteCancelled, When(Finalize).Finalize());
        During(CompleteError, When(Finalize).Finalize());

        // The "SetCompleted*" thing is what triggers purging of the state context info from the store (eg. Redis)...  without that, the 
        // old completed state keys will gradually accumulate and dominate the Redis store.
        SetCompletedWhenFinalized();
    }
}

调试此测试时，它的集合中_harness有BeginProcessing消息Sent，但集合中没有任何内容_saga.Created。似乎我缺少一些管道以使线束在发送消息时实际驱动状态机？

====

从测试中删除.Wait()调用SetUp()并将TearDown()测试更新为以下内容不会改变行为：

    [TestMethod]
    public async Task That_Can_Start()
    {
        try
        {
            await _harness.Start();
            // Arrange

            // Act
            await _harness.InputQueueSendEndpoint.Send(new BeginProcessing
            {
                ProcedureId = Guid.NewGuid(),
                Steps = new List<string> {"A", "B", "C"}
            });

            // Assert
            var sagaContext = _saga.Created.First();
            sagaContext.Saga.RemainingSteps.ShouldHaveCountOf(3);
        }
        finally
        {
            await _harness.Stop();
        }
    }
4

1 回答 1

1

_harness.InputQueueSendEndpoint.Send事实证明，如上所示的测试代码Send在StateMachineSagaTestHarness. 结果，测试代码的“断言”阶段在创建 saga 并允许处理发送的消息之前执行。

稍微深入研究一下SagaTestHarness代码，我发现了一些辅助方法，我可以使用它们来等待满足 saga 的某些条件。方法是：

/// <summary>
/// Waits until a saga exists with the specified correlationId
/// </summary>
/// <param name="sagaId"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<Guid?> Exists(Guid sagaId, TimeSpan? timeout = null)

/// <summary>
/// Waits until at least one saga exists matching the specified filter
/// </summary>
/// <param name="filter"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<IList<Guid>> Match(Expression<Func<TSaga, bool>> filter, TimeSpan? timeout = null)

/// <summary>
/// Waits until the saga matching the specified correlationId does NOT exist
/// </summary>
/// <param name="sagaId"></param>
/// <param name="timeout"></param>
/// <returns></returns>
public async Task<Guid?> NotExists(Guid sagaId, TimeSpan? timeout = null)

所以我决定使用诸如此类的东西await _saga.Match(s => null != s.RemainingSteps);来有效地复制我后来的断言，并等待超时（默认为 30 秒）或后来断言的条件变为真（因此可以安全地断言）......以先到者为准。

这会让我暂时摆脱困境，直到我想出一个更好的方法来知道安全带何时“被抓住”并准备好接受审讯。

于 2018-04-12T00:52:46.400 回答