AWS Emulatorでテストのたびに空の状態から始める方法

aws-emulator -t off

export AWS_ENDPOINT_URL=https://127.0.0.1:9297
export AWS_SHARED_CREDENTIALS_FILE=~/.aws/emulator/9297/emulator_credentials
export AWS_CONFIG_FILE=~/.aws/emulator/9297/config
export AWS_CA_BUNDLE=~/.aws/emulator/9297/ca.pem
export AWS_EC2_METADATA_DISABLED=true
export AWS_DISABLE_HOST_PREFIX_INJECTION=true
export AWS_REGION=us-east-1
export AWS_DEFAULT_REGION=us-east-1
unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN AWS_PROFILE

$ aws dynamodb create-table --table-name example-table --attribute-definitions AttributeName=id,AttributeType=S --key-schema AttributeName=id,KeyType=HASH --billing-mode PAY_PER_REQUEST --query 'TableDescription.[TableName,TableStatus]' --output text
example-table	CREATING
$ aws dynamodb wait table-exists --table-name example-table
$ aws dynamodb describe-table --table-name example-table --query 'Table.TableStatus' --output text
ACTIVE
$ aws dynamodb put-item --table-name example-table --item '{"id":{"S":"neko-1"},"name":{"S":"tama"}}'

$ aws s3api create-bucket --bucket example-bucket
{
    "Location": "/example-bucket",
    "BucketArn": "arn:aws:s3:::example-bucket"
}
$ aws s3 cp hello.txt s3://example-bucket/hello.txt
upload: ./hello.txt to s3://example-bucket/hello.txt

$ aws dynamodb list-tables
{
    "TableNames": [
        "example-table"
    ]
}
$ aws s3api list-buckets --query 'Buckets[].Name'
[
    "example-bucket"
]
$ aws s3 ls s3://example-bucket
2026-11-14 03:52:32          5 hello.txt
$ aws dynamodb scan --table-name example-table --query 'Items'
[
    {
        "id": {
            "S": "neko-1"
        },
        "name": {
            "S": "tama"
        }
    }
]

$ kill -TERM <エミュレーターのプロセスID>
emulator exit code: 0
$ ls -A ~/.aws/emulator/9297/

$ aws dynamodb list-tables
{
    "TableNames": []
}
$ aws s3api list-buckets --query 'Buckets[].Name'
[]
$ aws s3 ls s3://example-bucket

aws: [ERROR]: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist
$ aws dynamodb scan --table-name example-table --query 'Items'

aws: [ERROR]: An error occurred (ResourceNotFoundException) when calling the Scan operation: Requested resource not found

$ kill -KILL <エミュレーターのプロセスID>
emulator exit code: 137
$ ls -A ~/.aws/emulator/9297/
ca.pem
config
emulator_credentials
truststore.p12

$ sha256sum ~/.aws/emulator/9297/emulator_credentials   # 落ちる前
8c066c5a070b752df4e7e988eddb71d46fa9710d1ee02b4116778a7bfcd97b9a  ~/.aws/emulator/9297/emulator_credentials
$ sha256sum ~/.aws/emulator/9297/emulator_credentials   # 起動し直した後
d2b1e731a4961a68d3200370f4ed8b6fceb5d87a564d748a47a7db004bb8bf73  ~/.aws/emulator/9297/emulator_credentials
$ aws dynamodb list-tables
{
    "TableNames": []
}
$ aws s3api list-buckets --query 'Buckets[].Name'
[]

import boto3


def test_save_order():
    ddb = boto3.client("dynamodb")
    ddb.create_table(
        TableName="orders",
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        BillingMode="PAY_PER_REQUEST",
    )
    ddb.get_waiter("table_exists").wait(TableName="orders")
    ddb.put_item(TableName="orders", Item={"id": {"S": "order-1"}})
    assert ddb.scan(TableName="orders")["Count"] == 1


def test_upload_receipt():
    s3 = boto3.client("s3")
    s3.create_bucket(Bucket="receipts")
    s3.put_object(Bucket="receipts", Key="order-1.txt", Body=b"ok")
    keys = [o["Key"] for o in s3.list_objects_v2(Bucket="receipts")["Contents"]]
    assert keys == ["order-1.txt"]

$ python -m pytest -q   # 1回目
..                                                                       [100%]
2 passed in 3.25s
$ python -m pytest -q   # 2回目
E           botocore.errorfactory.ResourceInUseException: An error occurred (ResourceInUseException) when calling the CreateTable operation: Table already exists: orders
FAILED test_orders.py::test_save_order - botocore.errorfactory.ResourceInUseE...
1 failed, 1 passed in 0.24s

#!/bin/bash
# テストを走らせるたびに、新しいエミュレーターを起動して、終わったら止める
set -euo pipefail

