2023.11.24 2023.12.01

AWS IATKを試してみる

ryo

こんにちはryoです。

今日は新しくPublic PreviewになったAWS IATKを試してみたので紹介しようと思います。

Contents

AWS IATKとは？
AWS IATKの基礎概念
実際に試してみる
1. テスト対象のリソース
2. テストコードを書く
まとめ

AWS IATKとは？

AWS IATKは2023年11月16日に発表されたAWS環境で構築したアプリケーションのテストを記述するためのライブラリです。

https://aws.amazon.com/jp/about-aws/whats-new/2023/11/aws-integrated-application-test-kit-preview/

イベント駆動アプリケーションのテストに特化した機能がいくつか盛り込まれており、特にEventBridge周りのテストに必要な関数が用意されています。

またPublic Preview状態なのでAPIインターフェースに破壊的変更が入る可能性があります。

本番利用にはご注意ください。

AWS IATKの基礎概念

IATKのドキュメントには以下の3つがキーコンセプトとして紹介されています。

これらの用語はドキュメントでも多用されているため押さえておく必要があります。

System Under Test (SUT)
Test Harness
Arrange, Act, Assert Testing Pattern

https://awslabs.github.io/aws-iatk/#concepts

System Under Test (SUT)

テスト対象のシステムを表します。

Test Harness

テストのために利用するAWSリソースを表します。テスト対象のAWSリソースではないので注意してください。

これらのリソースはテスト開始時に作成され、テストの終了時に破棄されます。

Arrange, Act, Assert Testing Pattern

こちらはよくAAAパターンと呼ばれているテストの実行パターンを指します。

Arrangeはテストの準備、Actでテスト対象の実行、Assertで期待値の検証を表します。

実際に試してみる

ドキュメントにはいくつかサンプルが用意されています。

例えば、こちらサンプルはAPI GatewayとLambdaで構築したRest APIがイベントのパブリッシャーとなり、別のLambdaがコンシューマーとなりイベントを利用した処理を行います。

https://awslabs.github.io/aws-iatk/tutorial/examples/eb_testing/

今回は上記のシステムの簡略化を行い、EventBridgeのみを作成します。

テスト内容としてはカスタムイベントバスにイベントの送信を行い、正しくイベントが送信出来ているか検証を行います。

テスト対象のシステムとして、EventBridgeのカスタムイベントバスにイベントを送信し、後段でイベントに対して処理を行うものと想定します。

この他にもX-ray利用時のテストサンプルなども記載されているのでぜひご参照ください。

テスト対象のリソース

今回は簡略化のためEventBridgeのみをデプロイします（今回はイベントのコンシューマーは含まない）。

テスト対象のリソースとして、EventBridgeのカスタムイベントバスとイベントルールを用意します。

AWS CDKを利用して構築していきます。またCDK自体のデプロイ方法などは省略します。

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";
import * as event from "aws-cdk-lib/aws-events";

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

    // イベントバスの作成
    const eventBus = new event.EventBus(this, "TestEventBus");
    new cdk.CfnOutput(this, "EventBusName", {
      value: eventBus.eventBusName,
    });

    // イベントルールの作成
    const eventRule = new event.Rule(this, "TestEventRule", {
      eventBus: eventBus,
      eventPattern: {
        source: ["com.test"],
        detailType: ["Test"],
      },
    });
    new cdk.CfnOutput(this, "EventRuleName", {
      value: eventRule.ruleName,
    });
  }
}

このスタックでは、それぞれのリソースを作成し、イベントバス名とイベントルール名をCloudFormationのOutputとして登録します。

テストコードを書く

事前準備

AWS IATKは現在Pythonのライブラリのみを提供しているので、Pythonでテストコードを記述していきます。

今回はpipを利用してライブラリのインストールを行なっています。venv等の仮想環境はお好みで利用してください。

まずはrequirements.txtを用意し以下の内容で保存をします。

aws-iatk==0.1.0
boto3==1.29.3
boto3-stubs[events]==1.29.3
pytest==7.4.3

テストランナーとしてはunittestとpytestのどちらも利用出来ますが、この記事ではpytestを利用します。

保存が出来たら以下のコマンドでインストールを行います。

pip install -r requirements.txt

aws-iatkライブラリにおける認証情報の扱いなどはドキュメントに詳しく載っているので参照ください。

https://awslabs.github.io/aws-iatk/

簡単なテストから書いてみる

まずは簡単なテストから書いてみます。

このテストではCloudFormationのOutputから値を取得し、正しい値が取得出来ているかを確認しています。

from typing import Dict

import pytest
from aws_iatk import AwsIatk

stack_name = "AwsIatkTestStack"
region = "ap-northeast-1"

iatk = AwsIatk(region=region)


@pytest.fixture(scope="module")
def stack_output() -> Dict[str, str]:
    outputs = iatk.get_stack_outputs(
        stack_name=stack_name,
        output_names=["EventBusName", "EventRuleName"],
    ).outputs

    return outputs


def test_stack_output(stack_output: Dict[str, str]) -> None:
    assert stack_output["EventBusName"] == "イベントバス名"

get_stack_outputsメソッドを利用することでCloudFormationのOutputから値を取得し、Dictionary型に変換して返してくれます。

