背景

こんにちは。
普段よく使うIaCツールにAWS Cloud Development Kit (AWS CDK)がありますが、 長い間気になっていたものにテスト機能があります。
せっかくなので、勉強も兼ねてCDKのテスト機能を試してみました。

検証環境

今回は以下の環境で検証しています。また、CDKのコードはPythonを用いて実装します。

> aws --version
aws-cli/2.15.16 Python/3.11.6 Windows/10 exe/AMD64 prompt/off
> cdk --version
2.164.0 (build 75cf2e0)
> Python -V
Python 3.11.8
> pytest --version
pytest 6.2.5

platform win32 -- Python 3.11.8, pytest-6.2.5, py-1.11.0, pluggy-1.5.0

ドキュメントを読む

まずは公式ドキュメントを読んでみました。
CDKにはAWSのリソースを定義するモジュール以外にも、テスト目的で使用するモジュールがassertionsとして用意されています。このモジュールを使用することで、CDKによって作成されたCloudFormationのテンプレートが想定通りに作成されているか確認することができます。また、assertionsを用いて実装されたテストコードはpytestなどの一般的なテストフレームワークが使えるため、他のテストコードと同じ感覚でテストを行うことができます。
CDKのテストには2種類のアプローチがあります。

  • きめ細かなアサーション：CDKによって出力されたCloudFormationテンプレートに対して、想定されたパラメータと一致しているか確認するテストです。
  • スナップショットテスト：CDKのコードを変更した場合に、新しく出力されるCloudFormationのテンプレートと変更前のCloudFormationテンプレートの間で差分が発生しないか確認するテストです。CloudFormationのテンプレートに変更が発生しないことを確認できるため、CDKのコードに対するリファクタリングが容易になりそうです。

「きめ細かなアサーション」の方は使う機会が多そうであるため、こちらを深堀りしていきます。

テスト対象コード

今回の検証で使用するテスト対象のCDKコードは、VPC周りとEC2インスタンス2つ、IAMロールを構築する単純なコードです。長くなるため、折りたたんでいます。

テスト対象のソースコード
from constructs import Construct
from aws_cdk import (
    Stack,
    aws_ec2 as ec2,
    aws_iam as iam
)

class CdkTestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # VPCとサブネットを定義
        vpc = ec2.Vpc(
            self,
            id="test-vpc",
            vpc_name="test-vpc",
            availability_zones=["us-west-2a","us-west-2b"],
            ip_addresses=ec2.IpAddresses.cidr("192.168.0.0/16"),
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name="test-subnet-public",
                    subnet_type=ec2.SubnetType.PUBLIC
                ),
                ec2.SubnetConfiguration(
                    name="test-subnet-private",
                    subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
                ),
            ]
        )

        # EC2インスタンス用IAMロール
        iam_role = iam.Role(
            self,
            "test-ec2-role",
            assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    "AmazonSSMManagedInstanceCore"
                ),
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    "AmazonS3FullAccess"
                ),
            ],
        )

        # EC2インスタンス共通設定
        ec2_common_params = dict(
            vpc=vpc,
            role=iam_role,
            instance_type=ec2.InstanceType.of(
                instance_class=ec2.InstanceClass.T3,
                instance_size=ec2.InstanceSize.MICRO
            ),
            machine_image=ec2.MachineImage.latest_amazon_linux2()
        )

        # EC2インスタンス1台目
        ec2.Instance(
            self,
            id="test-ec2-instance01",
            instance_name="test-ec2-instance01",
            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC, availability_zones=["us-west-2a"]),
            **ec2_common_params
        )

        # EC2インスタンス2台目
        ec2.Instance(
            self,
            id="test-ec2-instance02",
            instance_name="test-ec2-instance02",
            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC, availability_zones=["us-west-2b"]),
            **ec2_common_params
        )

テストコード実装

実際にCDKのテストコードを書き、pytestでテスト実行をやってみました。

テンプレートの呼び出し

CDKのテストでは、生成されたCloudFormationのテンプレートを利用します。そのため、事前準備としてテストコード上でテンプレートを扱うためのスタック及びテンプレートを作成します。

import aws_cdk as cdk
import aws_cdk.assertions as assertions
from cdk_code.cdk_stack import CdkTestStack

app = cdk.App()

