[AWS] Running CI (Integration Testing) for each PR using AWS Emulator [GitHub Actions]

# Download the official deb and install it.
# EMULATOR_DEB_URL is the URL of the deb under the official LTS/latest
- name: Install AWS Emulator
  run: |
    curl -fsSLo aws-emulator.deb "$EMULATOR_DEB_URL"
    sudo apt-get install -y ./aws-emulator.deb
    aws-emulator -v    # Record the installed version in the log

# Start in the background and wait until the health check passes
- name: Start AWS Emulator
  run: |
    # -p: port number
    # --telemetry off: do not send usage data
    # --idle-timeout 1800: stop automatically if there is no API call for 1800 seconds
    #   (so the emulator does not keep running if the job stops midway)
    # nohup and the trailing &: keep the process alive after this step ends
    nohup aws-emulator -p 9297 --telemetry off --idle-timeout 1800 \
      > "$RUNNER_TEMP/emulator.log" 2>&1 &

    # Check once per second, up to 60 times, until /__admin/health succeeds.
    # --cacert: trust the self-signed CA that the emulator creates
    for i in $(seq 60); do
      curl -sf --cacert "$HOME/.aws/emulator/9297/ca.pem" \
        https://127.0.0.1:9297/__admin/health > /dev/null && exit 0
      sleep 1
    done

    # If it has not started after 60 seconds, print the emulator log and fail
    cat "$RUNNER_TEMP/emulator.log"
    exit 1

Version: 1.0.353
Release date: 2026.11.14
Commit: ca8a89dedfa8e4b5e67a3c90dc736f921ba409a1

- name: Point SDK and CLI at the emulator
  run: |
    # Where the emulator puts the credentials, config, and CA it creates at every startup
    d="$HOME/.aws/emulator/9297"
    {
      # The destination. When this is set, the SDK and CLI connect to the emulator instead of real AWS
      echo "AWS_ENDPOINT_URL=https://127.0.0.1:9297"

      # The credentials and config files that the emulator creates at startup
      echo "AWS_SHARED_CREDENTIALS_FILE=$d/emulator_credentials"
      echo "AWS_CONFIG_FILE=$d/config"

      # Trust the self-signed CA. The AWS CLI and boto3 use this AWS_CA_BUNDLE
      echo "AWS_CA_BUNDLE=$d/ca.pem"
      # The Node.js SDK does not use AWS_CA_BUNDLE, so set this too (see the Node.js section below)
      echo "NODE_EXTRA_CA_CERTS=$d/ca.pem"

      # Do not go looking for credentials in the EC2 instance metadata
      echo "AWS_EC2_METADATA_DISABLED=true"
      # Keep APIs whose host name gets a prefix (Cloud Map, Step Functions, and so on)
      # from being sent to a different host (official guidance)
      echo "AWS_DISABLE_HOST_PREFIX_INJECTION=true"

      # Region. The emulator is used with a single Region
      echo "AWS_REGION=us-east-1"
      echo "AWS_DEFAULT_REGION=us-east-1"
    } >> "$GITHUB_ENV"    # Pass them to later steps as environment variables

$ aws sts get-caller-identity
{
    "UserId": "AIDAH5I7GEMO7ADXNFAZU",
    "Account": "000000000000",
    "Arn": "arn:aws:iam::000000000000:user/emulator"
}

import time
import uuid

import boto3
import pytest
from botocore.exceptions import ClientError


def _suffix():
    # A random 8-character string so that bucket and table names do not collide.
    # A single run creates buckets and tables many times, so the names are varied
    return uuid.uuid4().hex[:8]


# One boto3 client is shared by the whole test run (scope="session").
# The endpoint and credentials are not written in the code. boto3 reads them automatically from
# the environment variables set in "CI Step 3: Switch the Target with Environment Variables"
# (AWS_ENDPOINT_URL, AWS_SHARED_CREDENTIALS_FILE, and so on).
# So the same code would also point at real AWS if you removed the environment variables
@pytest.fixture(scope="session")
def s3():
    return boto3.client("s3")


@pytest.fixture(scope="session")
def ddb():
    return boto3.client("dynamodb")


@pytest.fixture(scope="session")
def sqs():
    return boto3.client("sqs")

def test_s3_roundtrip(s3):
    bucket = f"ci-{_suffix()}"
    # Create a bucket
    s3.create_bucket(Bucket=bucket)
    # Write an object
    s3.put_object(Bucket=bucket, Key="a.txt", Body=b"hello")
    # Read the same key. If the content comes back, "create, write, read" works while keeping state
    body = s3.get_object(Bucket=bucket, Key="a.txt")["Body"].read()
    assert body == b"hello"


