Automate integration tests in Azure Data Factory

In my previous post, I set up and ran one basic test of a single pipeline. In this article, I refactor my VS testing solution to make it easier to add new tests and to test new pipelines.

The helper class I introduced in the last article does a number of different things, and does them in a very specific way:

• uses system settings from environment variables
• connects to a named data factory instance
• runs a named ADF pipeline.

This allows me to run a basic test, but it's not very reusable. It also mixes a number of different features together in one class, which isn't very convenient (and isn't great design). I'm going to separate the class into a number of related helper classes to make this easier to manage.

My original helper class used environment variables to hold secret settings for connecting to Azure, but I also want to be able to:

• store and access non-sensitive settings for my testing setup
• access secrets held in my Azure Key Vault.

A convenient way to store non-sensitive settings for NUnit is to store them in a runsettings file. A runsettings file contains XML describing test parameters as a collection of name/value pairs:

<?xml version='1.0' encoding='utf-8'?>
<RunSettings>
    <TestRunParameters>
        <Parameter name="KeyVaultUrl" value="https://firefive-adftest95-kv.vault.azure.net" />
        <Parameter name="DataFactoryResourceGroup" value="firefive-adftest95-rg" />
        <Parameter name="DataFactoryName" value="firefive-adftest95-adf" />
    </TestRunParameters>
</RunSettings>

You'll notice I've included the URL of my key vault as the first non-sensitive setting.

To specify the runsettings file for Visual Studio:

• save it somewhere convenient with a name ending in “.runsettings” – I'm using “tests.runsettings”

• configure it for the solution in Visual Studio, by clicking Test → Configure Run Settings → Select Solution Wide runsettings File, then choose the file you saved.

  

Microsoft provides a .NET API for connecting to an Azure Key Vault. This needs the URL of the key vault (which I've already stored in my runsettings file) and an object to provide authentication credentials. I'm going to use a DefaultAzureCredential object, like this:

var myCredential = new DefaultAzureCredential();

What's a default credential?! The DefaultAzureCredential class uses three environment variables to authenticate against Azure, which is why I don't need to specify any in the code:

• AZURE_TENANT_ID
• AZURE_CLIENT_ID
• AZURE_CLIENT_SECRET

I set these up in the previous post, so I'm good to go.

Bringing this together, my settings helper class looks like this:

public class SettingsHelper
{
    public string GetSetting(string settingName)
    {
        // return environment variable "settingName", if present
        var value = Environment.GetEnvironmentVariable(settingName);
        if (value?.Length > 0)
            return value;
 
        // return value of runsettings parameter "settingName", if present
        value = TestContext.Parameters[settingName];
        if (value?.Length > 0)
            return value;
 
        // if a key vault is specified, return the value of secret "settingName", if present
        if (_keyVaultClient != null)
        {
            value = _keyVaultClient.GetSecret(settingName).Value.Value;
            if (value?.Length > 0)
                return value;
        }
 
        throw new Exception($"Test setting '{settingName}' not found");
    }
 
    private readonly SecretClient _keyVaultClient;
 
    public SettingsHelper()
    {
        var kvUrl = TestContext.Parameters["KeyVaultUrl"];
        if (kvUrl?.Length > 0)
            _keyVaultClient = new SecretClient(new Uri(kvUrl), new DefaultAzureCredential());
    }
}

The core of the class is the GetSetting() method. It takes a settingName parameter, then looks for a setting of that name in three different places:

• local environment variables
• my runsettings file – the collection of name/value pairs is accessed conveniently in code using TestContext.Parameters
• my key vault.

I check the key vault last because it takes the most work to check, and because if a secret named settingName isn't found I can allow the resulting exception to be thrown from the test.

My data factory helper class looks very similar to the data factory code I used in the last post. It's a subclass of SettingsHelper (line 1), so I can continue to access all helper functionality in the test through a single helper object.

public class DataFactoryHelper : SettingsHelper
{
    public string PipelineOutcome { get; private set; }
 
    public async Task RunPipeline(string pipelineName)
    {
        PipelineOutcome = "Unknown";
 
        // authenticate against Azure
        var context = new AuthenticationContext("https://login.windows.net/" + GetSetting("AZURE_TENANT_ID"));
        var cc = new ClientCredential(GetSetting("AZURE_CLIENT_ID"), GetSetting("AZURE_CLIENT_SECRET"));
        var authResult = await context.AcquireTokenAsync("https://management.azure.com/", cc);
 
        // prepare ADF client
        var cred = new TokenCredentials(authResult.AccessToken);
        using (var adfClient = new DataFactoryManagementClient(cred) { SubscriptionId = GetSetting("AZURE_SUBSCRIPTION_ID") })
        {
            var adfName = GetSetting("DataFactoryName");
            var rgName = GetSetting("DataFactoryResourceGroup");
 
            // run pipeline
            var response = await adfClient.Pipelines.CreateRunWithHttpMessagesAsync(rgName, adfName, pipelineName);
            string runId = response.Body.RunId;
 
            // wait for pipeline to finish
            var run = await adfClient.PipelineRuns.GetAsync(rgName, adfName, runId);
            while (run.Status == "Queued" || run.Status == "InProgress" || run.Status == "Canceling")
            {
                Thread.Sleep(2000);
                run = await adfClient.PipelineRuns.GetAsync(rgName, adfName, runId);
            }
            PipelineOutcome = run.Status;
        }
    }
 
    public DataFactoryHelper()
    {
        PipelineOutcome = "Unknown";
    }
}

It has some important differences:

• RunPipeline() (line 5) now requires a pipelineName parameter – I can use it to run any pipeline I choose.
• All references to environment variables (lines 10, 11, 16) are now made via the GetSetting() method. If I forget to set a variable, I'll get a more helpful exception message from SettingsHelper.
• The data factory and resource group names are also accessed via GetSetting() (lines 18, 19). If I want to point this test suite at a different ADF instance, all I need to do is edit the runsettings file.

In the same way that the settings helper is only concerned with providing access to settings, the data factory helper is now only concerned with running ADF pipelines in general. This allows me to share its functionality between tests of different pipelines.

The pipeline helper still exists – it subclasses DataFactoryHelper and manages pipeline-specific information:

public class PLStageAuthorsHelper : DataFactoryHelper
{
    public async Task RunPipeline()
    {
        await RunPipeline("PL_Stage_Authors");
    }
}

At this stage, all it's doing is supplying the pipeline name to the data factory helper.

NUnit refers to a class containing tests as a test fixture. I've been careful to reorganise my helper class without changing my Given23Rows test fixture at all – it's still simple, still readable.

I'm going to make one change anyway, because I don't like Assert.AreEqual() – it's core NUnit syntax, but remember I want my tests to be as readable as possible. I prefer to use FluentAssertions, a set of extension methods that allows me to specify expected test outcomes much more naturally. Here's my original test:

[Test]
public void ThenPipelineOutcomeIsSucceeded()
{
    Assert.AreEqual("Succeeded", _helper.PipelineOutcome);
}

And here's the same test, re-written using a fluent assertion:

[Test]
public void ThenPipelineOutcomeIsSucceeded()
{
    _helper.PipelineOutcome.Should().Be("Succeeded");
}

I've organised the changes in my Visual Studio solution like this:

• General helper classes (not pipeline-specific helpers) are in the “Helpers” folder
• the “Pipelines” folder will contain pipeline-specific subfolders – “PL_Stage_Authors” contains the test fixtures and helper class belonging to my ADF pipeline of the same name.

I can run this test in the same way as before and get exactly the same result. So what's the benefit? To start with, separating helper behaviour out into layers like this makes it easier to slot in other pieces of functionality. I'm going to do that now for my next test: checking the number of rows that were staged by the pipeline.

I need to make a few changes to accomplish this:

• add a new test to Given23Rows
• add a StagedRowCount property to the pipeline helper class PLStageAuthorsHelper
• add a way to get the staged row count from the database.

This is the new test in Given23Rows, also written using a fluent assertion:

[Test]
public void Then23RowsAreStaged()
{
    _helper.StagedRowCount.Should().Be(32);
}

Here's PLStageAuthorsHelper updated with the new StagedRowCount property:

public class PLStageAuthorsHelper : DatabaseHelper
{
    public async Task RunPipeline()
    {
        await RunPipeline("PL_Stage_Authors");
    }
 
    public int StagedRowCount
    {
        get
        {
            return RowCount("stg.Authors");
        }
    }
}

I think that counting rows is likely to be generally useful for testing, so I haven't queried the database here. Instead I've called a RowCount() method (line 12), passing the parameter value “stg.Authors” – recall this is the name of the table where the pipeline writes its output.

I haven't defined RowCount() yet, but the code contains a clue about where I'm going to put it – the pipeline helper is no longer a subclass of the data factory helper (line 1) but of a new class, DatabaseHelper.

The new database helper class has a single method, RowCount(), which takes a tableName parameter. The method returns the number of rows in the specified table:

public class DatabaseHelper : DataFactoryHelper
{
    public int RowCount(string tableName)
    {
        using (var conn = new SqlConnection(GetSetting("AdfTestingDbConnectionString")))
        {
            conn.Open();
            using (var cmd = new SqlCommand($"SELECT COUNT(*) FROM {tableName}", conn))
            using (var reader = cmd.ExecuteReader())
            {
                reader.Read();
                return reader.GetInt32(0);
            }
        }
    }
}

The first line in the method body (line 5) creates a connection to a database. Which one? I'm using GetSetting() to retrieve the connection string for the [AdfTesting] DB – recall in the first post I said that this was stored in the Key Vault for ADF to use, so I'm going to take advantage of that.

The database helper subclasses the data factory helper, so my helper inheritance hierarchy now looks like this:

I hit the “Run All Tests” button in Test Explorer and… whoops!

Then23RowsAreStaged has failed because the row count wasn't what the test expected. You might have seen this coming when you looked at the new test definition – I specified the test incorrectly, by expecting 32 rows instead of 23. Another reason a test can fail is that it's badly-written!

Look closely at the output from the test run above, and you'll notice that the two tests each took ten seconds to execute. In the last post I mentioned that NUnit runs a fixture's [SetUp] method before each [Test] method is called – each test took so long here because the pipeline was executed once before each test was run, two times in total.

Running [SetUp] before each [Test] helps to ensure that tests are isolated from one another. My approach has been to use [SetUp] to run an ADF pipeline, then to check a feature of its outcome in [Test] – so repeated calls to [SetUp] don't make my tests any more isolated than they already are. From now on, I'm going to run test setup (and the ADF pipeline) only once per fixture, before any of the tests are run. I do this by changing the WhenPipelineIsRun() method's attribute to [OneTimeSetUp]:

[OneTimeSetUp]
public async Task WhenPipelineIsRun()
{
    _helper = new PLStageAuthorsHelper();
    await _helper.RunPipeline();
}

Now the method will only be called once, followed by each of the [Test] methods in the class. Tests will be faster (and cheaper) to run, because I'm triggering fewer ADF pipeline runs.

This test is still not isolated from external dependencies. In fact, I've made the situation worse, because as soon as the source [dbo].[Authors] table has a new row inserted, my row count test will fail. Even if the data doesn't change often, I still have no control over it, so the new test is very fragile.

My two tests are really of different types:

• ThenPipelineOutcomeIsSucceeded is a simple integration test – although it doesn't seem to do much, it shows that the pipeline can connect to the source and staging databases, that the source and target tables exist and that ADF has permission to read and write to them.
• Then23RowsAreStaged is an attempt at a functional test. In the previous post I described a functional test as an isolated test of whether the pipeline is doing things right. Checking that it stages 23 rows is one test that it's “doing things right”, but its setup isn't isolated at all.

What makes each of these either an integration test or a functional test isn't just a question of the [Test] method – it's a combination of the test and its [OneTimeSetUp]. My attempt at a functional test is fragile because I'm setting it up like an integration test. I'll talk about how to set up functional tests in my next post.

Now I have some reusable helpers, I'm in good shape to start testing a second ADF pipeline. I have another pipeline called “PL_Stage_Titles” – it does the same thing as “PL_Stage_Authors”, but for source table [dbo].[Titles], and it has some additional logging behaviour:

• “Log pipeline start” is a Lookup activity which calls a database stored procedure to make an entry in a log table and return an integer run ID.
• The returned run ID is stored in a RunId pipeline variable by “Set RunId”, a Set variable activity.
• The Copy data activity is very similar to the one in “PL_Stage_Authors”, but adds the RunId variable as a source column for insertion into [stg].[Titles] (boxed in red on the screenshot).
• The “Log pipeline end” activity calls another stored procedure to update the log table.

The refactored helper structure makes it easier and faster for me to write new tests. To start testing “PL_Stage_Titles” quickly, I'm going to copy the integration test setup I already wrote for the “PL_Stage_Authors” pipeline. I need to:

• copy and modify the pipeline helper class
• copy and modify the test fixture.

I'll create the test fixture and helper in a new “PL_Stage_Titles” subfolder of my Visual Studio “Pipelines” folder.

I copy “PLStageAuthorsHelper.cs” from folder “PL_Stage_Authors” and into “PL_Stage_Titles”, then make the following changes:

• modify the namespace to PL_Stage_Titles (line 1)
• change the class name (and the name of the .cs file itself) to PLStageTitlesHelper (line 3)
• revise the RunPipeline() method to run ADF pipeline “PL_Stage_Titles” (line 7)
• revise the StagedRowCount property to count rows in “stg.Titles” (line 14).

namespace PL_Stage_Titles
{
    public class PLStageTitlesHelper : DatabaseHelper
    {
        public async Task RunPipeline()
        {
            await RunPipeline("PL_Stage_Titles");
        }
 
        public int StagedRowCount
        {
            get
            {
                return RowCount("stg.Titles");
            }
        }
    }
}

I copy “Given23Rows.cs” from folder “PL_Stage_Authors” and into “PL_Stage_Titles”, then make the following changes:

• modify the namespace to PL_Stage_Titles (line 1)
• change the class name (and the name of the .cs file itself) to GivenExternalDependencies to reflect that this is an integration test (line 3)
• change the type of the _helper instance variable to PLStageTitlesHelper (line 5)
• remove method Then23RowsAreStaged – a row count check is more like a functional test, so isn't compatible with this this fixture's [OneTimeSetUp].

namespace PL_Stage_Titles
{
    public class GivenExternalDependencies
    {
        private PLStageTitlesHelper _helper;
 
        [OneTimeSetUp]
        public async Task WhenPipelineIsRun()
        {
            _helper = new PLStageTitlesHelper();
            await _helper.RunPipeline();
        }
 
        [Test]
        public void ThenPipelineOutcomeIsSucceeded()
        {
            _helper.PipelineOutcome.Should().Be("Succeeded");
        }
    }
}

I hit the “Run All Tests” button in Test Explorer. You'll notice that Test Explorer now shows the two test fixtures, grouped by namespace:

I still haven't fixed the bad test definition in Given23Rows – it's a functional test set up like an integration test so has bigger problems! I'll replace it in the next post.

In this article I refactored the basic test structure from the previous post. I separated helper functionality into smaller classes which are easier to understand, maintain and share between pipelines. Sharing functionality between tests of different pipelines makes it easier and faster to write new tests.

Using [OneTimeSetUp] makes my tests faster and cheaper to run, because it requires fewer ADF pipeline runs. This is sufficient to isolate tests from one another because I'm running ADF pipelines during test setup. The test setup I've been using here is suitable for integration testing – in the next post I'll look at pipeline isolation for reliable functional tests.