こんにちは、0日後に育休に入る志水です。本当は100日前に投稿したかったです。

みなさん、IaCしてますか？AaCしてますか？してますよね。
じゃあテストもコードで書いてますか？
と聞くと、やってない人も出てくるのかなと思います。

やっている人だとawspec使ってAWSリソースのテストをしたり、CDKの単体テストをコードで書いている人は多いかと思いますが、それ以降のテスト（結合テストやE2Eテストなど）をコードで書いてる人は少なくなるかなと思います。
というのも、awspecやCDKの単体テストの記事はよく見ますが、それ以降のテストの記事が無いなと感じました。
またCDKの単体テストでは、よくTypeScriptを使ったJestの記事が多く（公式もそうだし）、Pythonの方法があまりありませんでした。
もちろん要件によって変わってくるので書きづらいというのもあるとは思いますが、他の人がどういうテストを自動化しているのかを知るのは大事かなと思ったので、本記事ではPython(pytest)を使った私がよくやってるCDKのテストについて紹介します。

テストの種類の定義

まずテストの種類について定義をしたいと思います。
あくまで私がそう考えて利用しているだけで、正解ではなく本記事での定義とさせて頂きます。（そもそもテストの名前なんて何でも良いんじゃないか説はあります）
大きく分けると下記4つのテストに分けられると思います。

静的解析
単体テスト
結合テスト
E2Eテスト

上の方がテストの実行時間が短く繰り返し実行する事が可能となり、下の方が実際の利用方法に近いテストとなるため、要件がそのまま反映されたテストとなります。
テスト対象や範囲はCDKの流れに当てはめると下記となります。

静的解析、単体テストはローカル内での実行とし実行速度重視としています。
Test SizesでいうところのSmall testに当たります。
結合テスト、E2Eテストが実際のAWSリソースなどへのテストとなり、要件重視となっています。
静的解析は、コードに対してルールに沿っているかの確認をするテストとなります。
単体テストは、想定しているAWSリソースを作成しようとしているか確認したり、スタックのクラスの引数に正常・異常値を入れたときの挙動確認になります。
結合テストは、1つもしくは複数のAWSリソースが想定通りの設定・動きか確認になります。
E2Eテストは、要件通りのシステムの動きか確認するものになります。

まとめたものが下記になります。

テスト種別	実行環境	実行速度	テスト対象	目的
静的解析	ローカル	高速	コード	コードがルールに沿っているか確認
単体テスト	ローカル	速い	コードのAWSリソース、クラス	想定しているAWSリソースを作成しようとしているか確認
結合テスト	ローカル＋リモート	普通	AWSリソース	1つもしくは複数のAWSリソースが想定通りの設定・動きか確認
E2Eテスト	リモート	遅い	システム	要件通りのシステムの動きか確認

静的解析

静的解析とは、ソースコードを解析してコーディングパターンに沿っているかの確認を行うものです。
今回PythonでのCDKを利用する場合、Pythonコードとしての静的解析とCDKから変換したCloudFormationテンプレートの静的解析が利用出来ます。
それぞれの利用出来るツールとしては下記となります。

Python
- flake8
- Pylint
- Prospector
CloudFormation
- cfn-python-lint
- cfn-lint
- Conftest
- cfn-nag

Python LinterはVSCodeに仕込んだりGitのpre-commit hookに仕込むと使いやすいかなと思います。
CloudFormation LinterはCDKから一度 $ cdk synth でCloudFormationテンプレートに変換した後に静的解析を行うことになるので、makeタスクにして手動実行したり、pre-commit hookに仕込むことが可能かと思います。Python Linterと違い、リアルタイムに解析は出来ませんが、そもそもCDKから変換されているため、ある程度信頼が出来るCloudFormationテンプレートとなっているため、そこまで気にして解析を行う必要性も無いかもしれないです。

単体テスト

次に単体テストについて説明します。
単体テストでは、通信が発生せずローカル内で完結出来るテストとしています。そのため、単体テストでは通信が発生せず時間が短くなり、テストを頻繁に流せます。
もし、CDKのコードにAWS SDKのコードが入っている場合は、その部分のテストだけ連結テスト側に持っていきます。

CDKの単体テストとして公式の記事で下記の3種類が紹介されています。

Snapshot tests
Fine-grained assertions
Validation tests

Snapshot tests