def test_s3_missing_key_is_nosuchkey(s3):
    bucket = f"ci-{_suffix()}"
    s3.create_bucket(Bucket=bucket)
    # Reading a key that does not exist raises an exception (ClientError)
    with pytest.raises(ClientError) as e:
        s3.get_object(Bucket=bucket, Key="nope")
    # The kind of error is in Error.Code of the response. Check that it is NoSuchKey, as in AWS
    assert e.value.response["Error"]["Code"] == "NoSuchKey"

def test_dynamodb_conditional_write(ddb):
    table = f"ci-{_suffix()}"
    # Create an on-demand table whose only key is an ID (string)
    ddb.create_table(
        TableName=table,
        BillingMode="PAY_PER_REQUEST",
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
    )
    # Print the table status right after creation, so that it is recorded in the log.
    # (In the first run, a write right after creation failed with ResourceNotFoundException,
    # so I made the status visible)
    status = ddb.describe_table(TableName=table)["Table"]["TableStatus"]
    print(f"table status right after create_table: {status}")
    # Wait until the table becomes ACTIVE. Check once per second, up to 30 times.
    # Do not write right after creating the table; wait first
    ddb.get_waiter("table_exists").wait(
        TableName=table, WaiterConfig={"Delay": 1, "MaxAttempts": 30}
    )
    item = {"id": {"S": "1"}}
    # Conditional write: write only if an item with the same ID does not exist yet. The first write succeeds
    ddb.put_item(
        TableName=table, Item=item,
        ConditionExpression="attribute_not_exists(id)",
    )
    # Write the same ID again. The item already exists, so the condition fails and an exception is raised
    with pytest.raises(ClientError) as e:
        ddb.put_item(
            TableName=table, Item=item,
            ConditionExpression="attribute_not_exists(id)",
        )
    assert e.value.response["Error"]["Code"] == "ConditionalCheckFailedException"

def test_sqs_redelivery_after_visibility_timeout(sqs):
    """If the message is not deleted, it is redelivered after the visibility timeout.
    The same point as the DEV.to article from the AWS organization (an example of a defect mocks miss)."""
    # Create a queue with a 2-second visibility timeout (to keep the wait short)
    q = sqs.create_queue(
        QueueName=f"ci-{_suffix()}",
        Attributes={"VisibilityTimeout": "2"},
    )["QueueUrl"]
    sqs.send_message(QueueUrl=q, MessageBody="order-1")

    # First receive. From the moment it is received, the message is hidden from other receives for 2 seconds.
    # ReceiptHandle identifies this receive (the value changes when the message is redelivered)
    first = sqs.receive_message(QueueUrl=q, WaitTimeSeconds=1)["Messages"][0]
    handle1 = first["ReceiptHandle"]

    # Wait 3 seconds without deleting (longer than the 2-second visibility timeout)
    time.sleep(3)

    # Second receive. The message should be visible again.
    # In case it takes a while to become visible, repeat the receive up to 5 times
    second = None
    for _ in range(5):
        r = sqs.receive_message(QueueUrl=q, WaitTimeSeconds=1)
        if r.get("Messages"):
            second = r["Messages"][0]
            break
    assert second is not None, "The message was not redelivered"
    # The same message was redelivered
    assert second["Body"] == "order-1"
    # It is a different receive, so ReceiptHandle differs from the first one
    assert second["ReceiptHandle"] != handle1

def test_unimplemented_api_status_is_recorded():
    """An unimplemented API is expected to return 501. It may get implemented in a release,
    so only print and record the status (it is shown with -s)."""
    athena = boto3.client("athena")
    try:
        # If it is implemented, the call simply succeeds (HTTP 200)
        athena.list_work_groups()
        status = 200
    except ClientError as e:
        # If it is not implemented, an error is raised. Take the HTTP status from the response
        status = e.response["ResponseMetadata"]["HTTPStatusCode"]
    # If you run pytest with -s, this output is shown in the log
    print(f"athena list_work_groups -> HTTP {status}")
    assert status in (200, 501)

============================= test session starts ==============================
platform linux -- Python 3.12.16, pytest-9.2.0, pluggy-1.7.0
collected 5 items

tests/test_integration.py::test_s3_roundtrip PASSED
tests/test_integration.py::test_s3_missing_key_is_nosuchkey PASSED
tests/test_integration.py::test_dynamodb_conditional_write table status right after create_table: ACTIVE
PASSED
tests/test_integration.py::test_sqs_redelivery_after_visibility_timeout PASSED
tests/test_integration.py::test_unimplemented_api_status_is_recorded athena list_work_groups -> HTTP 200
PASSED

============================== 5 passed in 6.48s ===============================

import { S3Client, ListBucketsCommand } from "@aws-sdk/client-s3";

