How to successfully drive a MassTransitStateMachine via the InMemoryTestHarness?

Question

Following up to: How to write MassTransitStateMachine unit tests?

Here's a simple test class (using MS Test) for a simple state machine called ProcedureStateMachine (note: this is not a real production state machine for us... just an experiment I'd used to play around with MassTransitStateMachine a while back.. it seemed a handy self-contained place to experiment with getting unit testing going too):

[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);
    }
}

And here's the state machine class itself:

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();
    }
}

When debug this test, the _harness has the BeginProcessing message in its Sent collection, but there's nothing in the _saga.Created collection. It seems like I'm missing some plumbing to cause the harness to actually drive the state machine when the messages are sent?

====

Removing the .Wait() calls from SetUp() and TearDown() and updating the test to the following does NOT change the behavior:

    [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();
        }
    }

Answer 1

It turns out that the test code as shown above was suffering from a race condition between the _harness.InputQueueSendEndpoint.Send operation and some asynchronous (beyond what await on the Send waits for) behavior in the StateMachineSagaTestHarness. As a result, the "Assert" phase of the test code was executing before the saga had been created and allowed to handle the sent message.

Digging into the SagaTestHarness code a bit, I found a few helper methods that I was able to use to wait until certain conditions on the saga are met. The methods are:

/// <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)

So I settled on using things like await _saga.Match(s => null != s.RemainingSteps); and such to effectively duplicate my later asserts and wait either until the timeout (default is 30 seconds) or the later-asserted condition has become true (and therefore safe to Assert against).. whichever comes first.

This will get me unstuck for now until I can think of a better way to know when the harness is "caught up" and ready to be interrogated.