0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@Nana_777(山本 直弥（Nao）)
in

【AWS】AWS EmulatorでPRごとのCI（結合テスト）を回す【GitHubActions】

0
Last updated at Posted at 2026-11-14

はじめに

2026年11月11日に、AWS Emulator が公式にリリースされました。AWS SDK や AWS CLI の呼び出しを、AWS アカウントなしで、ローカルや CI 上のエミュレーターに向けられるツールです。公式ドキュメントは、ユースケースのひとつに CI/CD を挙げています。

「Continuous integration and delivery (CI/CD) – Run integration tests in your pipeline with no AWS account and no AWS resources.」
出典: https://docs.aws.amazon.com/emulator/latest/userguide/what-is-service.html

ただ、公式ドキュメントとリポジトリに、GitHub Actions のワークフロー例は、執筆時点では見当たりませんでした。既製のコンテナイメージもありません（公式リポジトリの Issue #14 で要望が出ています）。そこで、自分で GitHub Actions に組み込んで、PR ごとの結合テストとして回せるかを試しました。

前回の記事では、手元の Docker で、CDK と Terraform の構成がどこまで動くかを確認しています。今回はその続きで、CDK を使います。

試したのは次のことです。

  • 公式の deb パッケージを、GitHub Actions のランナーに入れて起動する
  • SDK で S3、DynamoDB、SQS を呼ぶテストと、CDK でデプロイした API（API Gateway + Lambda + DynamoDB）のテストを、AWS の認証情報なしで回す
  • 意図的に壊した PR を出して、CI が赤になるかを見る
  • deb のキャッシュが効くかを試す

GitHub Actions と、AWS SDK / CLI の基本的な使い方は、前提とします。

全体像

aws-emulator-github-actions-ci-overview.png

PR を作ると、GitHub Actions のランナーが起動します。ランナー上でエミュレーターを入れて起動し（ステップ1、2）、環境変数 AWS_ENDPOINT_URL で、テストの向け先をエミュレーターに変えます（ステップ3）。テストは2段階です。層1は、SDK で API を直接呼ぶテストで、エミュレーターの挙動を確かめます（ステップ4）。層2は、CDK でデプロイしたアプリのテストで、PR ごとの結合テストの型になります（ステップ5）。結果は、PR のチェックに緑か赤で返り、AWS には何も送られません。

層2のアプリは、前回の記事で使った検証用のサンプルです。実際のアプリに置き換えるときの型として読んでください。

従来のCIとの違い

AWS 上で動くアプリの結合テストは、実 AWS にデプロイして確かめるか、モックで済ませるかのどちらかが多いと思います。AWS Emulator は、その中間にあたります。

実AWSにデプロイするCI モック中心のCI AWS EmulatorのCI
認証情報 OIDC などで実ロールを渡す 不要 不要
待ち時間 デプロイ待ち ほぼなし ジョブ全体で約 2 分
費用と残骸 リソース費用と、消し忘れ なし 追加料金なし。ジョブが終わると全て消える
確かめられること 実際の AWS の挙動と権限 自分が書いた範囲 API の挙動とエラー。IAM は AWS と同じには評価されない

検証環境

  • GitHub Actions の ubuntu-latest（実際は ubuntu-26.04、x86_64）
  • AWS Emulator 1.0.353（latest を取得。Release date: 2026.11.14）
  • AWS CDK CLI 2.1148.1、aws-cdk-lib 2.275.0（前回の検証と同じ組み合わせ）、Python 3.12、Node.js 22
  • 検証日は 2026-11-14

エミュレーターは1日に何度も更新されます。結果は、この版のものです。

CI のステップ1〜2: エミュレーターを入れて起動する

deb は、Ubuntu や Debian などの Linux で使うインストール用のファイル（パッケージ）の形式で、Windows の .exe や .msi に近いものです。公式が配布していて、apt-get install で入れると aws-emulator コマンドが使えます。ファイル名の amd64 は、一般的な 64 ビット CPU 向けという意味です。

この deb を入れて、バックグラウンドで起動し、ヘルスチェックが通るまで待ちます。

