Integration Tests with CDK and Python - is my cloud native app doing what I want?

CDK is a great way to setup your infrastructure, but all too often you deploy your new services and then need to go "click", "click", "click". But, we're all developers who like automated tests, and hate click-ops.

Here, I will show you how to build an AWS Lambda service and use the command line to run an automated test which will:

1. Deploy an AWS service (in 3 regions concurrently!)
2. Invoke the lambda
3. Assert that the response matches what you expect
4. Destroy the service and clean up.

That all sounds like a slow process, but the testing tool (integ-runner) creates snapshots of your stacks and only tests those which have changed.

We show how to test a lambda deployment, but you can use this approach to:

• Check that you can (or cannot - because of permissions) get and put objects to an S3 bucket
• Call an API gateway or Load Balancer to test your web service running in Fargate
• Put a message into an SQS queue and wait for a timeout to check some async process completed.

You can see the complete code for this on GitHub - https://github.com/tomharvey/integration-tests-with-CDK-and-python/tree/part-one

Test first

We will write the test first and take the quickest path to making that test pass.

If you're new to CDK and python, see my blog post here for getting a running development environment. If you follow that post, you'll be ready to start.

1. Setup a new cdk project

This is all python, so cdk init --language python will bootstrap a new CDK project for us.

And, you'll want to install integ-runner with npm install -g @aws-cdk/integ-runner.

Warning - this “integ-runner” tool is in developer preview mode! So it will change and there are rough edges. Some very rough ones. But, please use the comments if you're following along and something stops working.

Finally, add aws_cdk.integ_tests_alpha to the requirements-dev.txt file and install all the requirements:

pip install -r requirements.txt
pip install -r requirements-dev.txt

2. Create your test file

CDK has bootstrapped a lot of directories and files for us. Including a tests directory. How kind.

