Testing serverless python applications with serverless offline + pytest

Recently, I found myself in a tedious and cumbersome development process that required deploying a serverless framework every time I needed to test my application. Overcoming this with the serverless-offline plugin as highlighted in one of my articles, I decided to find ways to have this incorporated in my testings suite in other to simulate real world events and eventually, applied to my CI/CD pipelines.

Testing is a critical aspect of the software development process, and it serves several important purposes(I had to learn the hard way haha). Writing self sufficient test cases for robust serverless applications, especially when considering integration into CI/CD pipelines, can be a complex and challenging task and this arises more often than not due to the ease with which resources and functions within the serverless application become tightly coupled.

---

Requirements

Before we begin, make sure you have the following:

• A simple serverless application to follow with, feel free to use my simple todo app (since that is what I will be using in this tutorial).

Creating serverless configuration file

The first step I usually follow is creating a separate serverless configuration file for testing, and this helps to isolate and tailor the configuration specifically for testing purposes. By having a dedicated configuration file, I can customize settings, environment variables, and plugins to optimize the testing environment without impacting the production setup. This separation ensures a streamlined and efficient testing process, allowing for easy adjustments and experimentation without affecting the production deployment.

Below is an example of a serverless-test.yml file:

app: 'todo-app'
service: 'todo-app'
frameworkVersion: '3'

provider:
  name: aws
  stage: test
  runtime: python3.11
  timeout: 30
  region: us-east-1
  memorySize: 512
  architecture: arm64
  environment:
    STAGE: ${self:provider.stage}

    TODO_APP_DB: ms-todo-app
    TODO_APP_DB_PK: id
    TODO_APP_DB_SK: updated_at