# 公式の deb をダウンロードして、インストールする。
# EMULATOR_DEB_URL は、公式の LTS/latest にある deb の URL
- name: Install AWS Emulator
  run: |
    curl -fsSLo aws-emulator.deb "$EMULATOR_DEB_URL"
    sudo apt-get install -y ./aws-emulator.deb
    aws-emulator -v    # 入った版を、ログに残す

# バックグラウンドで起動し、ヘルスチェックが通るまで待つ
- name: Start AWS Emulator
  run: |
    # -p: ポート番号
    # --telemetry off: 利用状況の送信を止める
    # --idle-timeout 1800: 1800 秒、API の呼び出しがなければ、自動で止まる
    #   （ジョブが途中で止まっても、エミュレーターが動き続けないようにする）
    # nohup と末尾の &: ステップが終わっても、プロセスが残るようにする
    nohup aws-emulator -p 9297 --telemetry off --idle-timeout 1800 \
      > "$RUNNER_TEMP/emulator.log" 2>&1 &

    # /__admin/health が成功するまで、1秒ごとに最大 60 回確認する。
    # --cacert: エミュレーターが作る自己署名の CA を、信頼させる指定
    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

    # 60 秒たっても起動しなければ、エミュレーターのログを出して、失敗にする
    cat "$RUNNER_TEMP/emulator.log"
    exit 1

インストールは 16〜37 秒かかり（実行によってばらつきます）、大半は apt の展開（10〜15 秒）でした。起動は約 1 秒で、nohup で起動したプロセスは、後続のステップからも使えました。aws-emulator -v の出力は、次のとおりです。

Version: 1.0.353
Release date: 2026.11.14
Commit: ca8a89dedfa8e4b5e67a3c90dc736f921ba409a1

起動直後のログに curl: option --cacert: is badly used here と出ることがありますが、ca.pem がまだ作られていない間のメッセージで、約 1 秒で成功するので問題ありません。

CI のステップ3: 向け先を環境変数で切り替える

環境変数を GITHUB_ENV に書いて、後続のステップに引き継ぎます。

- name: Point SDK and CLI at the emulator
  run: |
    # エミュレーターが、起動のたびに作る、認証情報、設定、CA の置き場所
    d="$HOME/.aws/emulator/9297"
    {
      # 接続先。これが設定されていると、SDK と CLI は、実 AWS ではなく、エミュレーターに接続する
      echo "AWS_ENDPOINT_URL=https://127.0.0.1:9297"

      # エミュレーターが、起動時に作る、認証情報と設定のファイル
      echo "AWS_SHARED_CREDENTIALS_FILE=$d/emulator_credentials"
      echo "AWS_CONFIG_FILE=$d/config"

      # 自己署名の CA を信頼させる。AWS CLI と boto3 は、この AWS_CA_BUNDLE を使う
      echo "AWS_CA_BUNDLE=$d/ca.pem"
      # Node.js の SDK は、AWS_CA_BUNDLE を使わないので、こちらにも設定する（後述の Node.js の章）
      echo "NODE_EXTRA_CA_CERTS=$d/ca.pem"

      # EC2 のインスタンスメタデータから、認証情報を探しに行かせない
      echo "AWS_EC2_METADATA_DISABLED=true"
      # ホスト名にプレフィックスが付く API（Cloud Map、Step Functions など）が、
      # 別のホストに飛ばないようにする（公式の案内）
      echo "AWS_DISABLE_HOST_PREFIX_INJECTION=true"

      # リージョン。エミュレーターは、1つのリージョンで使う
      echo "AWS_REGION=us-east-1"
      echo "AWS_DEFAULT_REGION=us-east-1"
    } >> "$GITHUB_ENV"    # 後続のステップに、環境変数として引き継ぐ

このステップは、起動と起動待ちのあとに置きます。認証情報のファイルと CA が、起動のたびに作り直されるためです。ジョブには、実 AWS の認証情報を置かず、aws-actions/configure-aws-credentials も使いません。

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

AWS の認証情報がないのに、エミュレーターのアカウント（000000000000）として応答しました。続けて S3 のバケットの作成と一覧も成功しています。

CI のステップ4: SDK でテストを回す（層1）

Gemini_Generated_Image_f6jlrhf6jlrhf6jl.jpg

