Unit Testing The Azure Cosmos DB Change Feed In xUnit And C#

Writing basic units for Azure Functions triggered by the Change Feed is straightforward with xUnit.

Introduction 

 

Writing basic units for Azure Functions triggered by the Change Feed is straightforward with xUnit

 

 

While refactoring some of our microservices at work, I came across a service that didn’t have any unit tests for them! This service uses the Azure Cosmos DB Change Feed to listen to one of our write-optimized containers related to customers. If a new customer is created in that container, we then pick up that Customer document and insert it into a read-optimized container (acting as an aggregate store) which has a read friendly partition key value.

 

This read-optimized container is then utilized by other services within our pipeline when we need to query the aggregate for our Customer data.

Most of our services are triggered by message brokers (Event Hubs, Service Bus, etc.) so our process for unit testing these services is pretty standardized. But for whatever reason, this service didn’t have any unit tests, which is pretty bad in my opinion. So I’d thought I’d have a crack at it.

 

Turns out, it’s fairly straightforward.

 

So in this article, I’m going to show you how straightforward it is to write basic unit tests for an Azure Function that is triggered by the Azure Cosmos DB Change Feed using xUnit. This is the unit testing framework we use at work and I also use it in my side projects as well since it’s free, open-source, and super easy to get your head around.

 

Wait, What is the Change Feed again?

 

The Azure Cosmos DB Change Feed is a persistent record of changes that take place in a container in the order that they occur. It listens to any changes in a container and then outputs a sorted list of documents that were changed in the order in which they were modified.

 

Let’s dive in!

 

Let’s imagine that we’re working with our Customer containers, we have an Azure Function that uses a CosmosDB trigger to listen to changes within our container and inserts the documents into our read-optimized containers. It might look something like this:

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using PizzaParlour.Core.Models;
using PizzaParlour.CustomerManager.Aggregate.Repositories;

namespace PizzaParlour.CustomerManager.Aggregate.Functions
{
    public class CustomerFeed
    {
        private readonly ILogger<CustomerFeed> _logger;
        private readonly ICustomerAggregateRepository _customerAggregateRepository;

        public CustomerFeed(
            ILogger<CustomerFeed> logger,
            ICustomerAggregateRepository customerAggregateRepository)
        {
            _logger = logger;
            _customerAggregateRepository = customerAggregateRepository;
        }

        [FunctionName(nameof(CustomerFeed))]
        public async Task Run([CosmosDBTrigger(
            databaseName: "PizzaParlourDB",
            collectionName: "Customers",
            ConnectionStringSetting = "CosmosDBConnectionString",
            LeaseCollectionName = "leases",
            CreateLeaseCollectionIfNotExists = true,
            LeaseCollectionPrefix = "Customers")]IReadOnlyList<Document> input)
        {
            try
            {
                if (input != null && input.Count > 0)
                {
                    foreach (var document in input)
                    {
                        var customer = JsonConvert.DeserializeObject<Customer>(document.ToString());

                        await _customerAggregateRepository.UpsertCustomer(customer);
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError($"Exception thrown: {ex.Message}");
                throw;
            }
        }
    }
}

 This function does the following:

• Listens to the ‘Customers’ container inside the ‘PizzaParlourDB’ database.
• Creates a lease collection if it doesn’t exist. This controls the checkpoint of our Change Feed.
• Creates a list of documents. This will be ordered in the order that the documents were modified.
• If there are documents in the list, the Function iterates through each document, attempts to deserialize the document into a Customer object, and then upserts that Customer into our read-optimized store.

This is a much-simplified version of the service that we developed at work (not going to share that with you, sorry?). But since we’re working with a simple function here, we can write a simple unit test for it.

 

Writing our Change Feed Unit Test

 

Let’s write up a simple unit test for our Change Feed function. We could write the following:

using Microsoft.Azure.Documents;
using Microsoft.Extensions.Logging;
using Moq;
using Newtonsoft.Json;
using PizzaParlour.Core.Models;
using PizzaParlour.CustomerManager.Aggregate.Functions;
using PizzaParlour.CustomerManager.Aggregate.Repositories;
using PizzaParlour.CustomerManager.Aggregate.UnitTests.Helpers;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;
using Xunit;

namespace PizzaParlour.CustomerManager.Aggregate.UnitTests.FunctionTests
{
    public class CustomerFeedShould
    {
        private Mock<ILogger<CustomerFeed>> _loggerMock;
        private Mock<ICustomerAggregateRepository> _customerAggregateRepoMock;

        private CustomerFeed _func;

        public CustomerFeedShould()
        {
            _loggerMock = new Mock<ILogger<CustomerFeed>>();
            _customerAggregateRepoMock = new Mock<ICustomerAggregateRepository>();

            _func = new CustomerFeed(
                _loggerMock.Object,
                _customerAggregateRepoMock.Object);
        }

        [Fact]
        public async Task UpsertNewDocument()
        {
            // Arrange
            var documentList = new List<Document>();
            var testCustomer = TestDataGenerator.GenerateCustomer();
            var customerDocument = ConvertCustomerObjectToDocument(testCustomer);
            documentList.Add(customerDocument);

            // Act
            await _func.Run(documentList);

            // Assert
            _customerAggregateRepoMock.Verify(
                r => r.UpsertCustomer(It.IsAny<Customer>()), Times.Once);
        }

        private Document ConvertCustomerObjectToDocument(Customer customer)
        {
            var customerJSON = JsonConvert.SerializeObject(customer);
            var document = new Document();
            document.LoadFrom(new JsonTextReader(new StringReader(customerJSON)));

            return document;
        }
    }
}

Let’s step through our UpsertNewDocument() test,

 

First, we create a new list of Documents. This will be our list that we use to invoke our Function. We then want to generate a test customer to add to our list. Before we can add this test customer to our list, we’ll need to convert it to a Document.

 

For this purpose, I’ve got a private method that takes a Customer object, serializes it to JSON, and then loads that JSON object into our Document and returns it.

 

Once our customer has been converted into a Document type, we then add it to our list. We then pass our list with our Customer document to invoke the function.

 

In my function, I have a Repository class that takes care of the Upsert logic for me, so I’m just verifying that the Repository method fires at least once, since I only have one Customer document in my list.

 

Conclusion

 

This was a very basic example of how we can write unit tests for our Change Feed functions. Depending on how you are using and configuring Change Feed functions, your tests might need to be more comprehensive than this basic sample.

 

Hopefully, this gives you some guidance on how you can write Unit tests for Azure Functions that are triggered by the Change Feed.

The code shown is taken from this sample on my GitHub if you wanted to refer to it.

 

If you have any questions, please feel free to comment below or reach out to me on Twitter.