kakakakakku blog

Weekly Tech Blog: Keep on Learning!

testcontainers-python: pytest 実行時に使い捨て可能な LocalStack を起動する

Testcontainers を使うと,テストコードを実行するときに必要になるデータベース・キャッシュ・キューなどの依存関係をコード上で管理できて,実行後にはコンテナを自動的に消してくれるという使い捨て可能な仕組みを簡単に作れる❗️Testcontainers のサイトに載っている「Test dependencies as code」という表現はピッタリだと思う👌

testcontainers.com

Testcontainers は Java / Go / .NET / Rust など多くの言語をサポートしているけど,今回は Python 用の testcontainers-python を試してみた.検証に使ったコードを紹介しつつ,簡単にまとめておく✍

また Testcontainers Cloud もあったりする🌩

testcontainers.com

前提

今回は以下の前提で試す💡なお testcontainers-python は MySQL / PostgreSQL / Redis / Kafka / RabbitMQ / LocalStack など多くの依存関係をサポートしているので,組み合わせてテストコードを実行することもできる.今回は LocalStack に限定する.

  • Amazon DynamoDB を操作するコードをテストする
  • Amazon DynamoDB のエミュレーターとして LocalStack を使う
  • テストフレームワークとして pytest を使う
  • testcontainers-pythonLocalStackContainer を使う

ディレクトリ構成は以下のようにした.特に決まってなく自由に変更できる👌

.
├── README.md
├── pyproject.toml
├── requirements-test.txt
├── src
│   └── app.py
├── tests
│   └── test_app.py
└── venv

ドキュメント

testcontainers-python に関しては以下のドキュメントを読むとイメージがつかめると思う.

testcontainers-python.readthedocs.io

testcontainers.com

しかし,残念ながら testcontainers-python の LocalStackContainer に関するドキュメントはほとんどなく(見つけられず)実際に GitHub のコードを読みながらメソッドなどを探したりしていた.

github.com

サンプルコード

👾 app.py

コードには特に意味はないけど,Amazon DynamoDB の Forum テーブルからアイテムを取得する search_forum() 関数を今回のテスト対象とする.Forum テーブルというのは Amazon DynamoDB のドキュメントに載っているサンプルでそのまま使うことにした.

そして,boto3 client は環境変数 ENV によって3種類作れるようにしてある👌

  • local: ローカル開発用 (LocalStack)
  • test: テスト用 (testcontainers-python / LocalStack)
  • その他: 実際の AWS アカウント

また testcontainers-python の LocalStackContainer を使うと http://localhost:63033http://localhost:63058 など実行時のポートが変化するため,後述する test_app.py で環境変数 TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL を設定することにした👌もっとイイ方法もありそう.ちなみに LocalStackContaineredge_port0.0.0.0:63167->24566/tcp のようにネットワークが構成されるため "コンテナ側のポート" を指定するオプションだった.

import boto3
import os

TABLE_NAME = 'Forum'


if os.environ['ENV'] == 'local':
    dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:4566')
elif os.environ['ENV'] == 'test':
    dynamodb = boto3.client('dynamodb', endpoint_url=os.environ['TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL'])
else:
    dynamodb = boto3.client('dynamodb')


def search_forum(name):
    return dynamodb.get_item(
        TableName=TABLE_NAME,
        Key={'Name': {'S': name}}
    )

👾 test_app.py

pytest 実行時に呼び出すフィクスチャとして @pytest.fixture デコレータで setup() 関数を実装した.LocalStackContainer で LocalStack のコンテナイメージを設定しているため,テストコードを実行する前に自動的に LocalStack コンテナが起動される👌ちなみに LocalStackContainerget_client() 関数は boto3 client を返しているため,boto3 に慣れていれば普段と同じように実装できる.

そして,Amazon DynamoDB テーブル Forum を作って,サンプルデータ(アイテム)を1つ登録している.テスト観点によっては Faker などを使ってリアルなサンプルデータを登録すると良さそう.

最後にテストケースとしては search_forum() 関数を呼び出して「アイテムを取得できる場合」「アイテムを取得できない場合」を確認している✔️

import os
import pytest
from testcontainers.localstack import LocalStackContainer

TABLE_NAME = 'Forum'