Snapshot testsは、前回保存したベースラインのCFnテンプレートと、今のテンプレートを見比べて同じであることをテストになり、リファクタ時に想定外な変更が入らないようにするために利用されます。
ただし、公式の資料でもTypeScriptのJestを利用した手順であり、Pythonのpytestでは用意されていないです。
その為、このテストを行いたい場合は自作する必要があります。

Fine-grained assertions

Fine-grained assertionsは、想定されるリソースがテンプレートに入っているかチェックするテストです。
Snapshot testsよりリソース単位、設定単位で見る事が出来るので、より細かいテストとなります。
テストする粒度に気をつけないと、何をテストしているか分かりづらくなるので、注意が必要です。
基本的にはデフォルトから変更した部分などをメインに確認すると良いかと思います。

Validation tests

Validation testsは無効なデータを渡した時のテストとなります。
例えばprd/devなどの環境名をスタックに渡すように設計している時に、想定していない環境名を渡した時のテストとなります。
環境の渡し方としては、context利用かスタックの引数などで渡せるので、この辺りをテストすることになるかと思います。

上記3つが定義されてはいるのですが、私が主に業務で実施しているのはFine-grained assertionsで残りはあまり実施出来ていません。というのも、Fine-grained assertionsである程度カバー出来るのと、Snapshot testsに関しては次に説明する連結テストのdiffテストでカバー出来ると思っているためです。

Terraformの場合

ちなみに、Terraformでは基本的にStateをリモート側に持っていっていくと思うので、ローカルでの確認方法はlocalstackを利用しFakeオブジェクトをデプロイしたテストになります。
ただ、この方法はローカル分の構築(dockerなどで)が必要になり管理対象が増えるのと、結合テストとして、実際にAWS環境へデプロイして確認するawspec、testinfra、Terratestでテストした方が効率が良いかなと思うので、あまり私はやってないです。

結合テスト

次に結合テストです。
結合テストはローカルでのテストではなく、AWSリソース側のテストを含むため、インターネットへの通信が発生します。
その為、単体テストより時間がかかるテストとなります。
確認内容としては、システム全体の要件ではなく、それを満たすための一部の要件を満たしているかのチェックを行う事が結合テストで求められていることになります。また、ここでCDKが正しくデプロイされているかのテストもします（diffテスト）。
どのようなテストを普段行っているかを下記4つほど例に上げて説明します。

diffテスト
パイプラインテスト
通信のテスト
署名付きURLテスト

diffテスト

まずdiffテストについて説明します。
CDKのコードの単体テストを行っていて、コードとして問題なくても実際にデプロイしていない可能性があるので、コードの中身とAWSリソースが同一かどうかを確認する必要があります。
下記がそのコードになります。やっていることは単純で、cdk diffで差分がない事を確認しています。

import subprocess

def test_cdk_diff():
  command = 'cdk diff hogehoge-stack'
  output = subprocess.check_output(command, shell=True, text=True, stderr=subprocess.STDOUT)
  assert('There were no differences' in output)

ちなみに、Terraformなら terraform plan で No changes をチェックすれば同じように可能です。

パイプラインテスト

次にパイプラインテストです。
下記のように開発者がGitHubにコードをpushするとCodePipeline経由でFargateにデプロイされ、一般ユーザがWebアクセスが出来るパイプラインを考えます。

この時、リポジトリpushからWebアクセスまでを確認するのが難しいので、パイプラインのみを確認し、問題なく完了してることを確認するテストが必要になります。
確認するためのコードは下記です。

import boto3

def check_codepipeline_status():
  client = boto3.client('codepipeline')
  pipeline_name = 'hogeline'
  response = client.list_pipeline_executions(
    pipelineName = pipeline_name,
    maxResults = 1
  )
  codepipeline_exec_id = response['pipelineExecutionSummaries'][0]['pipelineExecutionId']
  response = client.get_pipeline_execution(
    pipelineName = pipeline_name,
    pipelineExecutionId = codepipeline_exec_id
  )
  
  assert(response['pipelineExecution']['status'] == 'Succeeded')

通信のテスト