層1は、アプリではなく、エミュレーターの挙動を SDK で直接確かめる段階です。モックでは見えにくい挙動を、CI で拾えるかを見ます。boto3 と pytest で、5つのテストを書きました（コメントは記事用に足したもので、動きは検証したコードと同じです）。

まず、共通のコードです。

共通のコード（tests/test_integration.py の冒頭） 
import time
import uuid

import boto3
import pytest
from botocore.exceptions import ClientError


def _suffix():
    # バケット名やテーブル名が重ならないようにするための、ランダムな8文字。
    # 1回の実行の中で、バケットやテーブルを何度も作るため、名前を変えている
    return uuid.uuid4().hex[:8]


# boto3 のクライアントは、テスト全体で1つを使い回す（scope="session"）。
# エンドポイントと認証情報は、コードに書かない。「向け先を環境変数で切り替える」で設定した環境変数
# （AWS_ENDPOINT_URL、AWS_SHARED_CREDENTIALS_FILE など）を、boto3 が自動で読む。
# そのため、同じコードが、環境変数を外せば、実 AWS にも向く
@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")

S3: 往復と、存在しないキー

S3 では、作成、書き込み、読み出しが状態を保って通るかと、存在しないキーを読んだときに、AWS と同じ NoSuchKey が返るかを見ます。

S3 のテスト（test_s3_roundtrip、test_s3_missing_key_is_nosuchkey） 
def test_s3_roundtrip(s3):
    bucket = f"ci-{_suffix()}"
    # バケットを作る
    s3.create_bucket(Bucket=bucket)
    # オブジェクトを書き込む
    s3.put_object(Bucket=bucket, Key="a.txt", Body=b"hello")
    # 同じキーを読み出す。書いた内容が返れば、「作る、書く、読む」が、状態を保って通っている
    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)
    # 存在しないキーを読むと、例外（ClientError）になる
    with pytest.raises(ClientError) as e:
        s3.get_object(Bucket=bucket, Key="nope")
    # エラーの種類は、レスポンスの Error.Code に入っている。AWS と同じ NoSuchKey かを確認する
    assert e.value.response["Error"]["Code"] == "NoSuchKey"

DynamoDB: 条件付き書き込み

DynamoDB では、条件付き書き込みを試します。「同じ ID のアイテムがまだないときだけ書く」条件で2回書くと、1回目は成功し、2回目は ConditionalCheckFailedException になります。二重登録を防ぐ処理の基本の動きで、モックでは条件式の評価まで再現するのに手間がかかります。

DynamoDB のテスト（test_dynamodb_conditional_write） 
def test_dynamodb_conditional_write(ddb):
    table = f"ci-{_suffix()}"
    # ID（文字列）だけをキーにした、オンデマンド課金のテーブルを作る
    ddb.create_table(
        TableName=table,
        BillingMode="PAY_PER_REQUEST",
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
    )
    # 作成直後のテーブルの状態を、ログに出して記録する。
    # （最初の実行で、直後の書き込みが ResourceNotFoundException になったため、状態を確認できるようにした）
    status = ddb.describe_table(TableName=table)["Table"]["TableStatus"]
    print(f"table status right after create_table: {status}")
    # ACTIVE になるまで待つ。1秒ごとに、最大 30 回確認する。
    # 作ってすぐ書き込むのではなく、待ってから書き込む
    ddb.get_waiter("table_exists").wait(
        TableName=table, WaiterConfig={"Delay": 1, "MaxAttempts": 30}
    )
    item = {"id": {"S": "1"}}
    # 条件付き書き込み。「同じ ID のアイテムが、まだないときだけ」書く。1回目は成功する
    ddb.put_item(
        TableName=table, Item=item,
        ConditionExpression="attribute_not_exists(id)",
    )
    # 同じ ID を、もう一度書く。アイテムが既にあるので、条件を満たさず、例外になる
    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"

SQS: メッセージの再配信

SQS では、メッセージの再配信を試します。ビジビリティタイムアウトは、受信したメッセージが一定時間ほかの受信から見えなくなる仕組みで、この間に削除しないと、メッセージが再び見えて再配信されます。削除を忘れた処理は、同じメッセージを何度も処理します。AWS organization の DEV.to の記事が、「モックでは通るが、エミュレーターでは失敗する」例として挙げている観点です。