@pytest.fixture(scope='module', autouse=True)
def setup():
    with LocalStackContainer(image='localstack/localstack:3', region_name='ap-northeast-1') as localstack:
        os.environ['TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL'] = localstack.get_url()

        dynamodb = localstack.get_client('dynamodb')

        dynamodb.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {
                    'AttributeName': 'Name',
                    'KeyType': 'HASH',
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'Name',
                    'AttributeType': 'S',
                }
            ],
            BillingMode='PAY_PER_REQUEST',
        )

        item = {
            'Name': {'S': 'Amazon DynamoDB'},
            'Category': {'S': 'Amazon Web Services'},
            'Threads': {'N': '2'},
            'Messages': {'N': '4'},
            'Views': {'N': '1000'},
        }

        dynamodb.put_item(TableName=TABLE_NAME, Item=item)

        yield localstack


def test_search_forum():
    from app import search_forum
    
    item = search_forum('Amazon DynamoDB')
    assert item['Item']['Category']['S'] == 'Amazon Web Services'
    assert item['Item']['Views']['N'] == '1000'

    item = search_forum('Amazon S3')
    assert 'Item' not in item

testcontainers-python と LocalStack の検証はザッとこんな感じ \( 'ω')/

動作確認

期待通り実行できた👏

$ ENV=test pytest -p no:warnings
=========================================================================================== test session starts ============================================================================================
(中略)
configfile: pyproject.toml
collected 1 item

tests/test_app.py .                                                                                                                                                                                  [100%]

============================================================================================ 1 passed in 4.97s =============================================================================================

現時点だと Python 3.12 で boto3 の DeprecationWarning が出るため -p no:warnings で抑止している🛑

github.com

Lambda オーソライザーのポリシーを簡単に出力できる Powertools for AWS Lambda (Python) の APIGatewayAuthorizerResponse

Amazon API Gateway の Lambda オーソライザー(旧カスタムオーソライザー)を使ってアクセス制御をするときに Lambda オーソライザーの仕様に沿ったポリシーを出力する必要がある💡詳しくは以下のドキュメントに載っている.

docs.aws.amazon.com

今まではドキュメントに載っているコードを参考に実装することが多かったけど,Powertools for AWS Lambda (Python)Event Source Data ClassesAPIGatewayAuthorizerResponse を使うと比較的簡単にポリシーを出力できて良かった❗️

docs.powertools.aws.dev

サンプルコード

今回トークンベースの Lambda オーソライザーで検証したときに使ったコードを載せておく📝

from aws_lambda_powertools.utilities.data_classes import event_source
from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (APIGatewayAuthorizerTokenEvent, APIGatewayAuthorizerResponse)
from aws_lambda_powertools.utilities.typing import LambdaContext


@event_source(data_class=APIGatewayAuthorizerTokenEvent)
def lambda_handler(event: APIGatewayAuthorizerTokenEvent, context: LambdaContext):
    arn = event.parsed_arn

    policy = APIGatewayAuthorizerResponse(
        principal_id='user',
        region=arn.region,
        aws_account_id=arn.aws_account_id,
        api_id=arn.api_id,
        stage=arn.stage
    )

    if event.authorization_token == 'allow':
        policy.context = {
            'key1': 'value1',
            'key2': 'value2',
            'key3': 'value3'
        }
        policy.allow_all_routes()
    else:
        policy.deny_all_routes()

    return policy.asdict()

ドキュメントに載っている Authorization ヘッダーに allow という文字列が設定されていれば OK という簡易的な実装だけど,ポリシーを出力するときは APIGatewayAuthorizerResponse に値を設定して allow_all_routes() 関数もしくは deny_all_routes() 関数で出力すれば良くて簡単👍

また認証後に実行される AWS Lambda 関数などに追加情報を渡す場合は context に dict で Key-Value を設定すれば OK👌

docs.aws.amazon.com

動作確認

$ curl -H 'Authorization: allow' ${ENDPOINT}
ok

$ curl -H 'Authorization: deny' ${ENDPOINT}
{"Message":"User is not authorized to access this resource with an explicit deny"}

source-version-override: aws-actions/aws-codebuild-run-build でプルリクエストブランチを AWS CodeBuild のビルド対象にする