# Upldoad just essential files to AWS Lambda
package:
  patterns:
    - '!./**'
    - src/**
    - '!src/**/__pycache__/**'
    - 'docs/**'

plugins:
  - serverless-dynamodb
  - serverless-offline

custom:
  dynamodb:
    # If you only want to use DynamoDB Local in some stages, declare them here
    stages:
      - test
    start:
      host: 127.0.0.1
      port: 8001
      inMemory: true
      heapInitial: 200m
      heapMax: 1g
      migrate: true
      seed: true
      convertEmptyValues: true
    # Uncomment only if you already have a DynamoDB running locally
    # noStart: true

# Add functions
functions:
  todo_app_gateway_handler:
    handler: src.main.todo_app_gateway_handler
    description: "Todo app system gateway for different funtions"
    events:
      - http:
          path: /docs/{path+} # Matches every path under /docs
          method: get
          cors:
            origin: '*'
            headers: '*'

      - http:
          path: '/{path+}' # Matches every path under /
          method: ANY
          cors:
            origin: '*'
            headers: '*'

resources:
  # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/template-reference.html
  Resources:
    TodoAppDynamoDB:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:provider.environment.TODO_APP_DB}
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
        BillingMode: 'PROVISIONED'

---

Setting up pytest

Next, install the required python packages in your desired environment:

pip install pytest pytest-xprocess pytest-cov

pytest-xprocess is a pytest plugin for managing external processes across test runs.

pytest-cov is a pytest plugin for producing coverage report.

For ease and to avoid passing arguments through the cli, set up your pytest configuration in the pyproject.toml file like:

# ==== pytest ====
[tool.pytest.ini_options]
minversion = "6.0"
addopts = [
    "--junit-xml=./unittests.xml",
    "--cov=./src",
    "--cov-report=term",
    "--cov-report=xml",
    "--cov-branch",
    "-p tests.plugins.env_vars", # injecting environment variables
]
python_files = [
    "tests.py",
    "test_*.py",
]
log_cli = true # set to `false` if you do not want to log output

Also environment variables are set in test.plugins.env_vars as thus:

import os

import pytest


@pytest.hookimpl(tryfirst=True)
def pytest_load_initial_conftests(args, early_config, parser):
    os.environ["STAGE"] = "test"
    os.environ["TODO_APP_DB"] = "ms-todo-app"
    os.environ["TODO_APP_DB_PK"] = "id"
    os.environ["TODO_APP_DB_SK"] = "updated_at"

For generating realistic data I use python faker library alongside polyfactory — which works well with pydantic as that is what is being used for data validation, to generate my data factories. Feel free to check them out.

Lastly and most importantly, creating your conftest.py file to setup fixtures and manage processes.

import time
import sys
from pathlib import Path

import pytest
from xprocess import ProcessStarter

from src.config.settings import get_settings
from src.db import DynamoDB
from tests import logger

settings = get_settings()
BASE_DIR = Path(__file__).resolve().parent.parent


@pytest.fixture(scope="session")
def server(xprocess):
    """
    This fixture starts the serverless offline server and dynamodb.
    Also logs the serverless offline server output to the console.
    """

    class Starter(ProcessStarter):
        pattern = "Server ready"
        args = [
            "/bin/bash",
            "-c",
            "cd " + str(BASE_DIR) + " && serverless offline start " + "--config serverless-test.yml",
        ]

    logfile = xprocess.ensure("server", Starter)

    try:
        # Print logs in console
        sys.stderr.flush()
        sys.stdin.flush()
    except Exception:
        ...
    time.sleep(3)

    yield
    with open(str(logfile[1]), "r") as f:
        logger.info(f.read())

    xprocess.getinfo("server").terminate()


@pytest.fixture(scope="module")
def dynamodb(xprocess) -> DynamoDB:
    """
    This fixture starts the dynamodb server only and logs the output to the console.
    Scope is set to `module` to avoid db clashing with server db - so db only test cases are placed in the same module.
    """

    class Starter(ProcessStarter):
        pattern = "DynamoDB - created"
        args = [
            "/bin/bash",
            "-c",
            "cd " + str(BASE_DIR) + " && serverless dynamodb start --migrate --config serverless-test.yml --sharedDb",
        ]

    logfile = xprocess.ensure("dynamodb", Starter)

    try:
        # Print logs in console
        sys.stderr.flush()
        sys.stdin.flush()
    except Exception:
        ...

    time.sleep(3)
    yield
    with open(str(logfile[1]), "r") as f:
        logger.info(f.read())

    xprocess.getinfo("dynamodb").terminate()


def create_todo_app_db(db: DynamoDB):
    """
    This function creates the todo app database.
    """
    return db.dynamodb_client.create_table(
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
            {"AttributeName": "updated_at", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "id", "KeyType": "HASH"},
            {"AttributeName": "updated_at", "KeyType": "RANGE"},
        ],
        TableName=db._table.name,
        BillingMode="PAY_PER_REQUEST",
    )


@pytest.fixture(scope="function")
def clear_db():
    """
    This fixture helps in resetting the database after each test by deleting the table and recreating it.
    """
    yield
    dbs = [
        DynamoDB(settings.TODO_APP_DB),
    ]
    for db in dbs:
        db.dynamodb_client.delete_table(TableName=db._table.name)
        match db._table.name:
            case settings.TODO_APP_DB:
                create_todo_app_db(db)
            case _:
                raise ValueError(f"Invalid table name: {db._table.name}")
        time.sleep(1)

---

Writing test cases

Below is a test case to get a todo item from the database:

import requests

from src.config.settings import get_settings
from src.db import DynamoDB
from tests.factories import TodoDBSchemaFactory

settings = get_settings()
dynamodb = DynamoDB(table_name=settings.TODO_APP_DB)
BASE_URL = "http://localhost:3000"


def test_get_todo(server, clear_db):
    # Create a todo
    path = "/test/todos/{id}"
    todo = TodoDBSchemaFactory.build()
    dynamodb.put_item(todo.model_dump())

    resp = requests.get(BASE_URL + path.format(id=todo.id))
    assert resp.status_code == 200
    assert resp.json()["data"]["id"] == todo.id

---

Setting up coverage [Optional]

Since the server is a new process, retrieving the complete code coverage was a bit tricky but thanks to the pytest-cov package I was able to hook it up by placing a simple code block (which can be found in the documentation) into the __init__.py file of my parent folder.

import os 

if os.environ["STAGE"] == "test":
    try:
        from pytest_cov.embed import cleanup_on_sigterm
    except ImportError:
        pass
    else:
        cleanup_on_sigterm()

Once completed, use the command python -m pytest to run tests 🚀

Well, I think that's it for now. For the complete code implementation, you can refer to the Github repository at: https://github.com/charles-co/ms-todo-app. I hope you find this resource helpful in your development process. If you have any further questions or need additional assistance, please let me know. Happy coding! :)