Serverless Function Testing

The Serverless Testing Challenge

Serverless functions run in environments you do not control. You cannot SSH into the machine, you cannot inspect the filesystem, and you cannot attach a debugger in production. The execution environment is ephemeral -- it exists only for the duration of the invocation.

This means your testing strategy must be more rigorous than traditional application testing. If you miss a bug in a serverless function, your only debugging tools are logs and traces. There is no "let me quickly check on the server."

AWS Lambda with SAM Local

AWS SAM (Serverless Application Model) provides local emulation for Lambda functions. SAM Local runs a Docker container that mimics the Lambda execution environment, allowing you to test functions locally before deploying to AWS.

Local Invocation

# Invoke a single function locally with a test event
sam local invoke MyFunction --event events/api-gateway.json

# Start a local API Gateway emulator
sam local start-api --port 3000

# Run tests against the local emulator
pytest tests/integration/ --base-url http://localhost:3000

# Generate a sample event for testing
sam local generate-event apigateway aws-proxy > events/api-gateway.json
sam local generate-event sqs receive-message > events/sqs-message.json
sam local generate-event s3 put > events/s3-put.json

Unit Testing Lambda Handlers

Lambda handlers are just functions. Test them like any other function, mocking the AWS services they call:

# tests/unit/test_handler.py
import json
import pytest
from unittest.mock import patch, MagicMock
from src.handlers.order_processor import lambda_handler

def test_order_processing_success():
    """Test successful order processing."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123",
                "items": [{"sku": "WIDGET-A", "qty": 2}],
                "customerId": "CUST-456"
            })
        }]
    }
    context = MagicMock()
    context.function_name = "order-processor"
    context.memory_limit_in_mb = 256
    context.get_remaining_time_in_millis.return_value = 30000

    result = lambda_handler(event, context)

    assert result["statusCode"] == 200
    body = json.loads(result["body"])
    assert body["orderId"] == "ORD-123"
    assert body["status"] == "processed"

def test_order_processing_invalid_payload():
    """Test handling of malformed input."""
    event = {"Records": [{"body": "not json"}]}
    context = MagicMock()

    result = lambda_handler(event, context)

    assert result["statusCode"] == 400
    body = json.loads(result["body"])
    assert "error" in body

def test_order_processing_missing_required_fields():
    """Test handling of missing required fields."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123"
                # Missing: items and customerId
            })
        }]
    }
    context = MagicMock()

    result = lambda_handler(event, context)

    assert result["statusCode"] == 400

@patch("src.handlers.order_processor.dynamodb_client")
def test_order_persisted_to_dynamodb(mock_dynamo):
    """Test that processed orders are saved to DynamoDB."""
    event = {
        "Records": [{
            "body": json.dumps({
                "orderId": "ORD-123",
                "items": [{"sku": "WIDGET-A", "qty": 2}],
                "customerId": "CUST-456"
            })
        }]
    }
    context = MagicMock()

    lambda_handler(event, context)

    mock_dynamo.put_item.assert_called_once()
    call_args = mock_dynamo.put_item.call_args
    item = call_args["Item"]
    assert item["orderId"]["S"] == "ORD-123"
    assert item["status"]["S"] == "processed"

Google Cloud Functions Testing

Google Cloud Functions can be tested locally using the Functions Framework:

# Install the Functions Framework for local testing
pip install functions-framework

# Run locally
functions-framework --target=process_order --port=8080

# Test with curl
curl -X POST http://localhost:8080 \
  -H "Content-Type: application/json" \
  -d '{"orderId": "ORD-123"}'

Testing Cloud Functions with pytest

# tests/test_cloud_function.py
import pytest
from unittest.mock import patch, MagicMock
from flask import Flask
import json

# Import the function
from main import process_order

@pytest.fixture
def app():
    return Flask(__name__)

@pytest.fixture
def client(app):
    return app.test_client()

