Amazon Web Services ブログ

AWS 統合アプリケーションテストキット(IATK)のご紹介

この投稿は、プリンシパルスペシャリストソリューションアーキテクトの Dan Fox とシニアソリューションアーキテクトの Brian Krygsman によって執筆されました。

本日、AWS は AWS 統合アプリケーションテストキット(IATK) のパブリックプレビューを発表しました。AWS IATK は、クラウドベースのアプリケーションの自動テストを書くのに役立つソフトウェアライブラリです。このブログ記事では、AWS IATKのいくつかの初期機能を紹介し、ビデオ処理アプリケーションの例を使用して、これがどのように動作するのかを解説します。サーバーレスのテストをはじめるには、serverlessland.com/testing で詳細をご覧ください。

概要

AWS Lambda、Amazon EventBridge、AWS Step Functions などのサーバーレスサービスで構成されるアプリケーションを作成すると、アーキテクチャコンポーネントの多くはデスクトップにデプロイできず、AWS クラウドにのみ存在します。ローカルに展開されたアプリケーションでの作業とは対照的に、これらのタイプのアプリケーションは、自動テストを実行するためのクラウドベースの戦略の恩恵を受けます。今回のパブリックプレビューのリリースでは AWS IATK は Python アプリケーションにおける実装を支援します。AWS IATK は、今後のリリースで他の言語をサポートします。

テスト用のリソースを見つける

クラウドリソースの自動テストを作成するには、リソースの物理IDが必要となります。物理IDは、作成後にAWSがリソースに割り当てる名前です。たとえば、Amazon API Gateway にリクエストを送信するには、API エンドポイントを形成する物理IDが必要です。

別々の Infrastructure as Code スタックにクラウドリソースをデプロイすると、物理IDを見つけるのが難しい場合があります。CloudFormation の場合、テンプレート内のリソースの論理IDとスタック名を作成します。IATKを使用することで、論理IDとスタック名を入力すると、リソースの物理IDが取得できます。スタック名を入力することで、スタックの出力を取得することもできます。これらの便利な方法により、記述したテストのためにリソースを見つけやすくなります。

イベント駆動型アーキテクチャのテストハーネスの作成

イベント駆動型アーキテクチャの統合テストを書くには、アプリケーションをサブシステムに分割して論理的な境界を確立します。サブシステムは、判断するのに十分シンプルで、理解可能な入力と出力が含まれている必要があります。サブシステムをテストするための有用なテクニックの 1 つは、テストハーネスを作成することです。テストハーネスは、サブシステムをテストするために特別に作成するリソースです。

たとえば、統合テストは、入力テストイベントを渡すことでサブシステムプロセスを開始できます。IATK は、出力されるイベントのために Amazon EventBridge をリッスンするテストハーネスを作成します。（内部では、ハーネスは出力イベントを Amazon Simple Queue Service に転送する EventBridge ルールで構成されています。）その後、統合テストはテストハーネスにクエリして出力を調べ、テストが成功または失敗するかどうかを判断します。これらのハーネスは、クラウドにおけるイベント駆動型アーキテクチャの統合テストを作成するのに役立ちます。

非同期機能をテストするためのサービスレベルアグリーメントを確立する

同期的なサービスを作成すると、自動テストがリクエストを行い、即時にレスポンスされることを期待します。アーキテクチャが非同期の場合、サービスはリクエストを受け取り、その後で一連のアクションを実行します。決められた期間がない場合、アクティビティの成功をどのようにテストしたら良いでしょうか？

非同期システムのための合理的なタイムアウトを作成することを検討してください。サービスレベルアグリーメント (SLA) としてタイムアウトをドキュメント化します。SLA は外部に公開するか、内部標準としてドキュメント化するかを決定できます。IATK には、タイムアウトを確立できるポーリング機能が含まれています。この機能は、非同期システムがタイムリーにタスクを完了するかどうかをテストするのに役立ちます。

詳細なテストに AWS X-Ray を使用する

アプリケーションの内部の詳細をより可視化したい場合は、AWS X-Rayで計測します。AWS X-Rayを使用すると、複数のサービスを通じてイベントのパスを追跡できます。IATKは、AWS X-Ray のサンプリングレートを設定し、トレースツリーを取得し、トレース期間をアサートするのに役立つ便利な機能を提供します。これらの機能は、分散システムをより詳細に観察し、テストするのに役立ちます。

より詳しくは、aws-samples/serverless-test-samples で非同期アーキテクチャのテストをご覧ください。

サンプルアプリケーションの概要

