AWS Integrated Application Test Kit (IATK) を使ってみよう #AWS

みんな、テストやってる？

えっ、まさかサーバーレスだからってテストサボってないですよね？？？
イベント駆動で非同期だからってえいやでデプロイしてないですよね？？？
それ t_wxxx の前でも同じこと言えn(以下略

皆さん、サーバーレスでもテストは重要です。
非同期だろうがイベント駆動だろうが、テストから逃げることは許されません。
逃げちゃダメだ。

ただ、一言で「テスト」と言ってもその手法やスコープ、目的は様々です。
まずは本題に入る前に少し「テスト」そのものに関する考え方をおさらいしておきましょう。

何をどうテストするか？

以下のような図を見たことがある人は多いと思います。
そうです、いわゆるテストピラミッドというやつです

テストには、ユニットテスト(単体テスト) や 統合テスト(結合テスト)、E2Eテスト など様々な種類があるわけですが、闇雲に全てテストしまくればよいというわけではありません。当然テストにはコストがかかるからです。

開発者の生産性を妨げず、かつソフトウェアの品質を最大化するための最適なテスト戦略を示してくれているのがこの図というわけなのですが、要は実施が比較的簡単な下層のテストを多くしてカバレッジを広げましょうというやり方が、一般的には良しとされています。

他のコンポーネントに依存せず、独立して最小限の単位で意味のあるロジックをテストできるユニットテストを入念にやっておけば、結合テストで確認する点も少なくて済みますし、依存性が少ない分テストも安定します。

また、単体テストのやり方はテスト対象のシステムによって大きく変わることはありません。オンプレミスでもクラウドでも、サーバーレスでも非同期でも、最小限の単なる入力と出力にフォーカスしたビジネスロジックのテストでは、やるべきことは同じです。

また、一部他のAWSサービスなどの外部リソースの呼び出しがあったとしても、モックなどを活用して簡単にテストすることができます。

Lambda を使っていることは、単体テストをやらなくてよい免罪符にはなりません。

クラウドにおけるテストの課題

では結合テストはどうでしょうか？

クラウドの隆盛により、結合テストを取り巻く昨今の状況はやや複雑化しています。

複数のマイクロサービスが協調して動作する分散システム
キューやイベントバスを介した非同期呼び出し
マネージドサービスやサードパーティ SaaS の活用

いつ終わるか分からない非同期処理の結果や、EventBridge が Lambda 関数を適切に呼び出せたことをどのように確認すれば良いのでしょうか？確かに、このような要素で構成されたシステムをこれまでと同じやり方でテストするのは容易ではないかもしれません。

開発者は、新たな知識や技術を導入し、多くの工数をかけて複雑なテストケースを記述する必要があるでしょう。また、たとえテストが書けたとしても、そのテストは不安定なもので、無駄なクラウドリソースのコストを発生させる可能性があります。

AWS IATK の出番だ！

「あなたの不安定なテストはどこから？」

「私は AWS のサーバーレスサービスを使ったイベント駆動アプリケーションから」

それなら AWS Integrated Application Test Kit (AWS IATK)！

はい、ようやく本題です。

AWS IATK は、先ほどの諸々の課題を解決し、クラウドアプリケーションのテストを行いやすくする世界観を目指すオープンソースのテストライブラリです (と私は解釈してます)。

以下にリリース文を引用します (筆者訳)。

AWSは、AWS Integrated Application Test Kit (IATK) のプレビューを発表しました。この新しいオープンソースのテストライブラリは、開発者がクラウドアプリケーションのテストを、より高速かつ正確に作成することを容易にします。AWS IATK を利用することで、開発者はコードと AWS インテグレーションをクラウド上の環境に対して実行するテストを迅速に作成することができ、開発プロセスの早い段階でミスを発見しやすくなります。IATK には、テストイベントを生成するユーティリティ、Amazon EventBridge イベントバスのイベント配信とスキーマを検証するユーティリティ、AWS X-Ray トレースを使用して呼び出しを検証するアサーションが含まれています。

以前は、イベント駆動アプリケーションに対して効果的なテストを記述するために、開発者はテスト対象のアプリケーションからリソース ID を抽出するカスタムロジックを記述し、テストハーネスとして追加の AWS リソースを作成して管理する必要がありました。テスト設計は非同期操作を考慮しなければならず、その結果、テストの実行時間が長くなったり、テストの整合性が崩れることがありました。この複雑さが、多くの開発者にとってテストを利用しにくいものにしていました。IATK を使用すると、開発者はライブラリユーティリティを活用してこの重労働を行い、ビジネスロジックに集中することができます。IATK には、クラウドアプリケーションを呼び出すテストイベントの生成、AWS から必要な情報の取得、テストに必要なテストリソースを管理するユーティリティが含まれています。

Source: AWS Integrated Application Test Kit (preview)

ユーティリティライブラリなので、使ってみないとイメージが湧きにくいかもしれません。どんなことができるか、サンプルをベースに中身を見ていきましょう。

サンプルアプリケーション

AWS IATK はリリース時点で Python をサポートしています。Python で記述したテストを unittest や pytest で動かす前提になります。他の言語のサポートも今後予定されてますので、追加情報をお待ちください。

リリース時に、サンプルアプリケーションが GitHub で公開されています。中身や動かし方についてはブログで詳しく紹介されているので、そちらを読んでみるのもよいかもしれません。

これをそのまま使ってもいいのですが、今回は理解しやすくするために別のシンプルなアプリケーションを用意しました。コードは GitHub に置いておきました。

マイクロサービスベースで構成される EC サイトのアプリケーションで、注文サービス (Order Function) が注文レコードを作成してイベントを送信し、それを非同期で出荷サービス (Shipping Function) が処理するという流れになっています。詳しくは以下のアーキテクチャ図をご覧ください。

テスト対象

さて、このアーキテクチャにおいて統合テスト (結合テスト) で確認すべきことはなんでしょうか？

統合テストで注目すべきところは、サービスの境界や他のコンポーネントとの接続部分です。

今回は、たった2つですがマイクロサービスで構成されるアーキテクチャを想定して作っており、サービス間はイベントを介して互いに通知を行います。

なので、OrderFunction の振る舞いが、ShippingFunction をトリガーするフローに着目するのが良さそうです。

これはモノリシックで同期的な呼び出しチェーンであれば簡単にテストできますが、今回のような非同期処理では少しコツが必要です。

テストハーネス

ここで言う「テストハーネス」とは、AWS IATK が作成する一連の AWS リソースで、具体的には EventBridge ルールや、そのターゲットとなる SQS キューを指します。これらのリソースはテスト開始時に一時的に作成され、テスト終了時に削除されます。

LocalStack のようなエミュレーターやモックを使ってテストを行っている人も多いと思いますが、実際の動作に即したテストをやろうとすると、やはり AWS リソースそのものを実際に作成してテストするのが確実です。

AWS IATK を利用した統合テストでは、本番相当のテスト環境を事前にプロビジョニングしておき、そこにテスト時にさらに一時的なテストハーネスを作成して、テストに必要な環境を揃えます。これらの一時的なリソースのライフサイクル管理は、自前でやろうとすると本来手間となる部分ですが、そこは AWS IATK を使うことで楽ができます。

Arrange-Act-Assert パターン

Arrange-Act-Assert とは、テストケースを構成するステップを表した1つの方法論です。Arrange でテストに必要な環境をセットアップし、Act でテスト対象の振る舞いを実行させ、Assert でアウトプットを検証します。

AWS IATK では、これらのステップをサポートするヘルパー関数が提供されています。

テストケース

それでは Arrange-Act-Assert パターンに沿って、AWS IATK を利用したテストを書いてみましょう。

1. order_created イベントの受信

まずは、OrderFunction の処理完了後にイベントバスに期待するイベントが送信されているかをテストしてみます。このテストでは、そのイベントにマッチする EventBridge ルールが存在するかどうかも確認することができます。

Arrange

最初にテストハーネスを作成します。Python の unittest を使ったことがある方にはお馴染みだと思いますが、setUp() の中で定義していきます。

（※以降、重要なところのみコードを抜粋してあるので適宜補って読んでください）

test_event_bus.py

from unittest import TestCase
import aws_iatk

class TestEventBus(TestCase):
    aws_region = os.environ.get("AWS_REGION")
    if(aws_region is None):
        raise Exception("AWS_REGION environment variable is required")
    
    iatk_client = aws_iatk.AwsIatk(region=aws_region)
    
    listener_id = None

    def setUp(self):
        listener = self.iatk_client.add_listener(
            event_bus_name="Event bus name to be tested here",
            rule_name="EventBridge rule name to be tested here"
        )
        self.listener_id = listener.id

IATK をインポートして使えるようにし、リスナーを作成するだけです。ここでは、リスナー＝テストハーネスです。テスト対象となる EventBridge ルールをコピーして新しくルールを作成し、ターゲットとして SQS も作成してメッセージをリッスンします。これらはテスト終了後に削除します。

指定するリソース名ですが、上記のようにコード内にベタ書きするのはイケてない感じがします。これらはテスト環境をプロビジョニングした際の IaC の出力から拾いたいです。

IATK にはまさにそれ用の関数 get_physical_id_from_stack(), get_stack_outputs() が用意されているので、それを使って書き換えてみます。

test_event_bus.py

    listener_id = None
    stack_outputs = None

    def setUp(self):
        self.stack_outputs = self.iatk_client.get_stack_outputs(
            stack_name=os.environ.get("TEST_STACK_NAME"),
            output_names=["OrderApi","PrepareShippingEventRule"]
        ).outputs

        listener = self.iatk_client.add_listener(
            event_bus_name="default",
            rule_name=self.stack_outputs["PrepareShippingEventRule"]
        )
        self.listener_id = listener.id

あらかじめ CloudFormation を使って作成してあるテスト環境のスタック名を TEST_STACK_NAME で渡します。また、テンプレート内で API のエンドポイントや EventBridge ルール名を Outputs として出力してあるのでそれを取得しています。

リスナー ID を保持しているのは、終了時にリソースを削除するためです。これは、tearDown() の中で行います。

test_event_bus.py

def tearDown(self):
        self.iatk_client.remove_listeners(
            ids=[self.listener_id],
        )

Act

では作成したテストハーネスを介して動作を検証してみましょう。
今回やることは、API Gateway のエンドポイントに注文作成の POST リクエストを送るだけです。

test_event_bus.py

def test_order_created_event_published(self):        
        expected_event = {
            "source": "iatk_demo.order_service",
            "detail-type": "order_created"
        }
        sample_order = { 
            "order_id": str(uuid.uuid4()),
            "amount": 100 
        }
     
        response = requests.post(self.stack_outputs["OrderApi"], json=sample_order)
        self.assertEqual(response.status_code, requests.codes.ok)

API エンドポイントは前述の通りスタックの Outputs から取得してきています。
サンプルの適当な注文データを渡しているだけです。

Assert

最後に結果の検証です。
期待する動作としては、OrderFunction が適切にリクエストを処理し、order_created イベントをイベントバスに送信して、イベントバス経由で ShippingFunction がトリガーされることです。　　

前述の test_order_created_event_published() の続きです。

test_event_bus.py

        poll_outputs = self.iatk_client.poll_events(
            listener_id=self.listener_id,
            wait_time_seconds=20,
            max_number_of_messages=1,
        )
        
        self.assertEqual(len(poll_outputs.events), 1)
        
        actual_event = json.loads(poll_outputs.events[0])
        self.assertEqual(actual_event["source"], expected_event["source"])
        self.assertEqual(actual_event["detail-type"], expected_event["detail-type"])
        self.assertEqual(actual_event["detail"]["order_id"], sample_order["order_id"])
        self.assertEqual(actual_event["detail"]["order_status"], "created")

aws_iatk.AwsIatk#poll_events() にリスナー ID を渡すと、このリスナーがイベントをキャプチャするまで待機します。これで、EventBridge イベントバスにイベントが送信され、ターゲットに受け渡すまでの振る舞いを確かめることができます。

テストハーネスは前述の通り EventBridge ルールと SQS キューで構成されており、裏側では SQS からメッセージを定期的にポーリングするような動きになっているようです。

メッセージが観測されたら、ペイロードを確認するアサーションも入れています。

% python -m unittest tests/integration/test_event_bus.py
.
----------------------------------------------------------------------
Ran 1 test in 7.696s

OK

これで最初のテストは完了です！私の環境では7~10秒程度で終わりました。AWS リソースの作成処理が入っている割には早いなと思いました。

基本的な流れは掴めたかと思うので、他のヘルパーやユースケースについても簡単に見ておきましょう。

2. モックイベントの作成

先ほどはイベントのデータを適当にテストケースの中で当てはめてましたが、これだと今後システムが変化するにつれて実際のスキーマと齟齬が発生する可能性があります。どこかで聞いたことがある話ですね。

幸い、EventBridge にはスキーマという機能があり、イベントの構造を OpenAPI Specification または JSONSchema Draft4 specification で定義しておくことができます。

IATK は、このスキーマレジストリに登録されているスキーマを元に、モックイベントを生成する関数を持っています。

test_create_order_with_mock_event.py

def test_create_order_with_mock_event(self):        

        def set_mock_event_properties(event):
            event["order_id"] = str(uuid.uuid4())
            event["amount"] = 100
            return event
        
        mock_event = self.iatk_client.generate_mock_event(
            registry_name=self.stack_outputs["CustomEventRegistry"],
            schema_name=self.stack_outputs["CreateOrderEventSchema"],
            contexts=[set_mock_event_properties],
        ).event

スキーマの定義はこちら

template.yaml

CustomEventRegistry:
    Type: AWS::EventSchemas::Registry
    Properties:
      Description: Event schema registry for this demo
  CreateOrderEventSchema:
    Type: AWS::EventSchemas::Schema
    Properties:
      RegistryName: !GetAtt CustomEventRegistry.RegistryName
      SchemaName: CreateOrderEventSchema
      Description: 'Event used to trigger shipping function'
      Type: JSONSchemaDraft4
      Content: >
        {
          "$schema": "http://json-schema.org/draft-04/schema#",
          "type": "object",
          "properties": {
            "order_id": {
              "type": "string"
            },
            "amount": {
              "type": "integer"
            },
            "order_status": {
              "type": "string",
              "default": "N/A",
              "enum": [
                "N/A",
                "created",
                "shipped",
                "delivered"
              ]
            }
          },
          "required": [
            "order_id",
            "amount",
            "order_status"
          ]
        }

Arrange でイベントを用意する必要がある際に活用できる機能かと思います。

3. トレースの追跡

最後に、複数サービス/コンポーネントをまたがるリクエスト全体を追跡したテストです。AWS X-Ray との連携でこんなこともできてしまいます。

今回の構成では、リクエストは以下の経路を辿ります。

Amazon API Gateway
-> 2. AWS Lambda (OrderFunction)
-> 3. Amazon DynamoDB
-> 4. Amazon EventBridge
-> 5. AWS Lambda (ShippingFunction)
-> 6. Amazon DynamoDB

前提としてトレースを取得している必要があるので、各サービスでの X-Ray の有効化と、ライブラリへのパッチ適用を行います。

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Globals:
  Function:
    Timeout: 3
    Tracing: Active
  Api:
    TracingEnabled: true

order/app.py

import json
import boto3
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

patch_all()

eventbridge = boto3.client('events')
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):
...

