CDK with Pythonの単体テストの話

本記事は NRIネットコム Advent Calendar 2022 10日目の記事です。
🎁 9日目 ▶▶本記事 ▶▶ 11日目 🎄

　はじめに
CDK with Python における単体テスト
さいごに

　はじめに

入社して3ヶ月目の長内です。はじめてのブログ投稿になります。

NRIネットコムにインフラエンジニアとして入社し、主にAWSを利用したインフラ構築を行っています。
前職でもAWSを利用する機会はあり、構築にはTerraformを利用していましたがNRIネットコムで私が携わるプロジェクトでは、AWS Cloud Development Kit (CDK with Python) を利用しています。

CDKに関して日々学習しているわけですが、CDKのコードはTypeScriptが標準的でPythonを利用したコードはTypeScriptに比べて検索結果が少ないように感じています。
インフラエンジニアの方は、TypeScriptよりPythonに触れる機会が多いのかなと思っているのでCDK with Pythonがもっと普及してほしいです。

前置きが長くなりましたが、本投稿ではCDK with Pythonの単体テストについて書いていきます。
何故かというと私がCDK with Pythonで単体テストのコードを作成しようと思い、はじめて公式の記事を見たら簡潔に書かれており、これだけでは単体テストの作成が難しかったからです。そのため、AWS公式の記事の補足となるようなブログ内容としました。

CDK with Python における単体テスト

本ブログで紹介するCDK with Pythonの単体テストには、pytestの使用したサンプルコードを載せています。また、CDK with Pythonの単体テストについてここ詳しくは説明しませんが「CDK with Pythonの自動テスト事情　〜TypeScriptなんて羨ましくなんかないぞ〜」がテストに関してまとまっているので是非読んでください。

AWS公式の記事の中から、下記の3種類について1つずつサンプルコードを用いて説明します。

Fine-grained assertions
Match
Snapshot tests

本ブログの動作環境は、下記となります。

CDK with Python 2.53.0
Python 3.10.6
pytest 7.2.0
syrupy 3.0.5

スタックの例

公式の記事と異なり、EC2とRDSをデプロイするスタックを利用します。

from constructs import Construct
from aws_cdk import (
    Stack,
    aws_ec2 as ec2,
    aws_rds as rds,
    aws_iam as iam,
)
class Ec2Rds(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        db_pass = self.node.try_get_context("dbpass")
        """
        VPC作成
        """
        vpc = ec2.Vpc(
            self,
            id="vpc",
            cidr="172.32.0.0/16",
            max_azs=2,
            enable_dns_hostnames=True,
            enable_dns_support=True,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name="public",
                    subnet_type=ec2.SubnetType.PUBLIC,
                    cidr_mask=24,
                ),
                ec2.SubnetConfiguration(
                    name="private",
                    subnet_type=ec2.SubnetType.PRIVATE_ISOLATED,
                    cidr_mask=24,
                ),
            ],
            nat_gateways=1,
        )
        """
        RDS作成
        """
        db_sg = ec2.SecurityGroup(
            self,
            id="db-sg",
            security_group_name="blog-db-sg",
            vpc=vpc,
            allow_all_outbound=True,
        )
        rds_instance = rds.DatabaseInstance(
            self,
            id="db",
            engine=rds.DatabaseInstanceEngine.mysql(
                version=rds.MysqlEngineVersion.VER_8_0_30
            ),
            # t3.micro
            instance_type=ec2.InstanceType.of(
                ec2.InstanceClass.T3,
                ec2.InstanceSize.MICRO,
            ),
            credentials=rds.Credentials.from_generated_secret(
                username="blog_db_user",
            ),
            vpc=vpc,
            vpc_subnets=ec2.SubnetSelection(
                subnet_type=ec2.SubnetType.PRIVATE_ISOLATED,
            ),
            security_groups=[db_sg],
        )
        rds_instance.node.add_dependency(vpc)
        """
        EC2(Web)作成
        """
        web_sg = ec2.SecurityGroup(
            self,
            id="web-sg",
            security_group_name="blog-web-sg",
            vpc=vpc,
            allow_all_outbound=True,
        )
        web_sg.add_ingress_rule(
           peer=ec2.Peer.any_ipv4(),
           connection=ec2.Port.tcp(22),
           description="SSH",
        )
        web_sg.add_ingress_rule(
            peer=ec2.Peer.any_ipv4(),
            connection=ec2.Port.tcp(80),
            description="HTTP",
        )
        web_sg.add_ingress_rule(
            peer=ec2.Peer.any_ipv4(),
            connection=ec2.Port.tcp(443),
            description="HTTPS",
        )
        ec2_instance = ec2.Instance(
            self,
            id="web",
            instance_type=ec2.InstanceType("t2.micro"),
            machine_image=ec2.AmazonLinuxImage(
                edition=ec2.AmazonLinuxEdition.STANDARD,
                generation=ec2.AmazonLinuxGeneration.AMAZON_LINUX_2,
                virtualization=ec2.AmazonLinuxVirt.HVM,
                storage=ec2.AmazonLinuxStorage.GENERAL_PURPOSE,
            ),
            vpc=vpc,
            key_name="blog-ssh-key",
            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC),
            security_group=web_sg,
        )
        ec2_instance.node.add_dependency(rds_instance)
        iam_policy = iam.PolicyStatement(
            effect=iam.Effect.ALLOW,
            actions=["rds:*"],
            resources=[rds_instance.instance_arn],
        )
        ec2_instance.role.add_to_principal_policy(iam_policy)
        """
        セキュリティグループ設定
        """
        lookup_db_sg = ec2.SecurityGroup.from_security_group_id(
            self, id="lookup-db-sg", security_group_id=db_sg.security_group_id
        )
        lookup_db_sg.add_ingress_rule(
            peer=ec2.Peer.security_group_id(web_sg.security_group_id),
            connection=ec2.Port.tcp(3306),
            description="DB Security Group",
        )