IATKの機能を示すために、この投稿ではプラグインアーキテクチャで設計されたサーバーレスビデオアプリケーションの一部を使用します。コア開発チームが主要なアプリケーションを作成します。組織全体の分散された開発チームはプラグインを作成します。1つの AWS CloudFormation スタックがプライマリアプリケーションをデプロイします。別々のスタックは各プラグインをデプロイします。

プライマリアプリケーションとプラグイン間の通信は、EventBridge バスによって管理されます。プラグインは、アプリケーションのライフサイクルイベントをバスからプルし、20秒以内に完了通知イベントをバスに戻す必要があります。テストのために、コアチームは、適切にフォーマットされたサンプルライフサイクルイベントを発行することで、実稼働プロセスを模倣する AWS Step Functions ワークフローを作成しました。開発者は、開発およびテスト環境でこのテストワークフローを実行し、プラグインがイベントバスと適切に通信していることを確認します。

次のデモンストレーションは、プラグインの動作を検証するサンプルアプリケーションの統合テストを示しています。統合テストでは、IATK は Step Functions のワークフローを特定し、プラグインから送信されるイベント完了通知をリッスンするためのテストハーネスを作成します。その後、テストはワークフローを実行してライフサイクルプロセスを開始し、プラグインアクションを開始します。その後、IATK はタイムアウト付きのポーリングメカニズムを使用して、プラグインが 20 秒のサービスレベルアグリーメントに準拠していることを確認します。以下は処理のシーケンスです。

Sequence of processing

  1. 統合テストは、テストワークフローの実行を開始します。
  2. ワークフローは、ライフサイクルイベントをバスに PUT します。
  3. プラグインは、バスからライフサイクルイベントを PULL します。
  4. プラグインが完了すると、バスに完了イベントが PUT されます。
  5. 統合テストは、完了イベントをポーリングして、テストが SLA 内で合格するかどうかを判断します。

サンプルアプリケーションのデプロイとテスト

以下の手順に従って、このアプリケーションを確認し、ローカルに構築し、AWSアカウントにデプロイし、テストしてください。

サンプルアプリケーションのダウンロード

  1. ターミナルを開き、次のコマンドで GitHub からサンプルアプリケーションをクローンするか、コードをダウンロードします。このリポジトリには、サーバーレスアプリケーションをテストするための他のサンプルパターンも含まれています。 
    git clone https://github.com/aws-samples/serverless-test-samples
    Bash
  2. IATKサンプルアプリケーションのルートは python-test-samples/integrated-application-test-kit にあります。このディレクトリに移動します。 
    cd serverless-test-samples/python-test-samples/integrated-application-test-kit
    Bash

統合テストのレビュー

アプリケーションをデプロイする前に、テキストエディタで plugins/2-postvalidate-plugins/python-minimal-plugin/tests/integration/test_by_polling.py を開いて、統合テストが IATK をどのように使用するかを確認してください。テストクラスは、ファイルの上部にある IATK をインスタンス化します。

iatk_client = aws_iatk.AwsIatk(region=aws_region)
Python

setUp() メソッドでは、テストクラスは IATK を使用して CloudFormation スタック出力を取得します。これらの出力は、プラグインテスター AWS Step Functions ワークフローのようなデプロイされたクラウドコンポーネントの参照です。

stack_outputs = self.iatk_client.get_stack_outputs(
    stack_name=self.plugin_tester_stack_name,
    output_names=[
        "PluginLifecycleWorkflow",
        "PluginSuccessEventRuleName"
    ],
)
Python

テストクラスは、スタック出力で提供されるイベントルールを使用して、リスナーをデフォルトのイベントバスにアタッチします。このテストでは、後でこのリスナーを使用してイベントをポーリングします。

add_listener_output = self.iatk_client.add_listener(
    event_bus_name="default",
    rule_name=self.existing_rule_name
)
Python

テストクラスは、tearDown() メソッドでリスナーをクリーンアップします。

self.iatk_client.remove_listeners(
    ids=[self.listener_id]
)
Python

設定が完了すると、test_minimal_plugin_event_published_polling() メソッドによって実際のテストが実装されます。

テストは最初にトリガーイベントを初期化します。

trigger_event = {
    "eventHook": "postValidate",
    "pluginTitle": "PythonMinimalPlugin"
}
Python

次に、テストはプラグインテスター Step Functions ワークフローの実行を開始します。setUp の実行中にフェッチされた plugin_tester_arn が使用されます。

self.step_functions_client.start_execution(
    stateMachineArn=self.plugin_tester_arn,
    input=json.dumps(trigger_event)
)
Python

テストはリスナーをポーリングし、プラグインがイベントを発火するのを待ちます。SLA で決められたタイムアウトに達するか、最大数のメッセージを受信すると、ポーリングを停止します。

