8.2AWS 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"