SQS のテスト（test_sqs_redelivery_after_visibility_timeout） 
def test_sqs_redelivery_after_visibility_timeout(sqs):
    """メッセージを削除しないと、ビジビリティタイムアウト後に再配信される。
    AWS organization の DEV.to 記事（モックが見逃す不具合の例）と同じ観点。"""
    # ビジビリティタイムアウトを 2 秒にしたキューを作る（待ち時間を短くするため）
    q = sqs.create_queue(
        QueueName=f"ci-{_suffix()}",
        Attributes={"VisibilityTimeout": "2"},
    )["QueueUrl"]
    sqs.send_message(QueueUrl=q, MessageBody="order-1")

    # 1回目の受信。受信した時点で、メッセージは 2 秒間、ほかの受信から見えなくなる。
    # ReceiptHandle は、この受信を識別する値（再配信されると、値が変わる）
    first = sqs.receive_message(QueueUrl=q, WaitTimeSeconds=1)["Messages"][0]
    handle1 = first["ReceiptHandle"]

    # 削除せずに、3 秒待つ（ビジビリティタイムアウトの 2 秒を、超える）
    time.sleep(3)

    # 2回目の受信。メッセージが、再び見えるようになっているはず。
    # 見えるまでに時間がかかる場合に備えて、最大 5 回、受信を繰り返す
    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, "再配信されなかった"
    # 同じメッセージが、再配信されている
    assert second["Body"] == "order-1"
    # 受信が別なので、ReceiptHandle は、1回目と違う値になる
    assert second["ReceiptHandle"] != handle1

未実装の API: ステータスを記録する

最後は未実装の API です。公式は、未実装の API は 501 を返すと案内しています。ただ、リリースで実装されると結果が変わるため、Athena の list_work_groups を呼び、200 か 501 のどちらかなら通して、ステータスを記録するだけにしています。

未実装 API のテスト（test_unimplemented_api_status_is_recorded） 
def test_unimplemented_api_status_is_recorded():
    """未実装のAPIは501になる想定。リリースで実装される可能性があるため、
    ステータスを出力して記録するだけにする（-s で表示される）。"""
    athena = boto3.client("athena")
    try:
        # 実装済みなら、そのまま成功する（HTTP 200）
        athena.list_work_groups()
        status = 200
    except ClientError as e:
        # 未実装なら、エラーになる。レスポンスから、HTTP ステータスを取り出す
        status = e.response["ResponseMetadata"]["HTTPStatusCode"]
    # -s を付けて pytest を実行すると、この出力が、ログに表示される
    print(f"athena list_work_groups -> HTTP {status}")
    assert status in (200, 501)

実行結果です。5件とも成功しました。

============================= 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 ===============================

SQS は、3 秒待つと再配信されて、ReceiptHandle が変わりました。Athena は 501 になると思っていましたが、HTTP 200 でした。この版では実装済みのようです。

DynamoDB のテストは、最初の実行で一度失敗しました。テーブルを作った直後の put_item が ResourceNotFoundException になったのです。原因は分かっていません。ログには CreateTable の記録しかなく、失敗の記録はありませんでした。ACTIVE になるまで待つ処理（table_exists のウェイター）を入れてからは、10 回の実行がすべて成功しています。ただ、作成直後の状態は ACTIVE と出ていたので、待つ処理が問題を隠しているだけかもしれません。

CI のステップ4の補足: Node.js では NODE_EXTRA_CA_CERTS が要る（層1）

エミュレーターの証明書は、自己署名です。AWS CLI と boto3 は AWS_CA_BUNDLE を見ますが、Node.js の SDK では、そうなりませんでした。同じ S3 の呼び出しを、Node.js で試します。

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

// エンドポイントは AWS_ENDPOINT_URL、リージョンは AWS_REGION から、SDK が読む。
// forcePathStyle: バケット名をホスト名に入れない、パススタイルにする
const client = new S3Client({ forcePathStyle: true });

// バケットの一覧を取得する。エミュレーターに接続できれば、これまでに作ったバケットの数が返る
const r = await client.send(new ListBucketsCommand({}));
console.log("buckets:", (r.Buckets ?? []).length);