AWS CodeBuild Run Build for GitHub Actions (aws-actions/aws-codebuild-run-build) を使って GitHub Actions から AWS CodeBuild のビルドを実行すると buildspec.yml やビルド環境タイプを上書きできて便利〜という話は前にまとめた👌

kakakakakku.hatenablog.com

今回は source-version-override パラメータを活用して,プルリクエストを出したときにプルリクエストブランチを AWS CodeBuild のビルド対象にする仕組みを試してみた.以下に検証用の GitHub Actions ワークフローを載せておく📝プルリクエストをマージする前に動作確認ができるようになるから便利なパラメータだと思う❗️

name: Start AWS CodeBuild build

on:
  workflow_dispatch:
  push:
    branches:
      - master
  pull_request:
    branches:
      - master

permissions:
  id-token: write
  contents: read

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_ROLE_ARN }}
          aws-region: ap-northeast-1
      - name: Start AWS CodeBuild build
        uses: aws-actions/aws-codebuild-run-build@v1
        with:
          project-name: sandbox
          source-version-override: ${{ github.head_ref }}
        if: github.event_name == 'pull_request'
      - name: Start AWS CodeBuild build
        uses: aws-actions/aws-codebuild-run-build@v1
        with:
          project-name: sandbox
          source-version-override: pr/${{ github.event.pull_request.number }}
        if: github.event_name == 'pull_request'
      - name: Start AWS CodeBuild build
        uses: aws-actions/aws-codebuild-run-build@v1
        with:
          project-name: sandbox
          source-version-override: ${{ github.ref_name }}
        if: github.event_name == 'push'

AWS CodeBuild Run Build for GitHub Actions の source-version-override パラメータには GitHub の場合「コミット ID/プルリクエスト ID/ブランチ名/タグ名」を指定できる.AWS CodeBuild の API Reference (StartBuild) に以下のように書いてあった📝

The commit ID, pull request ID, branch name, or tag name that corresponds to the version of the source code you want to build. If a pull request ID is specified, it must use the format pr/pull-request-ID (for example pr/25). If a branch name is specified, the branch's HEAD commit ID is used. If not specified, the default branch's HEAD commit ID is used.

プルリクエストブランチ

プルリクエストを作った場合は github.event_name == 'pull_request' で判定しつつ,source-version-override パラメータにブランチ名を表す ${{ github.head_ref }} を設定すれば OK👌

- name: Start AWS CodeBuild build
  uses: aws-actions/aws-codebuild-run-build@v1
  with:
    project-name: sandbox
    source-version-override: ${{ github.head_ref }}
  if: github.event_name == 'pull_request'

もしブランチ名ではなく「プルリクエスト ID」を指定する場合は pr/pull-request-ID というフォーマットにする必要がある.source-version-override パラメータにプルリクエスト ID を表す pr/${{ github.event.pull_request.number }} を設定すれば OK👌

- name: Start AWS CodeBuild build
  uses: aws-actions/aws-codebuild-run-build@v1
  with:
    project-name: sandbox
    source-version-override: pr/${{ github.event.pull_request.number }}
  if: github.event_name == 'pull_request'

メインブランチ

プルリクエストをマージした場合は github.event_name == 'push' で判定しつつ,source-version-override パラメータにブランチ名を表す ${{ github.ref_name }} を設定すれば OK👌

- name: Start AWS CodeBuild build
  uses: aws-actions/aws-codebuild-run-build@v1
  with:
    project-name: sandbox
    source-version-override: ${{ github.ref_name }}
  if: github.event_name == 'push'

動作確認

期待通りに動いたー👏

AWS CDK で Amazon EventBridge Pipes の「ターゲット入力トランスフォーマー」を設定する

AWS CDK で Amazon SQS x Amazon EventBridge Pipes x AWS Step Functions の構成を設定する流れは前にまとめた📝

kakakakakku.hatenablog.com

前にまとめた設定では Amazon SQS キューに登録したメッセージをデフォルト設定のまま Amazon EventBridge Pipes 経由で AWS Step Functions に流しているけど,実際に使ってみると Amazon SQS のメッセージ形式のまま AWS Step Functions に流れてくるため,AWS Step Functions 側でインプットの取り回しがしにくく微妙に使いにくいことに気付く💨

具体例

例えば以下の JSON を Amazon SQS キューに登録する.

{
    "key1": "value1",
    "key2": "value2",
    "key3": "value3"
}