次に通信のテストになります。
通信のテストとは、システムの中の一部の通信が正しく通っているかのテストになります。
例えば、アプリAからアプリBへの通信を確認したいが出来ない場合も多くあると思います。
その場合、ネットワークの場合はVPC Reachability Analyzer、IAMの場合は IAM Policy Simulatorが利用可能です。
それぞれ、通信をシミュレーションして通信可能かどうかの確認が可能なので、これをテストに組み込む事で通信のテストが可能です。
一点注意しないといけないのは、VPC Reachability Analyzerは確認をするたびに費用がかかるので、自動テストに組み込むと大変な費用が請求される可能性があるので、こちらは手動での確認に留める方が良いかと思います。

署名付きURLテスト

次は署名付きURLテストです。
例えば、CDN(ここではAkamaiとします)を利用して、S3にHTMLを配置して静的サイトを公開している場合を考えます。

この時にAkamaiからの通信ではなくS3単体として問題ない事を確認したい場合に、テストコードからS3へ署名付きURLを発行し、アクセスすると200で返ってくることを確認しています。
コードとしては下記のようになります。

import boto3
import requests

def test_presigned_response():
  client = boto3.client('s3')
  presigned_url = client.generate_presigned_url(
    ClientMethod='get_object',
    Params={
      'Bucket': 'hoge-bucket',
      'Key': '/'
    },
    ExpiresIn=60,
    HttpMethod='GET'
  )
  response = requests.get(presigned_url)
  assert(response.status_code == 200)

E2Eテスト

最後にE2Eテストです。
E2Eテストは、要件通りのシステムの動きか確認するテストとなります。つまり、システムを利用する一般ユーザや運用者がこのシステムに対して求める動きのテストを行います。
このテストはシステムの全てを通るようなテストになるので一番時間がかかるテストになります。
また、脆弱性テストや負荷テストもここに含められるかなと思います。
どのようなテストを普段行っているかを下記2つほど例に上げて説明します。

Webアクセステスト
監視通知テスト

Webアクセステスト

Webアクセステストとは、名前そのままで下記のようにALB->Fargate->Auroraで構築したWebサイトをインターネット越しに叩いて確認するテストです。

その為、ユーザと同じ立ち位置にテストコードがいて、そこからアクセスをしてステータスコードを確認することになります。

import requests

def test_access_site():
  url = 'http://example.com'
  response = requests.get(url)
  assert(response.status_code == 200)

監視通知テスト

次に監視通知テストです。
こちらも運用者と同じ立ち位置で監視や通知の内容を確認するテストとなります。
その時に気にする必要があるのは、どうやって通知を起こすかという部分になります。
実際に障害が起こせるような対象であるかどうかです。

まず障害が起こせるパターンで考えてみましょう。
先程のWebアクセスの環境に対してDatadogから外形監視を行っていると考えます。
また、FargateのコントロールプレーンがECSと考えた時の図が下記です。

そして、ECSだとFargate上のタスクを落とす事は可能になります。
なので、下記コードでタスクを落としてDatadogの確認を行えば可能です。
Datadogの確認もdatadogpyで取得してassertをかければ自動化が可能です。

ecs_client.stop_task(
  cluster ='hoge-cluster',
  task = 'task-id',
  reason = 'alert test'
)

ちなみに、ECSタスクは基本的に自動で上がってくるので、私は地道にloopして起き上がってくるタスクをポコポコ倒していって通知を行っていました。
もちろん対応しているサービスであればカオスエンジニアリングサービスであるGremlinやAWS Fault Injection Simulatorでの実施でも良いかと思います。

次に障害が起こせないパターンを考えます。
例えばECSタスクのメモリを監視していて、値はCloudWatchに入り、メトリクスをDatadogに連携し、閾値を超えるとPagerDutyに通知されて、ユーザが確認できるというフローとします。
（ECSのリソース監視必要か？という問題はありますが。。）

この時、ECS Execでログインしてもメモリに負荷をかけられるコマンドがない場合など、外からECSタスクのメモリを制御出来ない場合が考えられます。その場合、本来はECSからCloudWatchに対してメトリクス値がputされますが、それをテストコードから擬似的にputすることで監視の通知確認が可能になります。
その後、PagerDutyのアラートを確認するようにすれば障害が起こせないパターンでも自動テストが可能になります。
他にも監視の閾値を変更するという方法もありますが、あくまで私個人の意見ですが、テスト対象の設定を変更してしまうと、テスト環境が変化してしまい、変化後の環境でのテストとなるので、正しくテストは出来ていないかなと思っています。