// The SDK reads the endpoint from AWS_ENDPOINT_URL and the Region from AWS_REGION.
// forcePathStyle: use path-style requests, which do not put the bucket name in the host name
const client = new S3Client({ forcePathStyle: true });

// Get the list of buckets. If the connection to the emulator works, the number of buckets created so far is returned
const r = await client.send(new ListBucketsCommand({}));
console.log("buckets:", (r.Buckets ?? []).length);

Error: self-signed certificate
    at TLSSocket.onConnectSecure (node:internal/tls/wrap:1787:34)
    ...
  code: 'DEPTH_ZERO_SELF_SIGNED_CERT',
  '$metadata': { attempts: 1, totalRetryDelay: 0 }

# Run the CDK bootstrap (which creates things like the S3 bucket CDK uses) and deploy against the emulator
- name: CDK bootstrap and deploy
  working-directory: cdk
  env:
    CI: "true"    # Run as CI
  run: |
    # Specify the emulator's account (000000000000) as the account ID
    npx cdk bootstrap aws://000000000000/us-east-1 --no-notices

    # $CDK_CTX is the argument passed to CDK in the form -c emuDir=<emulator directory>
    # (collected in an environment variable). The code below receives it
    # --require-approval never: go ahead without asking for confirmation of IAM changes
    npx cdk deploy --require-approval never --no-notices $CDK_CTX

// Receive the value passed at deploy time with -c emuDir=<emulator directory>
const emuDir: string | undefined = this.node.tryGetContext('emuDir');

// Environment variables for Lambda. The table name is always passed
const environment: Record<string, string> = { TABLE_NAME: items.tableName };
if (emuDir) {
  // Only when running on the emulator, pass the endpoint and credentials to Lambda.
  // The emulator does not pass these to the Lambda execution environment automatically
  environment.AWS_ENDPOINT_URL = 'https://127.0.0.1:9297';
  environment.AWS_SHARED_CREDENTIALS_FILE = `${emuDir}/emulator_credentials`;
  environment.AWS_CONFIG_FILE = `${emuDir}/config`;
  // Node.js inside Lambda also has to trust the self-signed CA (see the Node.js section above)
  environment.NODE_EXTRA_CA_CERTS = `${emuDir}/ca.pem`;
}

import * as cdk from 'aws-cdk-lib/core';
import { Construct } from 'constructs';
import * as ddb from 'aws-cdk-lib/aws-dynamodb';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as apigw from 'aws-cdk-lib/aws-apigateway';

export class AppStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // In CI, pass -c emuDir=.
    // It is needed so that the Lambda function run by the emulator can call the emulator's DynamoDB
    const emuDir: string | undefined = this.node.tryGetContext('emuDir');

    // The table that stores items. Its only key is an ID (string).
    // DESTROY: delete the table too on cdk destroy (because it is a throwaway CI environment)
    const items = new ddb.Table(this, 'Items', {
      partitionKey: { name: 'id', type: ddb.AttributeType.STRING },
      billingMode: ddb.BillingMode.PAY_PER_REQUEST,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Environment variables for Lambda (the part explained in the first half of this section)
    const environment: Record = { TABLE_NAME: items.tableName };
    if (emuDir) {
      environment.AWS_ENDPOINT_URL = 'https://127.0.0.1:9297';
      environment.AWS_SHARED_CREDENTIALS_FILE = `${emuDir}/emulator_credentials`;
      environment.AWS_CONFIG_FILE = `${emuDir}/config`;
      environment.NODE_EXTRA_CA_CERTS = `${emuDir}/ca.pem`;
    }

    // The Lambda function that handles the API. The code is taken from the lambda/ directory
    const appFn = new lambda.Function(this, 'App', {
      runtime: lambda.Runtime.NODEJS_22_X,
      handler: 'items.handler',
      code: lambda.Code.fromAsset('lambda'),
      environment,
    });

    // Grant Lambda read and write permissions on the table.
    // In PR 1 of "Breaking It with a PR", I deleted this one line to reproduce a forgotten permission
    items.grantReadWriteData(appFn);

    // API Gateway. The stage name is v1.
    // POST /i saves an item and GET /i/{k} retrieves one. The same Lambda function handles both
    const api = new apigw.RestApi(this, 'A', { deployOptions: { stageName: 'v1' } });
    const res = api.root.addResource('i');
    res.addMethod('POST', new apigw.LambdaIntegration(appFn));
    res.addResource('{k}').addMethod('GET', new apigw.LambdaIntegration(appFn));

    // Output the API ID and the table name so that the tests can get them after deployment
    new cdk.CfnOutput(this, 'ApiId', { value: api.restApiId });
    new cdk.CfnOutput(this, 'TableName', { value: items.tableName });
  }
}