https://awslabs.github.io/aws-iatk/api/python/aws_iatk.html#AwsIatk.get_stack_outputs

このコードの通りあくまでテストコードの実行はpytestで行うため、pytestを書いたことがある方であればすんなり書き始められると思います。

テストコードの追加

先ほどのテストコードを修正して、以下のように書き換えます。

このテストコードでは最初に書いた通り、カスタムイベントバスに対してイベントを送信し正しくイベントを送信出来ているか検証します。

import json
from typing import Dict, Generator, TypedDict

import boto3
import pytest
from aws_iatk import AwsIatk


class Input(TypedDict):
    listener_id: str
    event_bus_name: str
    event_rule_name: str


stack_name = "AwsIatkTestStack"
region = "ap-northeast-1"

iatk = AwsIatk(region=region)
boto3_client = boto3.client("events", region_name=region)


@pytest.fixture(scope="module")
def stack_output() -> Dict[str, str]:
    # CloudFormationのOutputから値を取得
    outputs = iatk.get_stack_outputs(
        stack_name=stack_name,
        output_names=["EventBusName", "EventRuleName"],
    ).outputs

    return outputs


@pytest.fixture(scope="function", autouse=True)
def setup_teardown(stack_output: Dict[str, str]) -> Generator[Input, None, None]:
    # Before test
    listener = iatk.add_listener(
        event_bus_name=stack_output["EventBusName"],
        rule_name=stack_output["EventRuleName"].split("|")[1],
    )
    yield {
        "listener_id": listener.id,
        "event_bus_name": stack_output["EventBusName"],
        # 'EventBus|EventRule'という形式で出力されるのでsplitする
        "event_rule_name": stack_output["EventRuleName"].split("|")[1],
    }

    # After test
    iatk.remove_listeners(ids=[listener.id])


def test_event_polling(setup_teardown: Input) -> None:
    # EventBridgeにイベントを送信
    put_events_res = boto3_client.put_events(
        Entries=[
            {
                "EventBusName": setup_teardown["event_bus_name"],
                "DetailType": "Test",
                "Source": "com.test",
                "Detail": "{}",
            },
        ],
    )
    assert put_events_res["FailedEntryCount"] == 0

    # 送信したイベントを取得
    received_events = iatk.poll_events(
        listener_id=setup_teardown["listener_id"],
        wait_time_seconds=10,
        max_number_of_messages=1,
    ).events

    assert len(received_events) == 1

    event = json.loads(received_events[0])
    assert event["detail-type"] == "Test"
    assert event["source"] == "com.test"

以下で関数ごとに説明を行います。

@pytest.fixture(scope="function", autouse=True)
def setup_teardown(stack_output: Dict[str, str]) -> Generator[Input, None, None]:
    # Before test
    listener = iatk.add_listener(
        event_bus_name=stack_output["EventBusName"],
        rule_name=stack_output["EventRuleName"].split("|")[1],
    )
    yield {
        "listener_id": listener.id,
        "event_bus_name": stack_output["EventBusName"],
        # 'EventBus|EventRule'という形式で出力されるのでsplitする
        "event_rule_name": stack_output["EventRuleName"].split("|")[1],
    }

    # After test
    iatk.remove_listeners(ids=[listener.id])

まずこちらの関数ではpytestのfixture機能を利用してテストの前後で処理を実行しています。

add_listenerメソッドを呼び出すとテストに必要なAWSリソースが作成されます。

このメソッドではテスト用のイベントルールとイベントの送信先であるSQSが作成されています。

テスト用に作成されるイベントルールのイベントパターンはadd_listenerメソッドに渡したイベントルールと同じものになります。

リソース同士の関係を図に表すと以下のようになります。

イベントを取得する際は、こちらのSQSからメッセージの取得を行っているようです。

テスト終了時にはremove_listenersメソッドを呼び出しこれらのリソースの削除を行います。

def test_event_polling(setup_teardown: Input) -> None:
    # EventBridgeにイベントを送信
    put_events_res = boto3_client.put_events(
        Entries=[
            {
                "EventBusName": setup_teardown["event_bus_name"],
                "DetailType": "Test",
                "Source": "com.test",
                "Detail": "{}",
            },
        ],
    )
    assert put_events_res["FailedEntryCount"] == 0

    # 送信したイベントを取得
    received_events = iatk.poll_events(
        listener_id=setup_teardown["listener_id"],
        wait_time_seconds=10,
        max_number_of_messages=1,
    ).events

    assert len(received_events) == 1

    event = json.loads(received_events[0])
    assert event["detail-type"] == "Test"
    assert event["source"] == "com.test"

こちらの関数はテストケースを表しています。

まずはboto3を利用してカスタムイベントバスにイベントを送信します（Arrange）。

次にpoll_eventsメソッドを利用することで送信したイベントが対象のイベントルールで正しく処理できているかを確認します。

このメソッドは先ほど作成したSQSからメッセージを取得し返却します。

他にはwait_until_event_matchedという取得とAssertionを行うメソッドも用意されています。

https://awslabs.github.io/aws-iatk/api/python/aws_iatk.html#AwsIatk.wait_until_event_matched

最後のAssertionでは取得したイベントに対して期待したイベントかどうかを検証します。

ここではメッセージ数やメッセージ内容の簡単な検証を行います。