# テストでの確認対象となるスタック・テンプレートの作成
stack = CdkTestStack(app, "CdkTestStack", env=cdk.Environment(region="us-west-2"))
template = assertions.Template.from_stack(stack)

リソース数の確認

指定したリソースタイプにおいて、テンプレートで定義されているリソース数が一致するか確認するテストです。今回は2種類のサブネットを2つのAZで作成するため、合計4つのサブネットが作成されます。よって以下のテストは成功します。

def test_subnet_resource_count_is():
    # 指定したリソースタイプのリソース数が想定数と一致するか確認
    template.resource_count_is(
        type="AWS::EC2::Subnet", 
        count=4
    )

指定したパラメータが含まれているか確認

指定したリソースタイプにおいて、テンプレート内にパラメータが含まれているか確認するテストです。
テストコード内で指定したCloudFormationのPropertiesの設定内容が含まれている場合にテストは成功するため、特定の設定値が想定通りに設定されていることを確認できます。

def test_vpc_has_resource_properties():
    # 指定したリソースタイプのテンプレートにパラメータが含まれているか確認
    template.has_resource_properties(
        type="AWS::EC2::VPC",
        props={
            "CidrBlock": "192.168.0.0/16",
            "EnableDnsHostnames": True
        }
    )

なお、注意点としては、上記のテストは条件に一致するリソースが１つでも存在するとテストは成功する点です。
例えば、下記のテストはEC2インスタンスのAZがus-west-2bであるか確認するテストです。
今回のテンプレートではus-west-2aとus-west-2bのAZに1台ずつインスタンスを定義していますが、このテストは成功します。

def test_ec2_has_resource_properties():
    # １つでも一致するリソースがあればテスト自体はPASSする
    template.has_resource_properties(
        type="AWS::EC2::Instance",
        props={
            "InstanceType": "t3.micro",
            "AvailabilityZone": "us-west-2b"
        }
    )

そのため、同じリソースタイプのリソースが複数存在する場合は、このテストだけ対象リソースがテストパターン通りの設定になっている保証はありません。すべてのリソースで共通する設定値に対してテストする場合には次節で紹介するテストメソッドを使用します。

すべてのリソースで同じ設定になっているか確認

リソース名の命名規則やすべてのリソースで共通する設定値に対して確認する際に利用するテストです。テストコード内で指定したCloudFormationのPropertiesの設定内容が指定したリソースタイプにおけるすべてのリソースに含まれている場合にテストは成功します。

def test_ec2_all_resource_properties():
    # すべてのリソースで同じ設定になっているか、デプロイ時に決まる値についてアサートをかける
    isinstance_name_capture = assertions.Capture()
    template.all_resources_properties(
        type="AWS::EC2::Instance",
        props={
            "InstanceType": "t3.micro",
            "Tags":[
                {
                    "Key": "Name",
                    "Value": isinstance_name_capture
                }
            ]
        }
    )

    # インスタンス名の形式を確認
    while (isinstance_name_capture.next()):
        assert re.match(r"^test-ec2-instance[0-9]{2}", isinstance_name_capture.as_string())

また、上記のテストコードではCapture()を使用しています。これはテストコード内でテンプレートを比較する際にはワイルドカードとして機能しつつ、生成されたテンプレートから一致する部分の値を取得できる機能です。別途assertを用いることで、テンプレート生成時に決まる値やテンプレートに入る値について確認することにも利用できます。

特殊な一致

配列に含まれている一部の値やネストされている中の一部オブジェクトなど特定の値のみ確認する場合に利用できる機能があります。Match()を利用することで、柔軟なテンプレートの確認が行えます。
以下はIAMロールを定義するテンプレートですが、プリンシパルとマネージドポリシーに設定された値のみを確認するテストコードです。Match.object_like()では、Principalに対する値が想定している値かどうかのみ確認しており、Statement内に含まれる他の項目についてはテスト結果に影響しません。またMatch.array_with()では、マネージドポリシーを列挙するManagedPolicyArnsの配列内に「AmazonSSMManagedInstanceCore」が含まれているかのみ確認しています。