AWS Step Functions の入力としては以下のような JSON が流れてくる(一部の値は書き換えた).AWS Step Functions 側では body に含まれている JSON のみで十分ということも多いと思う💡

[
  {
    "messageId": "0f6ddca6-8fc2-45e3-8c51-226eca45dbb0",
    "receiptHandle": "xxxxx",
    "body": "{\n    \"key1\": \"value1\",\n    \"key2\": \"value2\",\n    \"key3\": \"value3\"\n}",
    "attributes": {
      "ApproximateReceiveCount": "1",
      "SentTimestamp": "1709209227935",
      "SenderId": "xxxxx",
      "ApproximateFirstReceiveTimestamp": "1709209227937"
    },
    "messageAttributes": {},
    "md5OfBody": "b4fc128c9cb169639ef7083a9a4d78dd",
    "eventSource": "aws:sqs",
    "eventSourceARN": "arn:aws:sqs:ap-northeast-1:000000000000:sandbox-cdk-sqs-pipes-stepfunctions-queue",
    "awsRegion": "ap-northeast-1"
  }
]

ターゲット入力トランスフォーマーを使う

そんなときは Amazon EventBridge Pipes で「ターゲット入力トランスフォーマー (Target Input Transformer)」を設定すれば OK👌今回は Amazon SQS キューに登録したメッセージの body のみ AWS Step Functions に流したいため,以下のように設定する❗️

マネジメントコンソールなら

{
  "body": <$.body>
}

AWS CDK なら

CfnPipeinputTemplate に設定する.AWS CDK コード全体は AWS CDK で Amazon EventBridge Pipes(SQS ソース・Step Functions ターゲット)を設定する - kakakakakku blog 参照📝

new aws_pipes.CfnPipe(this, 'SandboxCdkPipes', {
  name: 'sandbox-cdk-sqs-pipes-stepfunctions-pipes',
  roleArn: pipeRole.roleArn,
  source: queue.queueArn,
  target: stateMachine.stateMachineArn,
  targetParameters: {
    inputTemplate: '{ "body": <$.body> }',
    stepFunctionStateMachineParameters: {
      invocationType: 'FIRE_AND_FORGET',
    }
  }
});

実行結果

「ターゲット入力トランスフォーマー」を設定してもう1度 Amazon SQS キューに同じ JSON を登録すると,以下のように body に含まれている JSON のみ AWS Step Functions に流せるようになった \( 'ω')/

[
  {
    "body": {
      "key1": "value1",
      "key2": "value2",
      "key3": "value3"
    }
  }
]

ちなみに AWS Step Functions 側で JSON 配列になっているのは Amazon EventBridge Pipes の仕様で,ドキュメントには Lambda または Step Functions エンリッチメントまたはターゲットの場合、バッチサイズが 1 であっても、バッチは JSON 配列としてターゲットに配信されます。 と書いてある.これはハマりポイントの1つかも💨

docs.aws.amazon.com

Dependabot で Terraform Provider を自動的にアップデートしよう

Dependabot version updates を使うと Terraform Provider のアップデートを自動化できる❗️設定は比較的簡単で package-ecosystemterraform を設定して,あとは必須の directoryschedule.interval でアップデートの対象ディレクトリとスケジュールを決めれば OK👌個人的な Terraform 検証用プライベートリポジトリに設定して数週間試してみた \( 'ω')/

さらに package-ecosystemgithub-actions を設定すると actions/checkout@v4 など GitHub Actions のアクションも自動的にアップデートできる❗️一度入れたらそのままということもよくあるし助かる〜.

docs.github.com

🤖 .github/dependabot.yml

version: 2
updates:
  - package-ecosystem: terraform
    directory: /
    schedule:
      interval: daily
    open-pull-requests-limit: 2
    target-branch: master
    ignore:
      - dependency-name: "*"
        update-types: ["version-update:semver-major"]
  - package-ecosystem: github-actions
    directory: /
    schedule:
      interval: daily
    open-pull-requests-limit: 2
    target-branch: master

動作確認

Terraform の AWS Provider と Terraform の GitHub リポジトリに設定してる GitHub Actions のアクション (actions/checkout) を自動的にアップデートするプルリクエストが作れたー👏

関連記事

kakakakakku.hatenablog.com