[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=<emul...