def test_iamrole_match():
    # matchを用いたパラメータ確認
    template.has_resource_properties(
        type="AWS::IAM::Role",
        props= {
            "AssumeRolePolicyDocument": 
                {
                    "Statement": [
                        # 指定したPrincipalが存在しているかのみ確認（他の項目については考慮しない）
                        assertions.Match.object_like(
                            {
                                "Principal": {
                                    "Service": "ec2.amazonaws.com",
                                },
                            }
                        ),
                    ],
                },
            "ManagedPolicyArns": assertions.Match.array_with(
            # 配列内に指定した値が含まれているかのみ確認
                [
                    {
                        "Fn::Join": [
                            "",
                            [
                                "arn:",
                                assertions.Match.any_value(),
                                ":iam::aws:policy/AmazonSSMManagedInstanceCore"
                            ]
                        ]
                    }
                ]
            )
        }
    )

まとめ

今回はCDKのテストについて調査してみました。CDKで用意されているテスト用のassertionsモジュールを使用することで、作成されるCloudFormationのテンプレートが想定通りの設定値で構成されているか確認することができます。
pytestなどの使い慣れたテストツールがそのまま利用できることや、パイプラインのテストフェーズなどに組み込むことで、デプロイ時に想定外の変更が発生していないか確認できることからリソースの保護機能としても利用できるメリットがある一方、CDKのテストにはCloudFormationのテンプレートを用いるため、テストコードを実装するにあたってはCloudFormationの知識も必要になり、学習コストが高くなるデメリットは感じられました。

テストコード全体
import aws_cdk as cdk
import aws_cdk.assertions as assertions
from cdk_code.cdk_stack import CdkTestStack
import re

app = cdk.App()
# テストでの確認対象となるスタック・テンプレートの作成
stack = CdkTestStack(app, "CdkTestStack", env=cdk.Environment(region="us-west-2"))
template = assertions.Template.from_stack(stack)

def test_vpc_resource_count_is():
    # 指定したリソースタイプのリソース数が想定数と一致するか確認
    template.resource_count_is(
        type="AWS::EC2::Subnet",
        count=4
    )

def test_vpc_has_resource_properties():
    # 指定したリソースタイプのテンプレートにパラメータが含まれているか確認
    template.has_resource_properties(
        type="AWS::EC2::VPC",
        props={
            "CidrBlock": "192.168.0.0/16",
            "EnableDnsHostnames": True
        }
    )

def test_ec2_has_resource_properties():
    # １つでも一致するリソースがあればテスト自体はPASSする
    template.has_resource_properties(
        type="AWS::EC2::Instance",
        props={
            "InstanceType": "t3.micro",
            "AvailabilityZone": "us-west-2b"
        }
    )

def test_ec2_all_resource_properties():
    # すべてのリソースで同じ設定になっているか、デプロイ時に決まる値についてアサートをかける
    isinstance_name_capture = assertions.Capture()
    template.all_resources_properties(
        type="AWS::EC2::Instance",
        props={
            "InstanceType": "t3.micro",
            "Tags":[
                {
                    "Key": "Name",
                    "Value": isinstance_name_capture
                }
            ]
        }
    )
    # isinstance_name_capture.next()
    while (isinstance_name_capture.next()):
        assert re.match(r"^test-ec2-instance[0-9]{2}", isinstance_name_capture.as_string())


def test_ec2_resource_properties_count_is():
    # 指定したパラメータをもつリソース数を確認
    template.resource_properties_count_is(
        type="AWS::EC2::Instance",
        props={
            "InstanceType": "t3.micro",
            "AvailabilityZone": "us-west-2a"
        },
        count=1
    )

def test_iamrole_match():
    # matchを用いたパラメータ確認
    template.has_resource_properties(
        type="AWS::IAM::Role",
        props= {
            "AssumeRolePolicyDocument": 
                {
                    "Statement": [
                        # 指定したPrincipalが存在しているかのみ確認（他の項目については考慮しない）
                        assertions.Match.object_like(
                            {
                                "Principal": {
                                    "Service": "ec2.amazonaws.com",
                                },
                            }
                        ),
                    ],
                },
            "ManagedPolicyArns": assertions.Match.array_with(
            # 配列内に指定した値が含まれているかのみ確認
                [
                    {
                        "Fn::Join": [
                            "",
                            [
                                "arn:",
                                assertions.Match.any_value(),
                                ":iam::aws:policy/AmazonSSMManagedInstanceCore"
                            ]
                        ]
                    }
                ]
            )
        }
    )

参考文献