Amazon Web Services ブログ
あらゆる言語でのCDKアプリケーションのテスト
AWS Cloud Development Kit (AWS CDK)は、使い慣れたプログラミング言語でクラウドアプリケーションのリソースを定義するためのオープンソースソフトウェア開発フレームワークです。AWS CDKはインフラストラクチャをプログラミング言語で定義できるため、アプリケーションのコードと同様に、インフラストラクチャのコードに対しても自動化されたユニットテストを記述することができます。テストは効果的なDevOpsの実践のために重要な要素であり、インフラストラクチャのコードをテストすることで、AWSクラウドで期待通りのリソースを確実に作成することができ、リグレッションが発生するのを防ぐことができるなどのメリットがあります。
今日、私はAWS CDK用のassertions モジュールを発表できることを嬉しく思います。これはCloudFormationテンプレートを中心に、CDKアプリケーションに対してユニットテストを書くのに役立つように設計されたAPIのセットです。
すべての言語をサポート
以前のAWS Blogの記事では、JavaScriptとTypeScriptでのみ利用可能なassertモジュールを使ってテストを書く方法を説明しました。assertモジュールと同様に、新しいCDK assertionsモジュールは、CDKアプリによって生成(synthesize)されたCloudFormationテンプレートを正確に検証するためのAPIセットを提供します。さらに、新しいassertionsモジュールは、CDKがサポートするすべての言語に対応しています。
新しいassertionsモジュールはCDKがサポートするすべての言語をサポートしていますが、この記事のスニペットはPythonで書かれています。これらのサンプルの完全なソースコードはGitHubで公開されており、TypeScript、Java、およびPythonで書かれた同等のコードが含まれています。
きめ細かな(Fine-grained) アサーション
assertionsモジュールは、テンプレートのある部分が与えられたオブジェクトにマッチすることをアサートしたり、テンプレートのある部分を検索するためのツールをいくつか提供しています。これらのツールを使って、与えられたタイプとプロパティを持つリソースが存在することをアサートしたり、特定の出力が存在することをアサートしたり、テンプレートが与えられた数のリソースを持っていることをアサートしたりすることができます。
下記のコードのように、AWS Step Functions のステートマシンと AWS Lambda 関数を作成するスタック(ProcessorStack) があるとします。Lambda関数は Amazon SNS のトピックをサブスクライブしており、メッセージ受けるとステートマシンを起動します。
from typing import List
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_sns as sns
from aws_cdk import aws_sns_subscriptions as sns_subscriptions
from aws_cdk import aws_stepfunctions as sfn
from aws_cdk import core as cdk
class ProcessorStack(cdk.Stack):
def __init__(
self,
scope: cdk.Construct,
construct_id: str,
*,
topics: List[sns.Topic],
**kwargs
) -> None:
super().__init__(scope, construct_id, **kwargs)
# 将来的にはこのステートマシンが何かの仕事をすることになるでしょう...。
state_machine = sfn.StateMachine(
self, "StateMachine", definition=sfn.Pass(self, "StartState")
)
# このLambda関数がステートマシンを起動します。
func = lambda_.Function(
self,
"LambdaFunction",
runtime=lambda_.Runtime.NODEJS_14_X,
handler="handler",
code=lambda_.Code.from_asset("./start-state-machine"),
environment={
"STATE_MACHINE_ARN": state_machine.state_machine_arn,
},
)
state_machine.grant_start_execution(func)
subscription = sns_subscriptions.LambdaSubscription(func)
for topic in topics:
topic.add_subscription(subscription)このスタックをきめ細かなアサーションでテストするにはどうすればよいでしょうか。
まず ProcessorStack スタックを作り、CloudFormationのテンプレートを生成します。生成したテンプレートに対してアサーションを行うことができます。
さらに ProcessorStackは、クロススタックリファレンスで他のスタックのリソース (SNSトピックの配列)を参照しますが、このスタックもテストコードの中で作成することができます
from aws_cdk import aws_sns as sns
from aws_cdk import core as cdk
from aws_cdk.assertions import Template
from app.processor_stack import ProcessorStack
def test_synthesizes_properly():
app = cdk.App()
# ProcessorStackは別のスタックのリソースを参照するため、
# SNSトピックのためのスタックを作成しておきます。
topics_stack = cdk.Stack(app, "TopicsStack")
# 今回テストするスタックが参照するトピックを作成します。
topics = [sns.Topic(topics_stack, "Topic1")]
# ProcessorStackを作成します。
processor_stack = ProcessorStack(
app, "ProcessorStack", topics=topics # Cross-stack reference
)
# アサーションのためにテンプレートを生成します。
template = Template.from_stack(processor_stack)これで、Lambda関数とサブスクリプションが作成されたことをアサートできます。
# ...
# 正しいプロパティを持つLambda関数が作成されることを確認します。
template.has_resource_properties(
"AWS::Lambda::Function",
{
"Handler": "handler",
"Runtime": "nodejs14.x",
},
)
# サブスクリプションが作られることを確認します。
template.resource_count_is("AWS::SNS::Subscription", 1)
# ...has_resource_properties() メソッドを使用すると、 テンプレートが指定された型のリソースを指定されたプロパティで持っていることをアサートすることができます。この例では、期待通りのリソースがそれぞれ存在し、テストで定義した特定のプロパティが設定されていることをアサートします。またテンプレートクラスには他にも多くのメソッドがあり、CloudFormationテンプレートのResources、Outputs、Mappingsの検証に使用できます。
マッチャー(Matcher)
前述の例のアサーションを見ると、テストされるプロパティはすべてリテラル値として定義されていることがわかります。assertionsモジュールのメソッドは、テンプレートアサーション中に部分的または特別なパターンマッチングを定義することができるマッチャーを使用することもできます。assertionsモジュールにはいくつかのマッチャーが組み込まれており、その全リストは assertions API reference に記載されています。たとえば、Match.object_like() メソッドは、期待する値 (has_resource_properties() に渡されるプロパティ) がターゲット値 (生成されたリソース上のプロパティ) のサブセットであることをチェックします。Match.object_like() マッチャーは、CDKライブラリのアップデートによって追加のプロパティが導入された場合にテストが失敗するのを防ぐためによく使われます。
次の例では、ユニットテストでさまざまなマッチャーを使用する方法を示します。Match.object_equals()マッチャーは、期待される値がサブセットではなくターゲットの値と完全に等しいかどうかをチェックします。一方、Match.any_value()マッチャーは、ターゲットの値が何であってもアサーションが通るようにします。これらの2つのマッチャーを組み合わせて使用することで、ステートマシンのIAM roleに対して完全にアサーションを行うことができます。
from aws_cdk.assertions import Match
# ...
# ステートマシンのIAMロールをマッチャーで完全にアサートします。
template.has_resource_properties(
"AWS::IAM::Role",
Match.object_equals(
{
"AssumeRolePolicyDocument": {
"Version": "2012-10-17",
"Statement": [
{
"Action": "sts:AssumeRole",
"Effect": "Allow",
"Principal": {
"Service": {
"Fn::Join": [
"",
[
"states.",
Match.any_value(),
".amazonaws.com",
],
],
},
},
},
],
},
}
),
)次の例では、Step Functionsステートマシンのアサーションがユニットテストに含まれています。ステートマシンは、Amazon State Languages (JSONベース)で書かれた文字列で定義されます。これはステートマシンを表現するためには優れていますが、助けとなるツールがなければテストするのは困難です。幸いなことに、assertionsモジュールには Match.serialized_json() というマッチャーが用意されており、対象となる文字列を JSON でデシリアライズし、期待される値と照合します。Match.serialized_json()マッチャーの中にマッチャーを入れ子にすることもできます。
# ...
# serialized_json matcherでステートマシンの定義をアサートします。
template.has_resource_properties(
"AWS::StepFunctions::StateMachine",
{
"DefinitionString": Match.serialized_json(
# Match.object_equals()は暗黙的に使用されますが、
# ここでは分かりやすくするために明示的に使用しています。
Match.object_equals(
{
"StartAt": "StartState",
"States": {
"StartState": {
"Type": "Pass",
"End": True,
# "Next"が存在しないことを確認しています。
# "Next"と"End"が両方Trueになることはありません。
"Next": Match.absent(),
},
},
}
)
),
},
)キャプチャー(Capture)
assertionsモジュールの Capture API を使用すると、assertionsモジュールがマッチングする際の値を取得し、後でそれらの値に対して独自のアサーションを実行することができます。Capture オブジェクトを作成し、他のマッチャーと同じようにアサーションで使用し、 関連する as_x()メソッド (as_string() や as_object() など) でその値を取得します。
このスニペットでは、CaptureとMatch.serialized_json()マッチャーの組み合わせを使用して、ステートマシンの開始状態の名前が “Start” で始まること、および開始状態がマシンの状態リスト内に実際に存在することをアサートしています。
import re
from aws_cdk.assertions import Capture
# ...
# ステートマシンの定義からいくつかのデータをキャプチャします。
start_at_capture = Capture()
states_capture = Capture()
template.has_resource_properties(
"AWS::StepFunctions::StateMachine",
{
"DefinitionString": Match.serialized_json(
Match.object_like(
{
"StartAt": start_at_capture,
"States": states_capture,
}
)
),
},
)
# 開始状態が "Start"で始まることをアサートします。
assert re.match("^Start", start_at_capture.as_string())
# ステートマシン定義のstatesオブジェクトに
# 開始状態が実際に存在することをアサートします。
assert start_at_capture.as_string() in states_capture.as_object()スナップショットテスト
CDKのアプリケーションをテストするもう一つの方法として、スナップショットテストがあります。スナップショットテストは、最初の実行時にオブジェクトのスナップショットを取得します。このスナップショットはバージョンコントロールにコミットされ、その後テストが実行されるたびに、オブジェクトとスナップショットが比較されます。スナップショットがオブジェクトと一致した場合、アサーションはパスします。スナップショットが一致しない場合には、アサーションは失敗します。
スナップショットテストは、標準的なユニットテストとは異なり、リグレッションを検出するメカニズムではありません。AWS CDKアプリの生成時に生成されるCloudFormationテンプレートは、CDKアプリケーションのコードとCDKフレームワークの両方から影響を受けます。場合によっては、CDKフレームワークのバージョンがアップグレードされると、生成されるテンプレートが変わることがあります。これは通常、新しいベストプラクティスがCDKに組み込まれたときに起こります。このような理由から、スナップショットテストは、CDKスタックに何か変化があった場合に警告を発する仕組みとして使用するのが最適です。スナップショットテストを行うことで、これらの変更をすばやく確認することができます。
CDKのコードをリファクタリングする場合も、スナップショットテストが有効です。リファクタリング中に何かが変わってしまうことは避けたいものですが、スナップショットテストはそれが起きたときに明確に教えてくれます。その他のほとんどのユースケースでは、きめ細かなアサーションがより良いツールとなります。
assertionsモジュールでスナップショットテストを行うには、まずスタックからCloudFormationテンプレートに生成し、テンプレート全体をオブジェクト (JavaではMap、Pythonではdict) に変換し、テストフレームワークのスナップショットテスト機能を使って、テンプレートがそのスナップショットと一致することをアサートします。
from aws_cdk import aws_sns as sns
from aws_cdk import core as cdk
from aws_cdk.assertions import Template
from app.processor_stack import ProcessorStack
# snapshotパラメータはPytestによって注入されます。
# これはスナップショットテストライブラリであるsyrupyが提供するfixtureです。
# https://docs.pytest.org/en/stable/fixture.html
def test_matches_snapshot(snapshot):
# Set up the app and resources in the other stack.
app = cdk.App()
topics_stack = cdk.Stack(app, "TopicsStack")
topics = [sns.Topic(topics_stack, "Topic1")]
# ProcessorStackを作成します。
processor_stack = ProcessorStack(
app, "ProcessorStack", topics=topics # Cross-stack reference
)
# テンプレートを生成します。
template = Template.from_stack(processor_stack)
assert template.to_json() == snapshotコンストラクトのテスト
assertionsモジュールを使用すると、スタックと同様にコンストラクトをテストすることができます。その場合はテストの中でコンストラクトを格納するスタックを新しく作成します。
例えば、DeadLetterQueueというコンストラクトがあるとします (以前 Testing infrastructure with the AWS Cloud Development Kit の記事で使用したものです)。
from aws_cdk import aws_cloudwatch as cloudwatch
from aws_cdk import aws_sqs as sqs
from aws_cdk import core as cdk
class DeadLetterQueue(sqs.Queue):
def __init__(self, scope: cdk.Construct, id: str):
super().__init__(scope, id)
self.messages_in_queue_alarm = cloudwatch.Alarm(
self,
"Alarm",
alarm_description="There are messages in the Dead Letter Queue.",
evaluation_periods=1,
threshold=1,
metric=self.metric_approximate_number_of_messages_visible(),
)これはこのようにテストできます。
from aws_cdk import core as cdk
from aws_cdk.assertions import Match, Template
from app.dead_letter_queue import DeadLetterQueue
def test_creates_alarm():
stack = cdk.Stack()
DeadLetterQueue(stack, "DeadLetterQueue")
template = Template.from_stack(stack)
template.has_resource_properties(
"AWS::CloudWatch::Alarm",
{
"Namespace": "AWS/SQS",
"MetricName": "ApproximateNumberOfMessagesVisible",
"Dimensions": [
{
"Name": "QueueName",
"Value": Match.any_value(),
},
],
},
)まとめ
テストは、コードが期待通りに動作することを確認し、変更を行う際のデグレや予期せぬ変更を防ぐために重要です。新しいAWS CDKのassertionsモジュールは、特にassertモジュールが利用できない言語で開発している場合に、インフラストラクチャをコードとしてテストするための新しい強力な方法を提供します。assertionsモジュールの詳細については、API referenceを参照してください。aws-cdk の GitHub リポジトリでは、バグレポート、機能リクエスト、プルリクエストを受け付けています。
この記事はTesting CDK Applications in Any Languageを翻訳したものです。翻訳はPrototype Engineerの工藤が担当しました。