CDK で Fine-Grained Assertion Test を行う

December 12, 2021

Fine-Grained Assertion Test とは

CDK により生成された CluodFomration テンプレート内にて、「リソース A はプロパティ B として C という値を持っているか」といったようなテストを行うためのものとなります。

pytest のインストール

ユニットテストを行う前に、まずは virtualenv に pytest をインストールしましょう。以下のコマンドでインストールできます。

$ python -m pip install pytest

DynamoDB テーブル用のテスト

cdk init でプロジェクトの初期化を行っている場合、以下のように tests ディレクトリ配下にファイルが既に生成されていると思います。

$ tree -L 2 tests/
tests/
├── __init__.py
├── __pycache__
│   └── __init__.cpython-39.pyc
└── unit
    ├── __init__.py
    ├── __pycache__
    └── test_cdk_workshop_stack.py

自動で作成された test_cdk_workshop_stack.py は新たに作成するので一度消してしまいましょう。

$ rm tests/unit/test_cdk_workshop_stack.py

あらたに同一のファイルを以下の内容で作成します。

tests/unit/test_cdk_workshop_stack.py
from aws_cdk import (
  Stack,
  aws_lambda as _lambda,
  assertions
)
from cdk_workshop.hitcounter import HitCounter
import pytest


def test_dynamodb_table_created():
  stack = Stack()
  HitCounter(stack, "HitCounter", 
    downstream=_lambda.Function(stack, "TestFunction", 
      runtime=_lambda.Runtime.PYTHON_3_9,
      handler='hello.handler',
      code=_lambda.Code.from_asset('lambda'),
    )
  )
  template = assertions.Template.from_stack(stack)
  template.resource_count_is("AWS::DynamoDB::Table", 1)

上記は HitCounter Construct により生成される CloudFormation テンプレートが AWS::DynamoDB::Table というリソースを 1 つ持つかどうかを確認しているテストとなります。実際に pytest により確認してみます。

$ pytest
============================================================================================================================================================= test session starts ==============================================================================================================================================================
platform darwin -- Python 3.9.6, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /Users/yuk/Development/cdk/cdk_workshop
collected 1 item                                                                                                                                                                                                                                                                                                                               

tests/unit/test_cdk_workshop_stack.py .                                                                                                                                                                                                                                                                                                  [100%]

============================================================================================================================================================== 1 passed in 3.58s ===============================================================================================================================================================

Lambda 関数用のテスト

次に HitCounter における Lambda 関数のテストを作成します。今回は Lambda 関数が作成されているかだけではなく、DOWNSTREAM_FUNCTION_NAME 及び HITS_TABLE_NAME とい環境変数も併せて作成されているかをチェックします。
なお、HitCounter では環境変数は他の Construct に対するリファレンスとなっていました。

cdk_workshop/hitcounter.py
self._handler = _lambda.Function(
  self, 'HitCountHandler',
  runtime=_lambda.Runtime.PYTHON_3_9,
  handler='hitcount.handler',
  code=_lambda.Code.from_asset('lambda'),
  environment={
    'DOWNSTREAM_FUNCTION_NAME': downstream.function_name,
    'HITS_TABLE_NAME': table.table_name,
  }
)

それでは test_cdk_workshop.py に以下のコードを追加しましょう。

tests/unit/test_cdk_workshop_stack.py
def test_lambda_has_env_vars():
  stack = Stack()
  HitCounter(stack, "HitCounter",
    downstream=_lambda.Function(stack, "TestFunction",
      runtime=_lambda.Runtime.PYTHON_3_9,
      handler='hello.handler',
      code=_lambda.Code.from_asset('lambda')
    )
  )
  template = assertions.Template.from_stack(stack)
  envCapture = assertions.Capture()

  template.has_resource_properties("AWS::Lambda::Function", {
    "Handler": "hitcount.handler",
    "Environment": envCapture,
  })

  assert envCapture.as_object() == {
    "Variables": {
      "DOWNSTREAM_FUNCTION_NAME": {"Ref": "TestFunctionXXXXX"},
      "HITS_TABLE_NAME": {"Ref": "HITCounterHitsXXXXXX"}
    } 
  }

実際に pytest を実行すると以下のように失敗します。

$ pytest

...

E     AssertionError: assert {'Variables':...ts079767E5'}}} == {'Variables':...HitsXXXXXX'}}}
E       Differing items:
E       {'Variables': {'DOWNSTREAM_FUNCTION_NAME': {'Ref': 'TestFunction22AD90FC'}, 'HITS_TABLE_NAME': {'Ref': 'HitCounterHits079767E5'}}} != {'Variables': {'DOWNSTREAM_FUNCTION_NAME': {'Ref': 'TestFunctionXXXXX'}, 'HITS_TABLE_NAME': {'Ref': 'HITCounterHitsXXXXXX'}}}
E       Use -v to get the full diff

tests/unit/test_cdk_workshop_stack.py:39: AssertionError
=========================================================================================================================================================== short test summary info ============================================================================================================================================================
FAILED tests/unit/test_cdk_workshop_stack.py::test_lambda_has_env_vars - AssertionError: assert {'Variables':...ts079767E5'}}} == {'Variables':...HitsXXXXXX'}}}
========================================================================================================================================================= 1 failed, 1 passed in 3.73s ==========================================================================================================================================================