NODE_EXTRA_CA_CERTS を設定したときは、成功しました（buckets: 3）。この環境変数だけを外すと、次のエラーになります。

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

このとき AWS_CA_BUNDLE は設定したままでした。それでも失敗したので、Node.js の SDK は AWS_CA_BUNDLE を使っていません。エミュレーターの CA は Node.js が信頼する CA に含まれないため、NODE_EXTRA_CA_CERTS で追加します。Java の SDK では、公式が truststore.p12 の設定を案内しています（試していません）。

CI のステップ5: CDK でデプロイして API をテストする（層2）

Gemini_Generated_Image_t9zxg2t9zxg2t9zx.jpg

ここからが、PR ごとの結合テストに近い部分です。前回の記事で動いた、アイテムを登録・取得する小さな API（API Gateway の POST /i と GET /i/{k}、Lambda、DynamoDB）を、CI で CDK からデプロイして呼び出します。CDK は、前回動いた組み合わせ（CLI 2.1148.1、aws-cdk-lib 2.275.0）に固定し、npm ci で入れます。

エミュレーターが実行する Lambda には、エンドポイントも認証情報も自動では渡されません。Lambda から DynamoDB を呼ぶには、環境変数を Lambda にも設定します。認証情報や CA のパスは、ランナーのホームディレクトリの下にあるので、デプロイ時に -c emuDir=... で CDK に渡しました。

# CDK の bootstrap（CDK が使う S3 バケットなどを作る）と deploy を、エミュレーターに対して実行する
- name: CDK bootstrap and deploy
  working-directory: cdk
  env:
    CI: "true"    # CI として実行する
  run: |
    # アカウント ID は、エミュレーターのアカウント（000000000000）を指定する
    npx cdk bootstrap aws://000000000000/us-east-1 --no-notices

    # $CDK_CTX は、-c emuDir=<エミュレーターのディレクトリ> の形の、CDK への引数
    # （環境変数にまとめてある）。次のコードで受け取る
    # --require-approval never: IAM の変更の確認を、聞かずに進める
    npx cdk deploy --require-approval never --no-notices $CDK_CTX

// デプロイ時に、-c emuDir=<エミュレーターのディレクトリ> で渡された値を受け取る
const emuDir: string | undefined = this.node.tryGetContext('emuDir');

// Lambda に渡す環境変数。テーブル名は、いつも渡す
const environment: Record<string, string> = { TABLE_NAME: items.tableName };
if (emuDir) {
  // エミュレーターの上で動かすときだけ、エンドポイントと認証情報を Lambda に渡す。
  // エミュレーターは、Lambda の実行環境に、これらを自動では渡さないため
  environment.AWS_ENDPOINT_URL = 'https://127.0.0.1:9297';
  environment.AWS_SHARED_CREDENTIALS_FILE = `${emuDir}/emulator_credentials`;
  environment.AWS_CONFIG_FILE = `${emuDir}/config`;
  // Lambda の中の Node.js も、自己署名の CA を信頼させる必要がある（「Node.js では NODE_EXTRA_CA_CERTS が要る」の章）
  environment.NODE_EXTRA_CA_CERTS = `${emuDir}/ca.pem`;
}

前回は、AWS::Lambda::Permission の statementId が100文字を超える問題と、同じスタックでのステージ名の重複で、デプロイが失敗しました。今回は、論理 ID を短くし、API を1つにして避けています。今の版で解消したのかは、切り分けていません。

デプロイするアプリの全体は、次のとおりです。長いので、折りたたんでいます。「PR で壊してみる」の章で壊す箇所は、スタックの grantReadWriteData と、Lambda の json(201, ...) です。

