The Problem Link to heading

Testing AWS lambdas can be a pain. I ran into a problem recently where I needed to use AWS SAM to locally test a function with external (Dockerized) dependencies, this blog details how I solved that.

Between ad-hoc solutions, AWS SAM’s ability to locally invoke functions, mocking resources using tools like LocalStack or Moto, there’s a bit of an ecosystem and plenty of pain points to lambda testing. After all, let us not stray and test in prod, right?

In this case, I had a lambda meant to query a database on a schedule (so EventBridge Scheduler here) then send the results to a webhook. How to test it was an early question to answer, as I was whipping up the core code.

Solutioning Link to heading

Local DB Querying Link to heading

I ended up creating a database in a docker-compose.yml and a simple Flask app to echo the JSON results living on localhost. Here’s a code example to demonstrate.

services:
  db:
    image: postgres
    restart: always
    environment:
      POSTGRES_PASSWORD: tutorial

The problem is that AWS SAM’s sam local invoke runs local lambdas in isolated Docker containers, where localhost refers to the docker network, and so networking gets funky and folks have come up with some cool solutions.

Personally, I ended up using host.docker.internal for the lambda to connect to the db running on Docker with port 5432 mapped on localhost, and for referring to localhost:5000 for the flask app. This allowed local communication to the db running in a Docker container and the flask app, which I just ran in a terminal off my machine.

# Use host.docker.internal for compose service, don't hardcode this kinda stuff in prod
db_url = "postgresql://postgres:tutorial@host.docker.internal:5432/postgres"
webhook_url = (
    "http://host.docker.internal:5000"  # Use host.docker.internal for localhost webhook
)

def parse_connection_url(url: str) -> dict:
    parsed = parse.urlparse(url)
    return {
        "hostname": parsed.hostname,
        "port": parsed.port,
        "username": parsed.username,
        "password": parsed.password,
        "dbname": parsed.path[1:],
    }


def get_stat_activity(db_url: str) -> Dict:
    config = parse_connection_url(db_url)

    query = """
    SELECT *
    FROM pg_stat_statements
    ORDER BY pg_stat_statements.queryid DESC
    LIMIT 5;
    """

    query_stats = []
    try:
        with psycopg.connect(
            host=config["hostname"],
            port=config["port"],
            user=config["username"],
            password=config["password"],
            dbname=config["dbname"],
            connect_timeout=10,
        ) as conn:

            with conn.cursor() as cur:
                cur.execute(query)
                results = cur.fetchall()

                for row in results:
                    query_stats.append(
                        {
                            "userid": row[0],
                            "dbid": row[1],
                            "queryid": row[2],
                            "query": row[3],
                        }
                    )

            conn.close()

    except Exception as e:
        logger.log(logging.ERROR, f"Error getting stats of queries: {e}")

    return {"query_stats": query_stats}


def payload_to_webhook(payload: Dict, webhook_url: str = webhook_url):
    data = json.dumps(payload).encode("utf-8")
    req = request.Request(
        webhook_url, data=data, headers={"Content-Type": "application/json"}
    )
    try:
        with request.urlopen(req, timeout=10) as response:
            logger.log(logging.INFO, f"Response: {response.status}")
            return response.read(), response
    except Exception as e:
        logger.log(logging.ERROR, f"Error sending payload to webhook: {e}")
        raise e


def lambda_handler(event, context):
    queries = get_stat_activity(db_url=db_url)
    logging.info(f"queries: {queries}")
    try:
        payload_to_webhook(queries, webhook_url)
    except Exception as e:
        logging.error(f"Error sending payload to webhook: {e}")
        raise e

Mocking Services Link to heading

For mocks, I’ve been using moto , it’s wonderfully simple and effective at mocking AWS services. LocalStack seemed like a lot, and I’m sure it has its powerful benefits, but moto does the job well. The docs have a class-focused approach and while I shy away from Python classes a lot of the time, writing more procedural code, here I went along with the ride:

class SSMParameters:
    def __init__(self, client: boto3.client):
        self.client = client

    def get_parameters(self, param_names: List[str]) -> Dict:
        ssm = self.client
        parameters = {}
        for param_name in param_names:
            response = ssm.get_parameter(Name=param_name, WithDecryption=True)
            parameters[param_name] = response["Parameter"]["Value"]
        return parameters

Gets tested with the following code:

import boto3
from moto import mock_aws
from app import SSMParameters

@mock_aws
def test_ssm_parameters_get_parameters():
    conn = boto3.client("ssm", region_name="us-east-1")
    conn.put_parameter(
        Name="/test/test_db_url",
        Value="postgresql://postgres:postgres@mock:5433/postgres",
        Type="String",
        Overwrite=True
    )
    ssm_params_instance = SSMParameters(client=conn)
    response = ssm_params_instance.get_parameters(["/test/test_db_url"])

    assert response == {"/test/test_db_url": "postgresql://postgres:postgres@mock:5433/postgres"}

Which works wonderfully with pytest! This is simple, easy decoration of Python code, and it turns the whole normal boto3 code into a mock for testing. Quite impressive, and I’m looking forward to using this library in different contexts with other services, finding its limits along the way – not all services are supported, for example, that would be silly difficult with how AWS delivers new services so frequently.

Conclusion Link to heading

So first of all, thanks for reading up to now! I appreciate you taking your time to possibly learn a bit with me – maybe you hadn’t heard of moto, or just needed a refresher on local testing lambdas, well, at least I was in both camps when I started this work. I encourage you to reach out on LinkedIn if this was a good read or helped you out!