def test_process_order_valid_request(app):
    """Test processing a valid order."""
    with app.test_request_context(
        method="POST",
        json={"orderId": "ORD-123", "amount": 99.99}
    ):
        from flask import request
        response = process_order(request)
        data = json.loads(response)
        assert data["status"] == "processed"

def test_process_order_missing_order_id(app):
    """Test handling of missing order ID."""
    with app.test_request_context(
        method="POST",
        json={"amount": 99.99}
    ):
        from flask import request
        response, status_code = process_order(request)
        assert status_code == 400

Serverless Testing Challenges and Solutions

Cold Start Testing

# tests/performance/test_cold_start.py
import time
import boto3
import statistics

def measure_cold_start(function_name, num_invocations=10):
    """Measure cold start latency by forcing new execution environments."""
    client = boto3.client("lambda")
    latencies = []

    for i in range(num_invocations):
        # Update environment variable to force a new container
        client.update_function_configuration(
            FunctionName=function_name,
            Environment={
                "Variables": {
                    "COLD_START_TRIGGER": str(time.time())
                }
            }
        )
        # Wait for update to propagate
        time.sleep(5)

        start = time.time()
        response = client.invoke(
            FunctionName=function_name,
            Payload=b'{"test": true}',
        )
        elapsed_ms = (time.time() - start) * 1000
        latencies.append(elapsed_ms)

    return {
        "p50": statistics.median(latencies),
        "p95": sorted(latencies)[int(len(latencies) * 0.95)],
        "p99": sorted(latencies)[int(len(latencies) * 0.99)],
        "max": max(latencies),
    }

Timeout Testing

def test_function_completes_before_timeout():
    """Verify the function completes within its configured timeout."""
    import time

    FUNCTION_TIMEOUT = 30  # seconds (from SAM template)
    SAFETY_MARGIN = 5      # seconds

    start = time.time()
    result = lambda_handler(large_event, context)
    elapsed = time.time() - start

    assert elapsed < (FUNCTION_TIMEOUT - SAFETY_MARGIN), \
        f"Function took {elapsed:.1f}s, dangerously close to {FUNCTION_TIMEOUT}s timeout"

Testing with LocalStack

LocalStack provides a local AWS cloud stack for testing serverless functions with real AWS service emulations:

# tests/integration/test_lambda_localstack.py
import boto3
import json
import pytest

@pytest.fixture(scope="module")
def aws_clients():
    """Create AWS clients pointing to LocalStack."""
    endpoint = "http://localhost:4566"
    return {
        "lambda": boto3.client("lambda", endpoint_url=endpoint,
                               region_name="us-east-1",
                               aws_access_key_id="test",
                               aws_secret_access_key="test"),
        "sqs": boto3.client("sqs", endpoint_url=endpoint,
                            region_name="us-east-1",
                            aws_access_key_id="test",
                            aws_secret_access_key="test"),
        "dynamodb": boto3.client("dynamodb", endpoint_url=endpoint,
                                 region_name="us-east-1",
                                 aws_access_key_id="test",
                                 aws_secret_access_key="test"),
    }

def test_lambda_processes_sqs_message(aws_clients):
    """Test the full flow: SQS message triggers Lambda, result in DynamoDB."""
    sqs = aws_clients["sqs"]
    dynamodb = aws_clients["dynamodb"]

    # Send message to SQS
    queue_url = sqs.create_queue(QueueName="orders")["QueueUrl"]
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({"orderId": "ORD-TEST", "amount": 42.00})
    )

    # Wait for Lambda to process (triggered by SQS event source mapping)
    import time
    time.sleep(5)

    # Verify the result in DynamoDB
    result = dynamodb.get_item(
        TableName="orders",
        Key={"orderId": {"S": "ORD-TEST"}}
    )
    assert "Item" in result
    assert result["Item"]["status"]["S"] == "processed"

The combination of unit tests (fast, isolated), local emulation (SAM Local, Functions Framework), and integration tests (LocalStack, Testcontainers) provides comprehensive coverage for serverless functions without requiring a deployed cloud environment for every test run.