CDK のスタック（lib/app-stack.ts） 
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);

    // CI では -c emuDir=<エミュレーターのディレクトリ> を渡す。
    // エミュレーターが実行する Lambda から、エミュレーターの DynamoDB を呼ぶために必要
    const emuDir: string | undefined = this.node.tryGetContext('emuDir');

    // アイテムを保存するテーブル。ID（文字列）だけをキーにする。
    // DESTROY: cdk destroy のときに、テーブルも削除する（CI の使い捨て環境のため）
    const items = new ddb.Table(this, 'Items', {
      partitionKey: { name: 'id', type: ddb.AttributeType.STRING },
      billingMode: ddb.BillingMode.PAY_PER_REQUEST,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Lambda に渡す環境変数（この章の前半で説明した部分）
    const environment: Record<string, string> = { 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`;
    }

    // API の処理を担当する Lambda。コードは lambda/ ディレクトリから取り込む
    const appFn = new lambda.Function(this, 'App', {
      runtime: lambda.Runtime.NODEJS_22_X,
      handler: 'items.handler',
      code: lambda.Code.fromAsset('lambda'),
      environment,
    });

    // Lambda に、テーブルの読み書きの権限を付ける。
    // 「PR で壊してみる」の PR 1 では、この1行を削除して、権限の付け忘れを再現した
    items.grantReadWriteData(appFn);

    // API Gateway。ステージ名は v1。
    // POST /i でアイテムを保存し、GET /i/{k} でアイテムを取得する。どちらも同じ Lambda が処理する
    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));

    // テストが、デプロイ後の API の ID とテーブル名を取得できるように、出力しておく
    new cdk.CfnOutput(this, 'ApiId', { value: api.restApiId });
    new cdk.CfnOutput(this, 'TableName', { value: items.tableName });
  }
}
Lambda のコード（lambda/items.js）

検証で使ったコードは、失敗したときの調査用に、実行環境の情報も返していました。ここでは、その部分を省いています。

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

// エンドポイントと認証情報は、Lambda の環境変数から、SDK が読む
const ddb = new DynamoDBClient({});

// API Gateway に返すレスポンスを作る、補助関数
const json = (statusCode, obj) => ({
  statusCode,
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(obj),
});

exports.handler = async (event) => {
  try {
    // POST /i: リクエストの本文から id と text を取り出して、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 } },
      }));
      // 保存できたら 201（Created）を返す。「PR で壊してみる」の PR 2 では、これを 200 に変えた
      return json(201, { saved: id });
    }

    // GET /i/{k}: パスの k を ID として、DynamoDB から取得する
    const id = event.pathParameters && event.pathParameters.k;
    const r = await ddb.send(new GetItemCommand({
      TableName: process.env.TABLE_NAME,
      Key: { id: { S: id } },
    }));
    // アイテムがあれば 200、なければ 404 を返す
    return r.Item ? json(200, { id, text: r.Item.text.S }) : json(404, { message: 'not found' });
  } catch (e) {
    // DynamoDB の呼び出しに失敗したら（権限不足など）、500 とエラーの種類を返す。
    // 「PR で壊してみる」の PR 1 では、ここに AccessDeniedException が返った
    return json(500, { error: e.name, message: e.message });
  }
};

テストは、デプロイ後の API を、HTTP で呼び出して確かめます。3つのテストを書きました。

テスト 確かめること
test_post_then_get POST で保存して、GET で同じ内容が返る（Lambda、API Gateway、DynamoDB が通しでつながっている）
test_get_missing_item_is_404 存在しない ID の GET が、404 になる
test_item_is_stored_in_dynamodb POST した内容が、API を介さずに DynamoDB を直接読んでも、保存されている

API は、Host ヘッダーで対象を指定して、エミュレーターのエンドポイントを呼びます。コードは、次のとおりです。

API のテスト（tests/test_api.py） 
import os
import uuid

import boto3
import pytest
import requests

# テスト対象のスタックの名前
STACK = os.environ.get("STACK_NAME", "EmulatorCiApp")


@pytest.fixture(scope="module")
def outputs():
    # デプロイしたスタックの出力（ApiId と TableName）を、CloudFormation から取得する。
    # エミュレーターの 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):
    # エミュレーターのエンドポイント（https://127.0.0.1:9297）
    base = os.environ["AWS_ENDPOINT_URL"]
    # エミュレーターは、API Gateway を1つのエンドポイントで受ける。
    # どの API かは、Host ヘッダーで指定する
    host = f"{outputs['ApiId']}.execute-api.us-east-1.amazonaws.com"
    # 自己署名の CA（AWS_CA_BUNDLE と同じファイル）
    ca = os.environ["AWS_CA_BUNDLE"]

    def _call(method, path, **kwargs):
        return requests.request(
            method,
            # エンドポイントに、ステージ名（v1）とパスを付ける
            f"{base}/v1{path}",
            headers={"Host": host},
            verify=ca,
            timeout=30,
            **kwargs,
        )

    return _call


def test_post_then_get(call):
    # 他のテストと ID が重ならないように、ランダムな ID を使う
    item_id = f"item-{uuid.uuid4().hex[:8]}"

    # POST で保存する。201 と、保存した ID が返る
    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}

    # GET で取得する。保存した内容が、そのまま返る
    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):
    # 保存していない ID を取得すると、404 になる。
    # このテストは POST を使わないので、POST の返却ステータスを変えても影響を受けない
    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

    # API を介さず、DynamoDB を直接読む。
    # API のレスポンスだけでなく、実際にテーブルへ保存されたことを確かめる
    ddb = boto3.client("dynamodb")
    got = ddb.get_item(
        TableName=outputs["TableName"], Key={"id": {"S": item_id}}
    )
    assert got["Item"]["text"]["S"] == "stored"
操作 結果
cdk bootstrap と cdk deploy 1回目で成功。bootstrap と deploy を合わせて 29〜40 秒（deploy の Total time は 13〜16 秒）
POST、GET、存在しない ID の GET 201、200、404。DynamoDB への保存も、get_item で確認
再デプロイ（変更なし） no changes（4〜7 秒）
cdk diff --fail There were no differences
cdk destroy destroyed（10〜14 秒） 
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 ===============================

前回の Terraform では、apply の直後に plan を実行すると DynamoDB で差分が出ましたが、CDK では、再デプロイも cdk diff も差分なしでした。cdk diff は CloudFormation の変更セットを作って差分を取るので、変更セットの作成には対応しているようです。

再デプロイ、cdk diff、削除のステップには continue-on-error: true を付けているため、失敗してもジョブは緑のままです。失敗したステップの conclusion も success と出るので、各ステップの outcome を、ジョブの Summary に書き出して確認しています。

PR で壊してみる（層2）

Gemini_Generated_Image_m82ulym82ulym82u.jpg

CI の価値は、PR に不具合があるときに、赤になって気づけることです。そこで、通る PR と、意図的に壊した PR を出しました。壊した PR は、マージせずに閉じています。

README だけを変えた PR は、緑でした（約 2 分 14 秒）。ジョブの環境変数に、AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_SESSION_TOKEN、AWS_PROFILE はなく、AWS の認証情報なしで回っています。

aws-emulator-github-actions-ci-pr-runs.png

PR で起動した実行の一覧です。V10a が README だけの PR、V10b が権限付与を外した PR（次の表の PR 1）、V10c が返すステータスを変えた PR（PR 2）です。V10b と V10c は赤になり、revert すると緑に戻りました。

壊した PR は、変更の種類が違う2つです。

PR 1: 権限付与を外す（IaC の変更） PR 2: 返すステータスを変える（アプリのコードの変更）
変更 DynamoDB への権限付与（grantReadWriteData）を削除 Lambda の POST の返却ステータスを 201 から 200 に変更
結果 赤。API のテスト3件とも失敗 赤。3件中2件が失敗（404 を確認するテストは成功）
エラー AccessDeniedException: User: arn:aws:sts::000000000000:assumed-role/… is not authorized to perform: dynamodb:PutItem on resource: … because no identity-based policy allows the dynamodb:PutItem action。GET は dynamodb:GetItem。API は 500 assert 200 == 201
失敗までの実行時間 84 秒 81 秒
revert 後 緑に戻った（123 秒） 緑に戻った（128 秒）

aws-emulator-github-actions-ci-pr-failed-run.png

PR 1 の赤になった実行の画面です。

PR 1 は、Lambda のロールの権限不足を、エミュレーターが AWS の形式のメッセージで検出しました。ただ、公式は、IAM を AWS と同じようには評価しないと案内しています。確かめられたのは、「権限の付け忘れに、CI で気づけた」ことまでです。

アプリのコードの誤りは、テストがその箇所を確認しているかで、検出できるかが決まります。PR 2 で 404 のテストが成功したのは、POST の変更の影響を受けないためです。

deb のキャッシュは効くか

deb は約 174 MiB あるので、毎ジョブ取得すると遅いのではと考えて、actions/cache でキャッシュしてみました。

結果は、キャッシュなしでダウンロード約 1 秒とインストール 20 秒、キャッシュありで復元約 3 秒とインストール 16 秒でした。ダウンロードのほうが復元より速く、時間の大半は apt の展開で、インストール時間は 16〜37 秒とばらつくため、20 秒と 16 秒の差は誤差の範囲です。効果が見えなかったので、キャッシュはしていません。

実務での使い方

今回の検証では、IaC のデプロイと、SDK を使うアプリの結合テストを、PR ごとに回せました。CDK でデプロイした API は、デプロイから削除まで約 2 分で通り、権限の付け忘れも、返すステータスの誤りも、PR の時点で検出できました。

一方、公式は、IAM を AWS と同じようには評価しないと案内しています。緑が示すのは「エミュレーターで通った」ことまでなので、権限が正しいことの確認は、別に必要です。

ワークフローには、実 AWS の認証情報（OIDC を含む）を置きません。公式も、テストジョブに実認証情報を置かないよう案内しています。状態はメモリ上だけなので、ジョブごとに作り直す前提で組みます。latest は1日に何度も更新されるため、結果が変わることがあります。

テストを書くときは、テーブルを作ったら ACTIVE を待ってから書き込むこと、Node.js の SDK では NODE_EXTRA_CA_CERTS を設定することに注意してください。

公式の指針との対応

Gemini_Generated_Image_mwarflmwarflmwar.jpg

AWS Well-Architected Framework と DevOps Guidance の、テストとデプロイに関する項目と照らしてみました。これは私の整理で、AWS が AWS Emulator と結びつけて述べているわけではありません。

DevOps Guidance の QA.FT.2（結合テスト）には、次の一文があります。

「Consider using dedicated emulation, containers, or cloud-based test environments to make tests more efficient, consistent, and safe.」
出典: https://docs.aws.amazon.com/wellarchitected/latest/devops-guidance/qa.ft.2-validate-system-interactions-and-data-flows-with-integration-tests.html

2023年9月公開の資料で、AWS Emulator を指したものではありませんが、結合テストの手段としてエミュレーションを挙げている点で、今回の使い方と合っています。ほかにも、次の項目が当てはまると考えました。

  • OPS05-BP08（複数の環境を使用する）: 開発者に最小限のコントロールのサンドボックス環境を提供する。AWS アカウントも費用も不要なので合います
  • OPS05-BP02（変更をテストし、検証する）: テストは各セットを10分以内に完了させる。今回はジョブ全体で約 2 分でした
  • REL08-BP02（デプロイの一部として機能テストを統合する）: すべての環境で同じデプロイの仕組みを使う。CDK は環境変数だけでエミュレーターに向けられました
  • OPS06-BP02（デプロイをテストする）: 一時的な並列環境を作る。PR ごとの使い捨て環境に、あたります

これらの項目は、本番に近い環境でのテストも求めています。エミュレーターは IAM を AWS と同じようには評価しないので、その部分は実 AWS で確認する必要があります。

さいごに

公式のドキュメントにワークフロー例がなくても、公式の deb を入れて起動し、環境変数で向け先を切り替えれば、PR ごとの結合テストを回せました。AWS の認証情報は要らず、CDK でデプロイした API は約 2 分で通り、壊した PR は赤になりCIの失敗が確認できました。

ただ、緑が示すのは「エミュレーターで通った」ことまでで、実 AWS での確認は引き続き必要です。AWS Emulator は1日に何度も更新されるので、本記事の結果は、1.0.353（2026-11-14）のものです。Werner Vogels 氏のブログに載った Swami Sivasubramanian 氏の投稿は、詳細を re:Invent で説明すると書いていて、CI に関する公式の案内が出れば、内容が変わるかもしれません。自分の構成で試して、公式の記録にない差異を見つけたら、GitHub の Issues に報告すると、改善につながると思います。

英語記事

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

参考文献

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?