LOG=$(mktemp)
aws-emulator -t off -p 0 --idle-timeout 600 > "$LOG" 2>/dev/null &
EMU_PID=$!
trap 'kill -TERM "$EMU_PID" 2>/dev/null; wait "$EMU_PID" 2>/dev/null || true; rm -f "$LOG"' EXIT

# 起動の行（message が emulator server information）が出るまで待って、接続先を読む
for i in $(seq 30); do
  INFO=$(grep '"emulator server information"' "$LOG" || true)
  [ -n "$INFO" ] && break
  sleep 1
done
[ -n "$INFO" ] || { echo "emulator did not start" >&2; exit 1; }
ENDPOINT=$(echo "$INFO" | python3 -c 'import json,sys; print(json.load(sys.stdin)["fields"]["endpoint_url"])')
DIR=$(dirname "$(echo "$INFO" | python3 -c 'import json,sys; print(json.load(sys.stdin)["fields"]["config_file"])')")

export AWS_ENDPOINT_URL="$ENDPOINT"
export AWS_SHARED_CREDENTIALS_FILE="$DIR/emulator_credentials"
export AWS_CONFIG_FILE="$DIR/config"
export AWS_CA_BUNDLE="$DIR/ca.pem"
export AWS_EC2_METADATA_DISABLED=true
export AWS_DISABLE_HOST_PREFIX_INJECTION=true
export AWS_REGION=us-east-1
export AWS_DEFAULT_REGION=us-east-1
unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN AWS_PROFILE

curl -sf --cacert "$AWS_CA_BUNDLE" "$ENDPOINT/__admin/health" > /dev/null
echo "emulator: $ENDPOINT"
python -m pytest -q "$@"

$ ./run-tests.sh   # 1回目
emulator: https://127.0.0.1:33901
..                                                                       [100%]
2 passed in 3.20s
$ ./run-tests.sh   # 2回目
emulator: https://127.0.0.1:42575
..                                                                       [100%]
2 passed in 3.20s

import json
import os
import subprocess

import boto3
import pytest


@pytest.fixture(autouse=True)
def fresh_emulator(monkeypatch):
    """テスト関数ごとに、空のエミュレーターを起動して、終わったら止める"""
    proc = subprocess.Popen(
        ["aws-emulator", "-t", "off", "-p", "0", "--idle-timeout", "600"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
    )
    # 起動の行（message が emulator server information）が出るまで読む
    info = None
    for line in proc.stdout:
        fields = json.loads(line).get("fields", {})
        if fields.get("message") == "emulator server information":
            info = fields
            break
    if info is None:
        proc.kill()
        pytest.fail("emulator did not start")

    folder = os.path.dirname(info["config_file"])
    monkeypatch.setenv("AWS_ENDPOINT_URL", info["endpoint_url"])
    monkeypatch.setenv("AWS_SHARED_CREDENTIALS_FILE", os.path.join(folder, "emulator_credentials"))
    monkeypatch.setenv("AWS_CONFIG_FILE", info["config_file"])
    monkeypatch.setenv("AWS_CA_BUNDLE", info["ca_file"])
    monkeypatch.setenv("AWS_EC2_METADATA_DISABLED", "true")
    monkeypatch.setenv("AWS_DISABLE_HOST_PREFIX_INJECTION", "true")
    monkeypatch.setenv("AWS_REGION", "us-east-1")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
    for name in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "AWS_PROFILE"]:
        monkeypatch.delenv(name, raising=False)
    # boto3.client() が使う既定のセッションは、最初に読んだ鍵を覚えている。
    # 前のテストの鍵が残らないように、テストごとに作り直させる
    monkeypatch.setattr(boto3, "DEFAULT_SESSION", None)

    yield info["endpoint_url"]

    proc.terminate()  # SIGTERM で止めると、メモリーの中身とポートのフォルダーのファイルが消える
    proc.wait(timeout=30)

def test_starts_empty():
    assert boto3.client("dynamodb").list_tables()["TableNames"] == []
    assert boto3.client("s3").list_buckets()["Buckets"] == []

$ python -m pytest -v --durations=0
collecting ... collected 3 items

test_orders.py::test_save_order PASSED                                   [ 33%]
test_orders.py::test_upload_receipt PASSED                               [ 66%]
test_orders.py::test_starts_empty PASSED                                 [100%]
============================== 3 passed in 4.04s ===============================

$ python -m pytest -q
E           botocore.errorfactory.AccessDenied: An error occurred (AccessDenied) when calling the CreateBucket operation: Sign the request with the emulator credentials
E           botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the ListTables operation: The security token included in the request is invalid.
FAILED test_orders.py::test_upload_receipt - botocore.errorfactory.AccessDeni...
FAILED test_orders.py::test_starts_empty - botocore.exceptions.ClientError: A...
2 failed, 1 passed in 3.95s

$ python check_cache.py
run 1: key in file == key in DEFAULT_SESSION: True
run 2: key in file == key in DEFAULT_SESSION: False