これでリクエストを流すと、コンソールから以下のキャプチャのようなサービスマップが見られます。

トレースをテストするには、get_trace_tree() または retry_get_trace_tree_until() を使用します。どちらでもテストは可能そうですが、タイムアウトやリトライを自分でケアする必要がないという点では後者を使う方が良さそうです。

（※記事執筆時点ではどちらもの動作もやや不安定だったので後で修正するかもです）

setUp(), tearDown() あたりはあまり変わらないので省略します。

test_trace.py

class TestTrace(TestCase):
    def test_order_trace(self):        
        sample_order = { 
            "order_id": str(uuid.uuid4()),
            "amount": 100 
        }
        expectedTraceTree = [
            [
                { "origin": "AWS::ApiGateway::Stage", "name": "iatk-demo/Proda" },
                { "origin": "AWS::Lambda", "name": self.stack_outputs["OrderFunction"] },
                { "origin": "AWS::Lambda::Function", "name": self.stack_outputs["OrderFunction"] },
                { "origin": "AWS::DynamoDB::Table", "name": "DynamoDB" },
            ],
            [
                { "origin": "AWS::ApiGateway::Stage", "name": "iatk-demo/Prod" },
                { "origin": "AWS::Lambda", "name": self.stack_outputs["OrderFunction"] },
                { "origin": "AWS::Lambda::Function", "name": self.stack_outputs["OrderFunction"] },
                { "origin": "AWS::Events", "name": "Events" },
                { "origin": "AWS::Lambda", "name": self.stack_outputs["ShippingFunction"] },
                { "origin": "AWS::Lambda::Function", "name": self.stack_outputs["ShippingFunction"] },
                { "origin": "AWS::DynamoDB::Table", "name": "DynamoDB" },
            ]
        ]
 
        response = requests.post(self.stack_outputs["OrderApi"], json=sample_order)
        self.assertEqual(response.status_code, requests.codes.ok)
        trace_id = response.headers['X-Amzn-Trace-Id']
        
        self.iatk_client.poll_events(
            listener_id=self.listener_id,
            wait_time_seconds=20,
            max_number_of_messages=1,
        )

        def assertion(output):
            tree = output.trace_tree
            self.assertEqual(len(tree.paths), 2)
            for i, path in enumerate(tree.paths):
                for j, seg in enumerate(path):
                    self.assertIsNone(seg.error)
                    self.assertIsNone(seg.fault)
                    self.assertEqual(seg.origin, expectedTraceTree[i][j]["origin"])
                    self.assertEqual(seg.name, expectedTraceTree[i][j]["name"])
        
        self.assertTrue(self.iatk_client.retry_get_trace_tree_until(
            tracing_header=trace_id,
            assertion_fn=assertion,
            timeout_seconds=20,
        ))