poll_output = self.iatk_client.poll_events(
    listener_id=self.listener_id,
    wait_time_seconds=self.SLA_TIMEOUT_SECONDS,
    max_number_of_messages=1,
)
Python

最後に、テストは適切な数のイベントを受け取り、それらが適切にフォーマットされているか検証します。

self.assertEqual(len(poll_output.events), 1)
self.assertEqual(received_event["source"], "video.plugin.PythonMinimalPlugin")
self.assertEqual(received_event["detail-type"], "plugin-complete")
Python

前提条件のインストール

このサンプルをビルドするには、次の前提条件が必要です。

サンプルアプリケーションコンポーネントのビルドとデプロイ

  1. AWS SAMを使用して、プラグインテスターを構築し、AWSアカウントにデプロイします。プラグインテスターは、前の図に示されている Step Functions ワークフローです。ビルドプロセス中に、ビルドコマンドに --use-container フラグを追加して、AWS SAM に渡されたコンテナでアプリケーションを作成するように指示できます。デプロイプロセス中にデフォルト値を受け入れるか、上書きすることができます。後で「スタック名」と「AWSリージョン」を使用して統合テストを実行します。 
    cd plugins/plugin_tester # plugin tester のディレクトリに移動します

sam build --use-container # plugin tester をビルドします
    Bash

    sam build

  2. テスターをデプロイする:

    sam deploy --guided # plugin tester をデプロイします
    Bash

  3. プラグインテスターがデプロイされたら、AWS SAMを使用してプラグインをデプロイします。 
    cd ../2-postvalidate-plugins/python-minimal-plugin # plugin directory に移動します

sam build --use-container # plugin をビルドします
    Bash

  4. プラグインをデプロイする: 
    sam deploy --guided # plugin をデプロイします
    Bash

テストを実行する

unittest や pytest などの標準的な Python テストランナーを使用して、IATK で書かれたテストを実行できます。サンプルアプリケーションのテストでは、ユニットテストを使用します。

  1. 仮想環境を使用して依存関係を整理します。サンプルアプリケーションのルートから、実行します。 
    python3 -m venv .venv # virtual environment を作成する

source .venv/bin/activate # virtual environment を activate する
    Bash
  2. IATK を含む依存関係をインストールします。 
    cd tests
pip3 install -r requirements.txt
    Bash
  3. テストを実行し、以前のデプロイから必要な環境変数を提供します。plugin_tester ディレクトリの samconfig.toml ファイルでその値を見つけることができます。

    cd integration

PLUGIN_TESTER_STACK_NAME=video-plugin-tester
AWS_REGION=us-west-2
python3 -m unittest ./test_by_polling.py
    Bash

ユニットテストがテストを実行すると、出力が表示されるはずです。 AWS アカウントで Step Functions コンソールを開き、PluginLifecycleWorkflow-<random value> ワークフローを選択して、プラグインテスターが正常に実行されたことを確認します。最近の実行は、成功したステータスを示しています。

他のIATK機能を確認する

サンプルアプリケーションには、モックイベントの生成やAWS X-Rayトレースの取得など、他のIATK機能の例が含まれています。

クリーンアップ

AWS SAM を使用して、AWS アカウントからプラグインとプラグインテスターの両方のリソースをクリーンアップします。

  1. プラグインのリソースを削除する: 
    cd ../.. # plugin directory に移動する
sam delete # plugin を削除する
    Bash

  2. プラグインテスターのリソースを削除します。 
    cd ../../plugin_tester # plugin tester directory に移動する 
		sam delete # plugin tester を削除する
    Bash

IATK がテスト中に作成した一時的なテストハーネスリソースは、tearDown メソッドが実行されるとクリーンアップされます。実行中に問題が発生した場合、一部のリソースは削除されない可能性があります。IATK は、作成したすべてのリソースにタグを追加します。これらのタグを使用してリソースを見つけ、手動で削除できます。また、独自のタグを追加することもできます。

まとめ

AWS 統合アプリケーションテストキット（IATK）は、クラウドアプリケーションの自動テストを書くのに役立つ便利なソフトウェアライブラリです。このブログ記事は、IATK の初期Python バージョンの機能のいくつかを示しています。 サーバーレスアプリケーションの自動テストの詳細については、serverlessland.com/testing をご覧ください。また、serverlessland.com/testing/patterns または GitHub の AWS serverless-test-samples リポジトリでコード例を表示することもできます。

より多くのサーバーレス学習リソースについては、Serverless Landをご覧ください。

翻訳はソリューションアーキテクトの淡路が担当しました。原文はこちらです。