Testing Complex Interactions With Moq Sequences
How do you test long running processes that handle random sequences of events?
As part of a recent project, I needed to create a background service for a .Net Core API. The service had to process outbox notifications of user events and post them to a message queue for consumption by other services. The task itself was straightforward but background services present a couple issues for testability, and I spent more time creating tests to prove that the service worked as expected than I did creating the service. In this article I’ll discuss the techniques I used to thoroughly test my new service.
Before diving into testing, let’s discuss the service requirements. The service should continuously poll a repository for new events and run until cancelled. During each polling pass, the service should post any events found to the message queue, notify the repository of the last successfully processed event and then wait for a period of time before starting the next pass. To avoid trying to do too much work in a single pass, the service should have a configurable batch size to process per pass. The interval to wait between polling passes should also be configurable. In the case of a full batch (i.e., the number of events returned from the repository is equal to the batch size) the service should not wait before starting the next pass. If the service is cancelled during processing a batch of user events it will immediately exit, even if that means not publishing all the events in the batch or notifying the repository of the events that were published.
The events processed by the service will have an unique integer event id and additional details about the event. This is record that we’ll use for a single event:
public sealed record UserEvent(Int64 EventId, String EventDetails);The repository will record the id of the last successfully processed event and when queried for new events will return a list of events newer than the last successfully processed event, up to a specified batch size. The repository will also provide a method to update the last successfully processed event. This is the interface for our repository:
public interface IRepository
{
Task<IReadOnlyList<UserEvent>> GetEventsAsync(Int32 batchSize);
Task<Boolean> UpdateLastProcessedEventIdAsync(Int64 eventId);
}The message queue will have a single responsibility, to publish UserEvents. The message queue will provide at least once delivery of UserEvents and we will assume that consumers of the published events will use the EventId to detect and handle duplicate events. This is the interface for our message queue:
public interface IMessageQueue
{
Task<Int64> PublishEventAsync(UserEvent userEvent);
}Finally, here is the settings object for the configurable service values:
public sealed class OutboxSettings
{
public Int32 BatchSize { get; set; }
public Int32 PollingInterval { get; set; }
}Given the above requirements and interface definitions, here is an example implementation of the service.
public sealed class OutboxProcessorService : BackgroundService
{
private readonly IRepository _repository;
private readonly IMessageQueue _messageQueue;
private readonly OutboxSettings _settings;
public OutboxProcessorService(
IRepository repository,
IMessageQueue messageQueue,
OutboxSettings settings)
{
_repository = repository;
_messageQueue = messageQueue;
_settings = settings;
}
protected override async Task ExecuteAsync(CancellationToken token)
{
while(!token.IsCancellationRequested)
{
var batch = await _repository.GetEventsAsync(_settings.BatchSize);
await PublishItemsAsync(batch, token);
if (batch.Count != _settings.BatchSize
&& !token.IsCancellationRequested)
{
await Task.Delay(_settings.PollingInterval, token);
}
}
}
private async Task PublishItemsAsync(
IReadOnlyList<UserEvent> batch,
CancellationToken token)
{
Int64 lastProcessedId = default;
foreach(var item in batch)
{
if (token.IsCancellationRequested)
{
return;
}
await _messageQueue.PublishEventAsync(item);
lastProcessedId = item.EventId;
}
if (batch.Count > 0 && !token.IsCancellationRequested)
{
await _repository.UpdateLastProcessedEventIdAsync(lastProcessedId);
}
}
}This version of the service has a couple of issues to overcome to be able test it successfully. Because ExecuteAsync is protected, we can’t invoke it directly. Instead, we must resort to spinning up the entire API and performing an integration test rather than a focused unit test. And the use of Task.Delay is an example of the Ambient Context anti-pattern. (https://freecontent.manning.com/the-ambient-context-anti-pattern/)
Refactoring for Testability
First, let’s refactor the code to make it more testable. The first refactoring is to replace the ambient Task.Delay with an injected object that wraps Task.Delay and can be mocked during testing.
public interface IAsyncDelayer
{
Task<Int32> Delay(Int32 milliseconds, CancellationToken token);
}
public sealed class AsyncDelayer : IAsyncDelayer
{
public async Task<Int32> Delay(Int32 milliseconds, CancellationToken token)
{
await Task.Delay(milliseconds, token);
return milliseconds;
}
}The second refactoring is to extract the bulk of the original service’s ExecuteAsync method into a separate object that can be tested independently of the service and its associated environment.
public interface IOutboxProcessor
{
Task ProcessOutboxItemsAsync(CancellationToken token);
}
public class OutboxProcessor : IOutboxProcessor
{
private readonly IRepository _repository;
private readonly IMessageQueue _messageQueue;
private readonly IAsyncDelayer _delayer;
private readonly OutboxSettings _settings;
public OutboxProcessor(
IRepository repository,
IMessageQueue messageQueue,
IAsyncDelayer delayer,
OutboxSettings settings)
{
_repository = repository;
_messageQueue = messageQueue;
_delayer = delayer;
_settings = settings;
}
public async Task ProcessOutboxItemsAsync(CancellationToken token)
{
while (!token.IsCancellationRequested)
{
var batch = await _repository.GetEventsAsync(_settings.BatchSize);
await PublishItemsAsync(batch, token);
if (batch.Count != _settings.BatchSize
&& !token.IsCancellationRequested)
{
await _delayer.Delay(_settings.PollingInterval, token);
}
}
}
private async Task PublishItemsAsync(
IReadOnlyList<UserEvent> batch,
CancellationToken token)
{
Int64 lastProcessedId = default;
foreach (var item in batch)
{
if (token.IsCancellationRequested)
{
return;
}
await _messageQueue.PublishEventAsync(item);
lastProcessedId = item.EventId;
}
if (batch.Count > 0 && !token.IsCancellationRequested)
{
await _repository.UpdateLastProcessedEventIdAsync(lastProcessedId);
}
}
}And finally, we have the refactored service:
public class OutboxProcessorService : BackgroundService
{
private readonly IOutboxProcessor _outboxProcessor;
public OutboxProcessorService(IOutboxProcessor outboxProcessor)
=> _outboxProcessor = outboxProcessor;
protected override async Task ExecuteAsync(CancellationToken token)
=> await _outboxProcessor.ProcessOutboxItemsAsync(token);
}Testing the OutboxProcessor
Our goal is to prove that the new OutboxProcessor class meets the requirements that we identified earlier. During each polling pass the service will find between zero and batchSize user events and we need to confirm that the message queue and the delayer are invoked as expected for each variation of (the number of events found is zero, less than batchSize or equal to batchSize). We also need to confirm that the OutboxProcessor correctly handles multiple polling passes and that when the service is canceled it shuts down as expected. So, we will focus on the following scenarios:
- An empty batch will publish no items, nor will it update the last processed item key and it will delay before checking for new items.
- A partial batch will publish all items in the batch, and it will update the last processed item key, then delay before checking for new items.
- A full batch will publish all items in the batch, and it will update the last processed item key, then it will NOT delay before checking for new items.
- Service shutdown will interrupt processing and exit cleanly. A service shutdown could occur during any of the repository or message queue methods, and we will include tests for a cancellation during getting user events from the repository, while publishing an user event to the message queue, while updating the repository, and while waiting for the next polling pass.
- For good measure, we will include a scenario that includes multiple empty, partial and full batches to simulate an extended process lifetime.
There are a couple of ways to verify that methods are invoked when using Moq. You can use the implicit Verifiable constructor the explicit Verify construct while setting up your mocks. These are sufficient for many scenarios but are difficult if not impossible to use when trying to verify multiple mocks working in concert. I’ve seen attempts to use the Callback construct to roll your own verification by capturing the calls in a list or other store and then examining the list after the method under test is run. Ugh. Fortunately, Moq has two features that are perfect for our needs — MockBehavior.Strict and MockSequence.
MockBehavior.Strict is used in the mock constructor to indicate that every invocation of the mock must have a corresponding setup. And MockSequence allows you to setup a sequence of invocations that can span multiple mocked objects. If the sequence of calls made during the test deviates from those in the mock setups then a Moq.Exception will be thrown.
We start by creating a new test class and setting up a few values that are used in multiple tests. We have an instance of the OutBoxSettings that all tests will use and an array of UserEvent objects for cases where the repository returns one or more events and an empty list of UserEvent objects for tests involving empty batches.
public class OutboxProcessorTests
{
private readonly OutboxSettings _outboxSettings = new()
{
BatchSize = 3,
PollingInterval = 30000
};
private readonly UserEvent[] _outboxEvents = Enumerable.Range(1, 10)
.Select(x => new UserEvent(x, $"Details for user event {x}"))
.ToArray();
private readonly IReadOnlyList<UserEvent> _emptyBatch = new List<UserEvent>();
}Now, let’s look at how to use a sequence to test the first scenario.
[Fact]
public async Task ProcessOutboxItemsAsync_PublishesNoItemsAndDelays_WhenEmptyBatchEncountered()
{
// Arrange.
var cancellationTokenSource = new CancellationTokenSource();
var token = cancellationTokenSource.Token;
var repositoryMock = new Mock<IRepository>(MockBehavior.Strict);
var messageQueueMock = new Mock<IMessageQueue>(MockBehavior.Strict);
var delayerMock = new Mock<IAsyncDelayer>(MockBehavior.Strict);
var sequence = new MockSequence();
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_emptyBatch);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.Callback(() => cancellationTokenSource.Cancel())
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
var sut = new OutboxProcessor(
repositoryMock.Object,
messageQueueMock.Object,
delayerMock.Object,
_outboxSettings);
// Act.
await sut.ProcessOutboxItemsAsync(token);
// Assert - assert not needed because of MockBehavior.Strict
}The Arrange section is most of the test. First, we create the CancellationTokenSource that will be used to control the lifetime of the test. Mocks for the repository, message queue and delayer are created, with MockBehavior.Strict. Then we create the MockSequence.
Next, we set up our expectations of the exact sequence of method invocations that should occur, along with expected parameter values. Note that here we are using the exact expected parameter values. If you don’t need to validate the exact parameter value, you could substitute It.IsAny<T>() or similar methods for the parameter value. We’re using the normal Moq Setup syntax, but we include the InSequence construct to add the setup to the sequence for the test. Note that we also simulate shutdown of the process by including a Callback in the delayer setup that invokes the Cancel method of the CancellationTokenSource object.
Finally, we create our System Under Test (sut) object and invoke the method we are testing. There is no need to assert anything because MockBehavior.Strict will cause an exception to be thrown if the actual sequence of method invocations deviates from the setup.
Now let’s create the test for the partial batch scenario. It follows the same pattern as the previous test, just with additional setups to cover publishing the batch items and updating the last processed item key.
[Fact]
public async Task ProcessOutboxItemsAsync_PublishesBatchItemsAndDelays_WhenPartialBatchEncountered()
{
// Arrange.
var cancellationTokenSource = new CancellationTokenSource();
var token = cancellationTokenSource.Token;
var repositoryMock = new Mock<IRepository>(MockBehavior.Strict);
var messageQueueMock = new Mock<IMessageQueue>(MockBehavior.Strict);
var delayerMock = new Mock<IAsyncDelayer>(MockBehavior.Strict);
var sequence = new MockSequence();
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_outboxEvents[0..2].ToList());
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[0]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[1]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
repositoryMock.InSequence(sequence)
.Setup(x => x.UpdateLastProcessedEventIdAsync(_outboxEvents[1].EventId))
.ReturnsAsync(true);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.Callback(() => cancellationTokenSource.Cancel())
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
var sut = new OutboxProcessor(
repositoryMock.Object,
messageQueueMock.Object,
delayerMock.Object,
_outboxSettings);
// Act.
await sut.ProcessOutboxItemsAsync(token);
}Once the pattern is established you can stamp out tests for the other scenarios simply by altering the sequence of setups. You can find the full set of tests in the GitHub repository that contains the code in this article. Finally, we arrive at this extended scenario that tests the handling of multiple batches.
[Fact]
public async Task ProcessOutboxItemsAsync_PublishesMultipleBatches_WhenHandlingLargeEventStream()
{
// Arrange.
var cancellationTokenSource = new CancellationTokenSource();
var token = cancellationTokenSource.Token;
var repositoryMock = new Mock<IRepository>(MockBehavior.Strict);
var messageQueueMock = new Mock<IMessageQueue>(MockBehavior.Strict);
var delayerMock = new Mock<IAsyncDelayer>(MockBehavior.Strict);
var sequence = new MockSequence();
// First pass, empty batch.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_emptyBatch);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
// Second pass, full batch.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_outboxEvents[0..3].ToList());
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[0]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[1]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[2]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
repositoryMock.InSequence(sequence)
.Setup(x => x.UpdateLastProcessedEventIdAsync(_outboxEvents[2].EventId))
.ReturnsAsync(true);
// Third pass, partial batch.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_outboxEvents[3..4].ToList());
messageQueueMock.InSequence(sequence)
.Setup(x => x.PublishEventAsync(_outboxEvents[3]))
.ReturnsAsync((UserEvent evt) => evt.EventId);
repositoryMock.InSequence(sequence)
.Setup(x => x.UpdateLastProcessedEventIdAsync(_outboxEvents[3].EventId))
.ReturnsAsync(true);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
// Fourth pass, empty batch.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_emptyBatch);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
// Fifth pass, empty batch.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.ReturnsAsync(_emptyBatch);
delayerMock.InSequence(sequence)
.Setup(x => x.Delay(_outboxSettings.PollingInterval, token))
.ReturnsAsync((Int32 interval, CancellationToken token) => interval);
// Sixth pass, cancel and exit.
repositoryMock.InSequence(sequence)
.Setup(x => x.GetEventsAsync(_outboxSettings.BatchSize))
.Callback(() => cancellationTokenSource.Cancel())
.ReturnsAsync(_outboxEvents[4..7].ToList());
var sut = new OutboxProcessor(
repositoryMock.Object,
messageQueueMock.Object,
delayerMock.Object,
_outboxSettings);
// Act.
await sut.ProcessOutboxItemsAsync(token);
}Whew! That’s an uncomfortably large test and it’s almost entirely setup code. On the plus side, these tests do provide good coverage of the OutboxProcessor class and prove that it does work as expected. But while it was easy to stamp out different sequences of events when creating multiple tests one after the other, the verbosity of MockBehavior.Strict quickly spirals out of control, as demonstrated by the last test. The result is a dense block of setup code in each test. That’s not something that I’d want to leave for a future developer try to read or maintain. We can do better and in my next article I’ll show you how to improve on the tests, making the setup much cleaner and easier to follow and maintain.
I wouldn’t advocate using sequences or MockBehavior.Strict for all of the tests you create. But I do think that under certain circumstances this is a valuable technique to be aware of.
Thanks for reading!
About Me
I’m just Some Random Programmer Guy, at least according to the nameplate that was once waiting for me on my first day at a startup. I’ve been around a while and my career has spanned FORTRAN and PL/1 on IBM mainframes to .Net Core microservices with some interesting forays with C/C++, OS/2, VB, C#, SQL, NoSQL and more along the way.
Code for this article is available my public github repository.