const { DynamoDBClient, PutItemCommand, GetItemCommand } = require('@aws-sdk/client-dynamodb');

// The SDK reads the endpoint and credentials from the Lambda environment variables
const ddb = new DynamoDBClient({});

// A helper that builds the response returned to API Gateway
const json = (statusCode, obj) => ({
  statusCode,
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(obj),
});

exports.handler = async (event) => {
  try {
    // POST /i: take id and text from the request body and save them to DynamoDB
    if (event.httpMethod === 'POST') {
      const { id, text } = JSON.parse(event.body || '{}');
      await ddb.send(new PutItemCommand({
        TableName: process.env.TABLE_NAME,
        Item: { id: { S: id }, text: { S: text } },
      }));
      // Return 201 (Created) when saved. In PR 2 of "Breaking It with a PR", I changed this to 200
      return json(201, { saved: id });
    }

    // GET /i/{k}: use k in the path as the ID and retrieve the item from DynamoDB
    const id = event.pathParameters && event.pathParameters.k;
    const r = await ddb.send(new GetItemCommand({
      TableName: process.env.TABLE_NAME,
      Key: { id: { S: id } },
    }));
    // Return 200 if the item exists, 404 if not
    return r.Item ? json(200, { id, text: r.Item.text.S }) : json(404, { message: 'not found' });
  } catch (e) {
    // If the DynamoDB call fails (for example, missing permissions), return 500 and the kind of error.
    // In PR 1 of "Breaking It with a PR", AccessDeniedException was returned here
    return json(500, { error: e.name, message: e.message });
  }
};

import os
import uuid

import boto3
import pytest
import requests

# The name of the stack under test
STACK = os.environ.get("STACK_NAME", "EmulatorCiApp")


@pytest.fixture(scope="module")
def outputs():
    # Get the outputs of the deployed stack (ApiId and TableName) from CloudFormation.
    # This queries the emulator's CloudFormation
    cf = boto3.client("cloudformation")
    out = cf.describe_stacks(StackName=STACK)["Stacks"][0]["Outputs"]
    return {o["OutputKey"]: o["OutputValue"] for o in out}


@pytest.fixture(scope="module")
def call(outputs):
    # The emulator's endpoint (https://127.0.0.1:9297)
    base = os.environ["AWS_ENDPOINT_URL"]
    # The emulator receives API Gateway requests at a single endpoint.
    # Which API is meant is specified with the Host header
    host = f"{outputs['ApiId']}.execute-api.us-east-1.amazonaws.com"
    # The self-signed CA (the same file as AWS_CA_BUNDLE)
    ca = os.environ["AWS_CA_BUNDLE"]

    def _call(method, path, **kwargs):
        return requests.request(
            method,
            # Append the stage name (v1) and the path to the endpoint
            f"{base}/v1{path}",
            headers={"Host": host},
            verify=ca,
            timeout=30,
            **kwargs,
        )

    return _call


def test_post_then_get(call):
    # Use a random ID so that it does not collide with other tests
    item_id = f"item-{uuid.uuid4().hex[:8]}"

    # Save with POST. 201 and the saved ID are returned
    r = call("POST", "/i", json={"id": item_id, "text": "HelloEmulator from CI"})
    assert r.status_code == 201, r.text
    assert r.json() == {"saved": item_id}

    # Retrieve with GET. The saved content comes back as it is
    r = call("GET", f"/i/{item_id}")
    assert r.status_code == 200, r.text
    assert r.json() == {"id": item_id, "text": "HelloEmulator from CI"}


def test_get_missing_item_is_404(call):
    # Getting an ID that was never saved returns 404.
    # This test does not use POST, so changing the status code that POST returns does not affect it
    r = call("GET", f"/i/nope-{uuid.uuid4().hex[:8]}")
    assert r.status_code == 404, r.text


def test_item_is_stored_in_dynamodb(call, outputs):
    item_id = f"item-{uuid.uuid4().hex[:8]}"
    r = call("POST", "/i", json={"id": item_id, "text": "stored"})
    assert r.status_code == 201, r.text

    # Read DynamoDB directly, without going through the API.
    # Check that the item was actually stored in the table, not only that the API responded
    ddb = boto3.client("dynamodb")
    got = ddb.get_item(
        TableName=outputs["TableName"], Key={"id": {"S": item_id}}
    )
    assert got["Item"]["text"]["S"] == "stored"

tests/test_api.py::test_post_then_get PASSED
tests/test_api.py::test_get_missing_item_is_404 PASSED
tests/test_api.py::test_item_is_stored_in_dynamodb PASSED

============================== 3 passed in 0.39s ===============================