Fine-grained assertions

本テストではスタック（Ec2Rds）からAWS CloudFormation テンプレートを生成して、テンプレートの中に任意のプロパティがあることを確認します。

はじめにassertions.Template.from_stack にEC2Rdsのスタックを引数として渡すことで、CloudFormation テンプレートを生成することができます。生成したテンプレートは、template変数に代入されています。

import aws_cdk as cdk
from aws_cdk import assertions
from stacks.ec2_rds import Ec2Rds
def test_vpc_resource_properties():
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)

この時点で template変数には下記の内容が代入されています。（VPCのプロパティのみ抜き出しています）

{...},
'vpcA20221210': {
  'Properties': {
    'CidrBlock': '172.32.0.0/16',
    'EnableDnsHostnames': True,
    'EnableDnsSupport': True,
    'InstanceTenancy': 'default',
    'Tags': [
      {
        'Key': 'Name',
        'Value': 'Default/blog/vpc'
      }
    ]
  },
'Type': 'AWS::EC2::VPC'},
{...}

template変数には、CloudFormation テンプレートのスタック全体の構成情報が代入されています。
template.has_resource_properties を用いることで、スタック全体の構成情報の中から必要なプロパティだけを抜き出し、スタック内で指定しているプロパティがCloudFormation テンプレートに正しく設定されているか部分一致で確認することができます。

import aws_cdk as cdk
from aws_cdk import assertions
from stacks.ec2_rds import Ec2Rds
def test_vpc_resource_properties():
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)
    template.has_resource_properties(
        "AWS::EC2::VPC",
        {
            "CidrBlock": "172.32.0.0/16",
            "EnableDnsHostnames": True,
            "EnableDnsSupport": True,
        },
    )

また、resource_count_isを用いることで、作成するリソース数を確認できます。

def test_vpc_resource_count():
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)
    template.resource_count_is("AWS::EC2::VPC", 1)

Match

CDKでは一部プロパティをスタック作成時に指定していない場合、必要に応じてCDKが自動生成します。
テストコードで部分一致ではなく完全一致を行いたい場合、自動生成された文字列だけをassertions.Match.any_valueを用いることでワイルドカードとして許可することができます。

import aws_cdk as cdk
from aws_cdk import assertions
from stacks.ec2_rds import Ec2Rds
def test_ec2_iam_policy_resource_properties():
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)
    template.has_resource_properties(
        "AWS::IAM::Policy",
        assertions.Match.object_equals(
            {
                "PolicyDocument": {
                    "Statement": [
                        {
                            "Action": "rds:*",
                            "Effect": "Allow",
                            "Resource": {"Fn::Join": assertions.Match.any_value()},
                        }
                    ],
                    "Version": "2012-10-17",
                },
                "PolicyName": assertions.Match.any_value(),
                "Roles": [{"Ref": assertions.Match.any_value()}],
            },
        ),
    )

またassertions.Match.array_withを用いることで、複数のプロパティが代入された配列に対して部分一致を行うことができます。
下記のテストコードでは、SecurityGroupIngressにHTTPS/HTTPに対して制限無くアクセスできるようになっているのかを確認しています。

def test_security_group_ingress_resource_properties():
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)
    template.has_resource_properties(
        "AWS::EC2::SecurityGroup",
        {
            "SecurityGroupIngress": assertions.Match.array_with(
                [
                    {
                        "CidrIp": "0.0.0.0/0",
                        "Description": "HTTP",
                        "FromPort": 80,
                        "IpProtocol": "tcp",
                        "ToPort": 80,
                    },
                    {
                        "CidrIp": "0.0.0.0/0",
                        "Description": "HTTPS",
                        "FromPort": 443,
                        "IpProtocol": "tcp",
                        "ToPort": 443,
                    },
                ]
            )
        },
    )

Snapshot tests

本テストでは、新たにコード変更を行う前にスナップショット（CloudFormation テンプレート）を取得し、コード変更後にスナップショットとの差異を確認します。

スナップショットは、CDKのコードから生成されたCloudFormation テンプレートのため、CDKコードを変更したとしても生成されるテンプレートに変更が発生していない場合、テストが失敗することはありません。本テストの用途としては、CDKコードの文法修正などによる意図しない構成変更を防ぐことです。

本テストは、pytest以外にsyrupyというライブラリの追加が必要となります。
インストールは、下記コマンドで行います。

$ pip install syrupy

下記関数を定義するだけで、テストコードの準備は完了します。

import aws_cdk as cdk
from aws_cdk import assertions
from stacks.ec2_rds import Ec2Rds
def test_snapshot(snapshot):
    app = cdk.Stack()
    stack = Ec2Rds(app, construct_id="blog")
    template = assertions.Template.from_stack(stack)
    assert template.to_json() == snapshot

テスト実行前にスナップショットを取得します。スナップショットの取得には、--snapshot-updateオプションを指定するだけです。

$ pytest tests/unit/test_ec2_rds.py --snapshot-update

スナップショットは、下記のディレクトリに保存されています。

$ tree tests/unit/__snapshots__/
tests/unit/__snapshots__/
|-- test_ec2_rds.ambr
`-- test_vpc_ec2_rds.ambr

さいごに

CDK with Pythonを利用している方、これから利用しようと思っている方の役に立つブログになっていれば幸いです。

今回作成したコードはこちらです。

サンプルコード (aws_cdk_with_python_unit_test_sample)