API Gateway のレスポンスからトレースヘッダーを取り出して、トレースツリーの各セグメントを検証します。
あまり細かく見る必要はないので、origin と name だけアサートする例を書いてみました。

成功する時もありますが、不規則なタイムアウトとよく分からない例外がランダムで発生しているので、このコードは引き続き検証します。X-Ray 難しい。。。Issue も立っているので、ライブラリ側で修正がかかるかもしれません。まあまだプレビューですからね。

その他の注意

ここまであまり明確に書いてこなかったですが、テストハーネスを作ったりリスナーを動かすために AWS IAM のパーミッションが必要です。

例えば add_listener() であれば、ドキュメントによると以下のパーミッションが必要になるので、テストを実施する環境にアタッチするように気をつけてください。

events:DescribeEventBus
events:DescribeRule
events:PutRule
events:PutTargets
events:DeleteRule
events:RemoveTargets
events:TagResource
sqs:CreateQueue
sqs:GetQueueAttributes
sqs:GetQueueUrl
sqs:DeleteQueue
sqs:TagQueue

まとめ

第一印象としては「もっと期待してたのに！」と思う方も多いかもしれません。
リリースを斜め読みすると「なんかすごそうなの出てきた...！」と思っちゃいますよね。

ただ私個人としてはすごく期待してます。ようやくテスト領域に踏み出してきたなと。
これまで手を出しづらかったクラウド環境の統合テストの領域に、AWS が公式のヘルパーツールを出してきたという状況は大きな意味を持つと思っています。これは、確実に開発者がテストを書く後押しになるはずです。

まだプレビューなので、ここからどんどんカバーするシナリオが増え、安定性も増していくのをみんなで見守りましょう。そして Contribute しましょう。

以上です