That seems like a nice place to put your integration tests. But integ-runner expects them to be in a test (singular) directory. How irritating (the rough edges get rougher, don't worry). And, while we can tell integ-runner to use any directory, let's go ahead and stick with the defaults.

So, make a test folder with mkdir test in the terminal, or however you like to create new tests in your project.

And create a file in there called integ_.hello.py to create a new test file for the "hello world" Lambda we will create.

Use the terminal touch test/integ_.hello.py or however you like to add files to your project.

It's important that this file:

1. Starts with integ_.
2. Ends with .py

So be careful how you name these files. But anything can go between the integ_. and the .py

3. Write a CDK app and stack

We're going to write this directly in the test file.

You might be used to writing the app in app.py and creating your stacks elsewhere. Maybe even building re-usable constructs, but we will leave that tidy approach to another day. This is about getting quickly to a passing test.

So, below is a single file with the:

1. App
2. Stack
3. Test deployment of your services to AWS
4. Instruction to invoke the lambda
5. Assertion of the response

We will go though this file line-by-line at the end of the article. But, right now, it's a race to get a passing test.

import aws_cdk as cdk
from aws_cdk import (
    Stack,
    aws_lambda,
    integ_tests_alpha,
)
from constructs import Construct


class HelloTestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        self.function = aws_lambda.Function(
            self,
            "MyFunction",
            runtime=aws_lambda.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=aws_lambda.Code.from_asset("./lambda"),
        )


app = cdk.App()
stack = HelloTestStack(app, "HelloTestStack")

integration_test = integ_tests_alpha.IntegTest(app, "Integ", test_cases=[stack])

function_invoke = integration_test.assertions.invoke_function(
    function_name=stack.function.function_name,
)

function_invoke.expect(
    integ_tests_alpha.ExpectedResult.object_like(
        {"StatusCode": 200, "Payload": '"Hello world"'}
    )
)

app.synth()

And you will also need your lambda function code in a folder.

You can create this folder and file in the terminal with:

mkdir test/lambda
touch test/lambda/index.py

Or however you like to add folders and files to your project. But, yes, we're going to create our lambda code package inside the test folder for now. There is a "rough edge" otherwise, and we're aiming for tests passing.

And in that file in test/lambda/index.py you will want:

# test/lambda/index.py
def handler(event, context):
    return "Hello world"

Which is a lambda function, which, when invoked will return "Hello world".

4. Run your test for the first time

Now, you can run integ-runner to run the test, and we will get our first failure:

Verifying integration test snapshots...

  NEW        integ_.hello 5.812s

This is an expected error.

Because it takes time to deploy services and run them, integ-runner will create "snapshots" of your stacks. It saves what your stack looks like, so when you next run the test it knows if anything has changed.

But the first time you run it - there is no snapshot.

5. Really run your test

You want to use an additional option to tell integ-runner to go ahead and test any changed, or new stacks:

integ-runner --update-on-failed

This will take a few minutes...

You'll see that "NEW" error that you saw above, but then:

Running integration tests for failed tests...
Running in parallel across regions: us-east-1, us-east-2, us-west-2

Which is cool - it's deploying your lambda to 3 regions and running the tests in all 3 regions at the same time.

You can change the regions that you want to use. But let’s leave that for another time.

You can also change the props passed into your stack. So you could write tests to deploy your lambda with python version 3.12 as well as 3.11 and test things before you deploy updated 3.12 to production. Or ARM alongside x86. But, not today. We're going for the quickest route to a passing test.

Once we have a green test we can refactor and do more complex things.

After about 3 or 4 min you should get a passing result! Looking like the below

Verifying integration test snapshots...

  NEW        integ_.hello 5.715s

Snapshot Results: 

Tests:    1 failed, 1 total
Failed: /workspaces/py-integ-runner/test/integ_.hello.py

Running integration tests for failed tests...

Running in parallel across regions: us-east-1, us-east-2, us-west-2
Running test /workspaces/python-aws-cdk-integration-test-playground/py-integ-runner/test/integ_.hello.py in us-east-1
  SUCCESS    integ_.hello-Integ/DefaultTest 189.601s
       AssertionResultsLambdaInvoke24cf31b9b6a07a940ece1b49bb7eb7b2 - success

Test Results: 

Tests:    1 passed, 1 total

Yay!

What is the test file doing?

First up is the Stack that you would create. Here, it just defines a single lambda, but in a real world example it could define a lot more.

Note how the Code.from_asset sets the path to the lambda code files (‘./lambda’) and how the handler specifies that it should execute the function called “handler” inside the file called “index.py”.

class HelloTestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        self.function = aws_lambda.Function(
            self,
            "MyFunction",
            runtime=aws_lambda.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=aws_lambda.Code.from_asset("./lambda"),
        )

Next, we instantiate the App and the Stack. This is usually in app.py on the root of your CDK project. But we need a separate App for this testing process.

app = cdk.App()
stack = HelloTestStack(app, "HelloTestStack")

Then, we define a Test, and one single test case (this is where you could otherwise pass multiple test cases, where your lambda would have different python versions, architectures, or entirely different domain code).

integration_test = integ_tests_alpha.IntegTest(app, "Integ", test_cases=[stack])

We tell the Test what actions we want to take on each test_case stack. Here, we invoke the lambda.

function_invoke = integration_test.assertions.invoke_function(
    function_name=stack.function.function_name,
)

And lastly (almost), we assert that the response is as expected.

function_invoke.expect(
    integ_tests_alpha.ExpectedResult.object_like(
        {"StatusCode": 200, "Payload": '"Hello world"'}
    )
)

Finally - for real - and this is often missing from the IntegTest documentation, you need to app.synth()

Don't forget that final line there. And this synth function call needs to come after all of your assertions.

Now what?

We now have a green and passing test. It's time to refactor.

But, there are some very rough edges to refactoring. See https://pypi.org/project/cdk-integ-runner-cwd-fix/ for some hints on how to get around those edges.

You also have a pattern for how to test and run deployed services end to end. So, it's time to apply that to your SQS/RDS/S3 or whatever other services.

Finally, you have an integration test, but that doesn't mean you don't still benefit from unit tests. A unit test to check that the handler returns "Hello world" - without deploying infrastructure - would run in milliseconds. So, this 4 minute integration test is only one piece of your development framework.

How will you fit it into your testing approach?