CDK でインテグレーションテストをやってみる
こんにちは、ししいです。
突然ですが皆さんインテグレーションテストやってますか？
AWS で実装してると結構難しいんじゃないかな？って思います。
我らが CDK さんが、インテグレーションテストをサポートしてくれているらしいので、やってみたいと思います。
インテグレーションテストとは？
複数のモジュールを統合したテスト、のことだそうです。
それだけだと「どこまでがモジュールになる？」とか「どれだけ統合すれば良き？」とか色々迷います。このあたり決まった定義とかないと思います。
なのでここでは Mock を使わないで API に対してリクエストを投げ（AWS だとピタゴラスイッチを発火させ）、その結果を確認するテスト を指すと思ってください。
「Mock を使わない」の具体だと AWS 上で動くサービスをローカルで開発するときに、DynamoDB とか SQS とか色々 Mock して開発やらテストやらを書いていると思います。が、それをまったく使わないで最初から最後までテストをしましょうという感じです。
はじめに
今回は CDK を使ってインテグレーションテストを行います。
全体構成ですが、こんな感じです。
具体的な動きは以下の通りです。
- ① クライアントから API Gateway にリクエストを送る
- ② API Gateway は認証 Lambda を呼び出す
- ③ 認証 Lambda は DynamoDB にアクセスして認証とユーザーを引き出す （サンプルなので適当認証。テーブルにユーザーがいれば認証 OK、いないと NG）
- ④ API Gateway が認証を通過したら、本来の Lambda を呼び出す
テスト内容は以下の３パターンを確認してみます。
- (1) 認証トークンがない場合のテスト
- (2) 認証トークンがあり、認証 NG
- (3) 認証トークンがあり、認証 OK
前提
- CDK のバージョンは 2.130.0
- 自分でよしなにできる AWS 環境がある
- CDK Bootstrap が完了している
- CDK for Python でおこないます。ただし現バージョンでは CDK for Python 側に大きなバグがあります。なので CDK のバージョンによってはこのコードは動かない可能性があるので、その辺りはご注意ください。
では 作っていきましょう 👻
テスト対象のサービスを作成する
今回デプロイする Stack は以下の通りです。
ではでは。
まずエントリーポイントですが、以下のようになります。
import aws_cdk as cdk
from api.stack import Api
app = cdk.App()
api = Api(app, "cdk-integ-tests-stack")
app.synth()
ほぼ空っぽですね。Api って Stack を作って、それを App に登録するだけです。
余談で、 普段ブログで書く場合は、コピペ１回で動作確認できるように１つのファイルにリソースすべてまとめてます。が、今回は deploy する Stack とテスト用の Stack が共通なものなのでファイルを分けてます。
次に ApiStack の中身です。
import pathlib
from aws_cdk import RemovalPolicy, Stack, aws_apigateway, aws_dynamodb, aws_lambda
from constructs import Construct
class Api(Stack):
def __init__(
self,
scope: Construct,
id_: str,
):
super().__init__(scope, id_)
# DynamoDB
table = aws_dynamodb.Table(
self,
"table",
partition_key=aws_dynamodb.Attribute(name="pk", type=aws_dynamodb.AttributeType.STRING),
removal_policy=RemovalPolicy.DESTROY,
billing_mode=aws_dynamodb.BillingMode.PAY_PER_REQUEST,
)
# Lambda
code = aws_lambda.Code.from_asset(str(pathlib.Path(__file__).parent.joinpath("runtime").resolve()))
authorizer_func = aws_lambda.Function(
self,
"authorizerFunc",
code=code,
runtime=aws_lambda.Runtime.PYTHON_3_11,
handler="authorizer.handler",
environment={"TABLE_NAME": table.table_name},
)
table.grant_read_data(authorizer_func)
hello_func = aws_lambda.Function(
self,
"helloFunc",
code=code,
runtime=aws_lambda.Runtime.PYTHON_3_11,
handler="hello.handler",
)
# API Gateway
api = aws_apigateway.RestApi(self, "apiGateway")
authorizer = aws_apigateway.TokenAuthorizer(
self,
"authorizer",
identity_source=aws_apigateway.IdentitySource.header("Authorization"),
handler=authorizer_func,
)
hello_resources = api.root.add_resource("hello")
hello_resources.add_method("GET", aws_apigateway.LambdaIntegration(hello_func), authorizer=authorizer)
# Output
self.url = api.url
self.table_name = table.table_name
ここは認証 Lambda と DynamoDB と API Gateway を作ってます。
一応過去にブログで触れているので、詳細はそちらに任せてここでは説明省略します。
またこの Stack はプロパティでurlとtable_nameを宣言してます。これはエントリーポイントでは利用してませんが、追々テストで使います。
urlには API Gateway の URL が入り、table_nameには DynamoDB のテーブル名が入ります。
続いて認証 Lambda の中身は以下です。
import os
import boto3
table_name = os.getenv("TABLE_NAME")
table = boto3.resource("dynamodb").Table(table_name)
def handler(event, context):
token = event["authorizationToken"]
resources = [event["methodArn"]]
return authorizer(token, resources)
def authorizer(user_id, resources):
"""
認証処理（実運用でこんな認証しないでね）
テーブルにidがあれば成功。なければ失敗。
"""
context = {}
effect = "Deny"
try:
response = table.get_item(Key={"pk": user_id})
context["userName"] = response["Item"]["user_name"]
effect = "Allow"
except Exception as e:
# 認証失敗
pass
return {
"principalId": "hoge",
"policyDocument": {
"Version": "2012-10-17",
"Statement": [{"Action": "execute-api:Invoke", "Effect": effect, "Resource": resources}],
},
"context": context,
}
ここは「こんな感じになってます」くらいでいいと思います。本質と関係ないので。
ここでは以下のような雑認証ロジックで認証します。
- クライアントから user_id を受け取り、DynamoDB のテーブルの pk にその user_id があれば認証 OK
また DynamoDB の item はuser_nameというキーを持っていて、これを Context に詰めて API 側の Lambda へ渡しています。
最後 API の Lambda です
def handler(event, context):
authorizer_context = event["requestContext"]["authorizer"]
user_name = authorizer_context["userName"]
return hello(user_name)
def hello(user_name):
return {
"statusCode": 200,
"body": f"Hello {user_name} !",
}
Context からuserNameを取得して Body にセットして返します。
ここで deploy してみます。
この記事はファイルが複数あってわかりにくいので、もしかしたらリポジトリ見てもらったほうが構成わかりやすいかもです。
$ cdk -a "python3 app.py" deploy
動作確認
deploy したものの動きを確認してみます。
まず deploy した API Gateway の URL を確認します。
コンソールにログインして取得するか、↓ みたいなコマンドで取得できます。
$ aws cloudformation describe-stacks --stack-name cdk-integ-tests-stack --query 'Stacks[0].Outputs' --output text
今回は ↓ でした
https://ni3jjuhbi3.execute-api.ap-northeast-1.amazonaws.com/prod/
このエントリーポイントの hello のリソースにリクエストを送ってみます。
$ curl https://ni3jjuhbi3.execute-api.ap-northeast-1.amazonaws.com/prod/hello --header ''
>> {"message":"Unauthorized"}
ヘッダーがないと{"message":"Unauthorized"}って怒られました。
次に認証用のヘッダーを付けますが、認証は通らない場合をためします。
$ curl https://ni3jjuhbi3.execute-api.ap-northeast-1.amazonaws.com/prod/hello --header 'Authorization: hoge'
{"Message":"User is not authorized to access this resource with an explicit deny"}
許可されてないよ！って怒られました。
次に DynamoDB に以下のユーザーを登録して再度リクエストを送ります。
まず ↓ な感じでテーブル名を取得します。ここではdk-integ-tests-stack-table8235A42E-1FVV7MFQ2T98Mでした。
$ aws cloudformation describe-stack-resources \
--stack-name cdk-integ-tests-stack \
--query 'StackResources[?ResourceType==`AWS::DynamoDB::Table`].PhysicalResourceId' \
--output text
>> cdk-integ-tests-stack-table8235A42E-1FVV7MFQ2T98M
次に pk がhogeのユーザーを登録します。
$ aws dynamodb put-item --table-name cdk-integ-tests-stack-table8235A42E-1FVV7MFQ2T98M --item '{
"pk": {"S": "hoge"},
"user_name": {"S": "hoge太郎"}
}'
最後にもう一度リクエストを送ります。
$ curl https://ni3jjuhbi3.execute-api.ap-northeast-1.amazonaws.com/prod/hello --header 'Authorization: hoge'
>> Hello hoge太郎 !
認証が通り、hello のリソースにアクセスできました。
今 deploy したものの動きは以上です。
テスト作成
インテグレーションテストは CDK の alpha 版に含まれているので、以下のようにインストールします。
$ pip install aws-cdk.integ-tests-alpha==2.130.0a0
また、テスト実行にはinteg-runnerを使いますので、こちらも ↓ な感じでインストールします。
$ npm install -g @aws-cdk/integ-runner
最後の準備ですが、cdk-integ-runner-cwd-fixをインストールします。
$ pip install cdk-integ-runner-cwd-fix
これなにか？という話ですが、CDK の alpha 版のテスト実行時にカレントディレクトリが変わってしまうバグがあるので、それを修正するためのものです。詳しくはここ
やっとテストがかけます。こんな感じ。
import aws_cdk as cdk
from aws_cdk.cloud_assembly_schema import CdkCommands, DestroyCommand, DestroyOptions
from aws_cdk.integ_tests_alpha import ExpectedResult, IntegTest, Match
from cdk_integ_runner_cwd_fix import fix_cwd
fix_cwd()
from api.stack import Api
app = cdk.App()
api = Api(app, "cdk-integ-tests-stack-test")
integ = IntegTest(
app,
"integ",
test_cases=[api],
cdk_command_options=CdkCommands(destroy=DestroyCommand(args=DestroyOptions(force=True))),
)
# ------------------------------------
# 認証テスト
# ------------------------------------
# ==> 認証失敗（トークンなし）
integ.assertions.http_api_call(url=f"{api.url}/hello", method="GET", headers={}).expect(
ExpectedResult.object_like(
{
"statusText": "Unauthorized",
"status": 401,
},
)
)
# ==> 認証失敗（ユーザーなし）
integ.assertions.http_api_call(url=f"{api.url}/hello", method="GET", headers={"Authorization": "fuga"}).expect(
ExpectedResult.object_like(
{
"statusText": "Forbidden",
"status": 403,
},
)
)
# ==> 認証成功
integ.assertions.aws_api_call(
# ユーザーを登録
"DynamoDB",
"putItem",
{
"TableName": api.table_name,
"Item": {
"pk": {"S": "hoge"},
"user_name": {"S": "hoge太郎"},
},
},
).next(
# リクエスト
integ.assertions.http_api_call(url=f"{api.url}/hello", method="GET", headers={"Authorization": "hoge"}),
).expect(
# 確認
ExpectedResult.object_like(
{
"body": Match.string_like_regexp("hoge太郎"),
"status": 200,
},
)
)
app.synth()
テスト全体の動きですが、テスト用の Stack を AWS の実際の環境へ deploy して、その後テストを実行します。
そしてこの「テスト」自体も AWS のカスタムリソースを使って実装されますので、テストも Stack の中のリソースの１つです。
なのでテスト全体は ↓ のようにappを作成して、きちんとapp.synth()する必要があります。
app = cdk.App()
... 略
app.synth()
そしてこのappにインテグレーションテストをする Stack を登録する必要があります。
コードだとこの部分
api = Api(app, "cdk-integ-tests-stack-test")
登録しかたは普通の CDK と一緒ですね。
次に以下のようにインテグレーションテスト用のリソースを作成します。
integ = IntegTest(
app,
"integ",
test_cases=[api],
cdk_command_options=CdkCommands(destroy=DestroyCommand(args=DestroyOptions(force=True))),
)
cdk_command_optionsは必須ではありません。が、今回は作ったリソースを強制削除するようなオプションをつけています。
あとはこのintegにアサーションを登録していけば完了です。
このへんは書き方の作法的な話なので「あぁこう書けばいいのね」くらいかなと思います。
そしてテストを実行します。
$ export CDK_INTEG_RUNNER_CWD=$(pwd) && npx integ-runner --directory ./integ_tests --parallel-regions ap-northeast-1
これでテストが実行されます。
結構時間がかかりますが、無事にテストが通ると ↓ のような表示になります。
ぱっとみ一番最初にテストが１つ転んでいるように見えますが、これは過去のスナップショットと比較して差分があることを示していいるので転んでいるわけではないです。
そしてその後テスト 3 つが無事に通っていることがわかります。
以上でしたー！
まとめ
API Gateway を使った Lambda に対して、CDK の alpha 版にあるインテグレーションテストを行ってみました。
Stack を deploy する時間はかかるものの、AWS の環境にゴミなど残さずきれいにテストをすることができました。また AWS の実際の環境に deploy してテストを行っているため、かなり信頼性の高いテストができそうだなと感じました。
ただし懸念点もあり、テスト開発段階では何度も AWS 環境に deploy が必要になるため通常のテスト作成よりも時間がかかります。またテスト実行にもそれなりに時間がかかります。この２つは時間的なコストにも金銭的なコストにも直に響いてくるので、導入は慎重になったほうがいいかもしれません。また、ここで頑張って作っても alpha 版なので大きくしようが変わったりするリスクもあります。
とはいえ、CDK でインテグレーションテストが完結するのはとても魅力的で、今後の動きに期待したいと思います。
今回のリポジトリはこちら
https://github.com/sisi100/cdk-integ-tests
感謝!
表紙は UnsplashのClint Adairさんが撮影した写真を使わせていただきました。 ありがとうございます！