テスト結果より、再度環境変数の参照先として正しい値を入れます。

tests/unit/test_cdk_workshop_stack.py
def test_lambda_has_env_vars():
  stack = Stack()
  HitCounter(stack, "HitCounter",
    downstream=_lambda.Function(stack, "TestFunction",
      runtime=_lambda.Runtime.PYTHON_3_9,
      handler='hello.handler',
      code=_lambda.Code.from_asset('lambda')
    )
  )
  template = assertions.Template.from_stack(stack)
  envCapture = assertions.Capture()

  template.has_resource_properties("AWS::Lambda::Function", {
    "Handler": "hitcount.handler",
    "Environment": envCapture,
  })

  assert envCapture.as_object() == {
    "Variables": {
      "DOWNSTREAM_FUNCTION_NAME": {"Ref": "TestFunction22AD90FC"},
      "HITS_TABLE_NAME": {"Ref": "HitCounterHits079767E5"}
    } 
  }

再度 pytest を実行しましょう。

$ pytest
============================================================================================================================================================= test session starts ==============================================================================================================================================================
platform darwin -- Python 3.9.6, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /Users/yuk/Development/cdk/cdk_workshop
collected 2 items                                                                                                                                                                                                                                                                                                                              

tests/unit/test_cdk_workshop_stack.py ..                                                                                                                                                                                                                                                                                                 [100%]

============================================================================================================================================================== 2 passed in 3.55s ===============================================================================================================================================================

今度は OK ですね。

ユニットテストで制約を儲ける

DynamoDB テーブルで暗号化が有効化されているかどうかを確認するテストを追加します。

tests/unit/test_cdk_workshop_stack.py
def test_dynamodb_with_encrption():
  stack = Stack()
  HitCounter(stack, "HitCounter", 
    downstream=_lambda.Function(stack, "TestFunction", 
      runtime=_lambda.Runtime.PYTHON_3_9,
      handler='hello.handler',
      code=_lambda.Code.from_asset('lambda'),
    )
  )
  template = assertions.Template.from_stack(stack)
  template.has_resource_properties("AWS::DynamoDB::Table", {
    "SSESpecification": {
      "SSEEnabled": True,
    },
  })

ここで pytest を実行してみましょう。

$ pytest

...

E           jsii.errors.JSIIError: Template has 1 resources with type AWS::DynamoDB::Table, but none match as expected.
E           The closest result is:
E             {
E               "Type": "AWS::DynamoDB::Table",
E               "Properties": {
E                 "KeySchema": [
E                   {
E                     "AttributeName": "path",
E                     "KeyType": "HASH"
E                   }
E                 ],
E                 "AttributeDefinitions": [
E                   {
E                     "AttributeName": "path",
E                     "AttributeType": "S"
E                   }
E                 ],
E                 "ProvisionedThroughput": {
E                   "ReadCapacityUnits": 5,
E                   "WriteCapacityUnits": 5
E                 }
E               },
E               "UpdateReplacePolicy": "Retain",
E               "DeletionPolicy": "Retain"
E             }
E           with the following mismatches:
E               Missing key at /Properties/SSESpecification (using objectLike matcher)

.venv/lib/python3.9/site-packages/jsii/_kernel/providers/process.py:326: JSIIError
=========================================================================================================================================================== short test summary info ============================================================================================================================================================
FAILED tests/unit/test_cdk_workshop_stack.py::test_dynamodb_with_encrption - jsii.errors.JSIIError: Template has 1 resources with type AWS::DynamoDB::Table, but none match as expected.
========================================================================================================================================================= 1 failed, 2 passed in 3.78s ==========================================================================================================================================================

ちゃんと Missing key at /Properties/SSESpecification (using objectLike matcher) というメッセージで失敗してることがわかります。
それでは hitcounter.py を編集し、暗号化オプションを追加してみましょう。

cdk_workshop/hitcounter.py
from constructs import Construct
from aws_cdk import (
  aws_lambda as _lambda,
  aws_dynamodb as ddb,
)

class HitCounter(Construct):

  @property
  def handler(self):
    return self._handler

  def __init__(self, scope: Construct, id: str, downstream: _lambda.IFunction, **kwargs):
    super().__init__(scope, id, **kwargs)

    table = ddb.Table(
      self, 'Hits',
      partition_key={'name': 'path', 'type': ddb.AttributeType.STRING},
      encryption=ddb.TableEncryption.AWS_MANAGED,
    )

...

再度 pytest を実行します。

$ pytest
============================================================================================================================================================= test session starts ==============================================================================================================================================================
platform darwin -- Python 3.9.6, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /Users/yuk/Development/cdk/cdk_workshop
collected 3 items                                                                                                                                                                                                                                                                                                                              

tests/unit/test_cdk_workshop_stack.py ...                                                                                                                                                                                                                                                                                                [100%]

============================================================================================================================================================== 3 passed in 3.76s ===============================================================================================================================================================

今度はパスします。


 © 2023, Dealing with Ambiguity