Amazon Web Services ブログ
AWS CodeStar プロジェクトにおける単体テストの実行
このブログ記事では、AWS CodeStar プロジェクトの一部として単体テストを実行する方法についてご紹介します。AWS CodeStar は AWS 上でアプリケーションを素早く開発、ビルド、そしてデプロイするのに役立ちます。AWS CodeStar を使えば、継続的デリバリ（CD）のツールチェーンを構築しソフトウェア開発を一箇所に集約して管理することができます。
単体テストはアプリケーションのコードの個々の単位をテストし、問題を素早く特定し取り去るのに役立ちます。単体テストは、自動化された CI/CD プロセスの１部分を構成することで、バッドコードが本番環境にデプロイされてしまうことを防ぐのにも役立てられます。
多くの AWS CodeStar プロジェクト テンプレートには、単体テストのフレームワークがあらかじめ設定されています。そのため、自信を持ってコードのデプロイを開始することができるでしょう。この AWS CodeStar プロジェクトの単体テストは、ビルドステージにおいて実行されるように設定されているので、もし単体テストがパスしなかったらそのコードがデプロイされることはありません。単体テストを内包している AWS CodeStar プロジェクト テンプレートの一覧については、AWS CodeStar ユーザガイドの AWS CodeStar プロジェクト テンプレートを参照してください。
シナリオ
私はスーパーヒーロー映画の大ファンです。なので、お気に入りの作品を一覧にして友人に投票してもらうウェブサービスを作ろうと思い立ちました。私が使ったのは、AWS Lambda で実行される Python のウェブフレームワークで、ソースコード リポジトリに AWS CodeCommit を使っているテンプレートです。CodeCommit はフルマネージドのソースコントロールサービスです。 Git リポジトリをホストし Git ベースの全てのツールと連携可能です。
ウェブサービスのエンドポイントを作成する方法は次のとおりです。
AWS CodeStar コンソールにサインインし、Start a project を選択します。すると、プロジェクト テンプレートの一覧が表示されます。
コード編集の方法として AWS Cloud9 を選択します。AWS Cloud9 は、コードの記述、実行、デバッグが全て実施可能なクラウドベースの統合開発環境 (IDE) です。
その他に必要な作業は次のとおりです。
- 必要に応じて投票データを保存および取得するためのデータベース テーブルの作成。
- 投票データの保存と取得のために作成された Lambda 関数のロジックの更新。
- （当たり前ですが）上記ロジックが期待通りに動作するかを検証する単体テストの更新。
データベースには Amazon DynamoDB を選択しました。Amazon DynamoDB は高速で柔軟な NoSQL データベースを提供します。
AWS Cloud9 で設定する
AWS CodeStar のコンソールから AWS Cloud9 のコンソールに移動してください。これにより、プロジェクトのコードが表示されます。トップレベルのフォルダでターミナルを開き、環境変数と必要なライブラリをセットします。
PYTHONPATH 環境変数を設定するために、ターミナル上で次のコマンドを実行してください。
プロジェクト内で単体テストを実行できるようにするために次のコマンドを使います。
コーディングを開始する
ローカル環境をセットアップしコードのコピーを手に入れたので、テンプレートファイルで DynamoDB テーブルを定義することによって DynamoDB テーブルをプロジェクトに追加してください。Serverless Application Model (SAM) のテンプレートファイルである template.yml を開いてください。Serverless Application Model (SAM) テンプレートは AWS CloudFormation を拡張して Amazon API Gateway の API、AWS Lambda の関数、そして Amazon DynamoDB のテーブルといったサーバレスアプリケーションで必要となるリソースを定義するためのシンプルな方法を提供します。
AWSTemplateFormatVersion: 2010-09-09
Transform:
- AWS::Serverless-2016-10-31
- AWS::CodeStar
Parameters:
ProjectId:
Type: String
Description: CodeStar projectId used to associate new resources to team members
Resources:
# 投票を保存するための DB テーブル
MovieVoteTable:
Type: AWS::Serverless::SimpleTable
Properties:
PrimaryKey:
# Candidate の名前を、テーブルの partition key として使用
Name: Candidate
Type: String
# 投票を保存と取得するための Lambda 関数を新規作成
MovieVoteLambda:
Type: AWS::Serverless::Function
Properties:
Handler: index.handler
Runtime: python3.6
Environment:
# Lambda 関数で使用する環境変数を設定
Variables:
TABLE_NAME: !Ref "MovieVoteTable"
TABLE_REGION: !Ref "AWS::Region"
Role:
Fn::ImportValue:
!Join ['-', [!Ref 'ProjectId', !Ref 'AWS::Region', 'LambdaTrustRole']]
Events:
GetEvent:
Type: Api
Properties:
Path: /
Method: get
PostEvent:
Type: Api
Properties:
Path: /
Method: postAWS サービスと接続するために Python の boto3 ライブラリを使います。それから、単体テストにおける AWS サービスの呼び出しをモックするために Python のモックライブラリを使います。次のコマンドを使ってこれらのライブラリをインストールしてください。
これらのライブラリを buildspec.yml に追加してください。buildspec.yml は、CodeBuild を実行する際に必要な YAML ファイルです。
index.py を開いてください。Lambda 関数のシンプルな投票ロジックをこのファイルに書きます。
import json
import datetime
import boto3
import os
table_name = os.environ['TABLE_NAME']
table_region = os.environ['TABLE_REGION']
VOTES_TABLE = boto3.resource('dynamodb', region_name=table_region).Table(table_name)
CANDIDATES = {"A": "Black Panther", "B": "Captain America: Civil War", "C": "Guardians of the Galaxy", "D": "Thor: Ragnarok"}
def handler(event, context):
if event['httpMethod'] == 'GET':
resp = VOTES_TABLE.scan()
return {'statusCode': 200,
'body': json.dumps({item['Candidate']: int(item['Votes']) for item in resp['Items']}),
'headers': {'Content-Type': 'application/json'}}
elif event['httpMethod'] == 'POST':
try:
body = json.loads(event['body'])
except:
return {'statusCode': 400,
'body': 'Invalid input! Expecting a JSON.',
'headers': {'Content-Type': 'application/json'}}
if 'candidate' not in body:
return {'statusCode': 400,
'body': 'Missing "candidate" in request.',
'headers': {'Content-Type': 'application/json'}}
if body['candidate'] not in CANDIDATES.keys():
return {'statusCode': 400,
'body': 'You must vote for one of the following candidates - {}.'.format(get_allowed_candidates()),
'headers': {'Content-Type': 'application/json'}}
resp = VOTES_TABLE.update_item(
Key={'Candidate': CANDIDATES.get(body['candidate'])},
UpdateExpression='ADD Votes :incr',
ExpressionAttributeValues={':incr': 1},
ReturnValues='ALL_NEW'
)
return {'statusCode': 200,
'body': "{} now has {} votes".format(CANDIDATES.get(body['candidate']), resp['Attributes']['Votes']),
'headers': {'Content-Type': 'application/json'}}
def get_allowed_candidates():
l = []
for key in CANDIDATES:
l.append("'{}' for '{}'".format(key, CANDIDATES.get(key)))
return ", ".join(l)上記のコードは、HTTPS リクエストの呼び出しをイベントとして受け取っています。HTTP GET リクエストの場合、投票結果をテーブルから取得します。HTTP POST リクエストの場合は、選択した候補で投票をセットします。また、悪意のあるリクエストを除外するために POST リクエストの中の入力値をバリデーションしています。
このサンプルコードでは投票候補リストを CANDIDATES 変数に保持していますが、そうではなく JSON ファイルに格納し Python の json ライブラリを使うことも可能です。
単体テストを更新してみましょう。tests フォルダ配下で test_handler.py を開いてロジックを検証するように変更してください。
import os
# DynamoDBのモックで使われる、モックの環境変数
os.environ['TABLE_NAME'] = "MockHelloWorldTable"
os.environ['TABLE_REGION'] = "us-east-1"
# 私達独自のロジックが入ったライブラリ
import index
# Boto3'のコアライブラリ
import botocore
# JSON を扱うために
import json
# 単体テストライブラリ
import unittest
## 設定に基づいて StringIO を取得
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
## Python のモックライブラリ
from mock import patch, call
from decimal import Decimal
@patch('botocore.client.BaseClient._make_api_call')
class TestCandidateVotes(unittest.TestCase):
## HTTP GET リクエストのフローをテスト
## （モックの）テーブルから取得した投票結果付きの成功レスポンスが返ってくることが期待されます
def test_get_votes(self, boto_mock):
# テスト対象のメソッドに入力するイベント
expected_event = {'httpMethod': 'GET'}
# DynamoDB テーブル内のモックの値
items_in_db = [{'Candidate': 'Black Panther', 'Votes': Decimal('3')},
{'Candidate': 'Captain America: Civil War', 'Votes': Decimal('8')},
{'Candidate': 'Guardians of the Galaxy', 'Votes': Decimal('8')},
{'Candidate': "Thor: Ragnarok", 'Votes': Decimal('1')}
]
# DynamoDB テーブルのレスポンスのモック
expected_ddb_response = {'Items': items_in_db}
# boto を介して DynamoDB 呼び出すことによって返ってくることが期待されるレスポンスのモック
response_body = botocore.response.StreamingBody(StringIO(str(expected_ddb_response)),
len(str(expected_ddb_response)))
# モック内の期待される値をセット
boto_mock.side_effect = [expected_ddb_response]
# 実行中に、これらのパラメータを使用して DynamoDB Scan 関数が呼び出されることが期待されます
expected_calls = [call('Scan', {'TableName': os.environ['TABLE_NAME']})]
# テスト対象の関数を呼び出します
result = index.handler(expected_event, {})
# 期待される関数呼び出しがモックに対して発生していること、そしてレスポンスが正しいことを検証するために、単体テストのアサーションを実行
assert result.get('headers').get('Content-Type') == 'application/json'
assert result.get('statusCode') == 200
result_body = json.loads(result.get('body'))
# 関数の呼び出し結果がテーブルから取り出した結果と一致するかを検証
assert len(result_body) == len(items_in_db)
for i in range(len(result_body)):
assert result_body.get(items_in_db[i].get("Candidate")) == int(items_in_db[i].get("Votes"))
assert boto_mock.call_count == 1
boto_mock.assert_has_calls(expected_calls)
## 選択した候補に投票を行う HTTP POST リクエストのフローをテスト
## 確認メッセージ付きの成功レスポンスが返ってくることが期待されます
def test_place_valid_candidate_vote(self, boto_mock):
# テスト対象のメソッドに入力するイベント
expected_event = {'httpMethod': 'POST', 'body': "{\"candidate\": \"D\"}"}
# DynamoDB テーブルのレスポンスのモック
expected_ddb_response = {'Attributes': {'Candidate': "Thor: Ragnarok", 'Votes': Decimal('2')}}
# boto を介して DynamoDBを呼び出すことによって返ってくることが期待されるレスポンスのモック
response_body = botocore.response.StreamingBody(StringIO(str(expected_ddb_response)),
len(str(expected_ddb_response)))
# モック内の期待される値をセット
boto_mock.side_effect = [expected_ddb_response]
# 実行中に、これらのパラメータを使用して DynamoDB UpdateItem 関数が呼び出されることが期待されます
expected_calls = [call('UpdateItem', {
'TableName': os.environ['TABLE_NAME'],
'Key': {'Candidate': 'Thor: Ragnarok'},
'UpdateExpression': 'ADD Votes :incr',
'ExpressionAttributeValues': {':incr': 1},
'ReturnValues': 'ALL_NEW'
})]
# テスト対象の関数を呼び出します
result = index.handler(expected_event, {})
# 期待される関数呼び出しがモックに対して発生していること、そしてレスポンスが正しいことを検証するために、単体テストのアサーションを実行
assert result.get('headers').get('Content-Type') == 'application/json'
assert result.get('statusCode') == 200
assert result.get('body') == "{} now has {} votes".format(
expected_ddb_response['Attributes']['Candidate'],
expected_ddb_response['Attributes']['Votes'])
assert boto_mock.call_count == 1
boto_mock.assert_has_calls(expected_calls)
## 存在しない候補に投票を行う HTTP POST リクエストのフローをテスト
## 確認メッセージ付きの成功レスポンスが返ってくることが期待されます
def test_place_invalid_candidate_vote(self, boto_mock):
# テスト対象のメソッドに入力するイベント
# 投票候補として正しい ID は A, B, C, そして D です
expected_event = {'httpMethod': 'POST', 'body': "{\"candidate\": \"E\"}"}
# テスト対象の関数を呼び出します
result = index.handler(expected_event, {})
# 期待される関数呼び出しがモックに対して発生していること、そしてレスポンスが正しいことを検証するために、単体テストのアサーションを実行
assert result.get('headers').get('Content-Type') == 'application/json'
assert result.get('statusCode') == 400
assert result.get('body') == 'You must vote for one of the following candidates - {}.'.format(index.get_allowed_candidates())
## 選択した候補に投票を行う HTTP POST リクエストのフローをテスト。ただし、選択した候補は POST のボディに不正なキーで紐付けられています。
## 適切なエラーメッセージをともなった失敗（400）レスポンスが返ってくることが期待されます
def test_place_invalid_data_vote(self, boto_mock):
# テスト対象のメソッドに入力するイベント
# "name" は期待される入力キーではありません
expected_event = {'httpMethod': 'POST', 'body': "{\"name\": \"D\"}"}
# テスト対象の関数を呼び出します
result = index.handler(expected_event, {})
# 期待される関数呼び出しがモックに対して発生していること、そしてレスポンスが正しいことを検証するために、単体テストのアサーションを実行
assert result.get('headers').get('Content-Type') == 'application/json'
assert result.get('statusCode') == 400
assert result.get('body') == 'Missing "candidate" in request.'
## 選択した候補に投票を行う HTTP POST リクエストのフローをテスト。ただし、リクエスト ボディは期待されている JSON 文字列形式ではありません。
## 適切なエラーメッセージをともなった失敗（400）レスポンスが返ってくることが期待されます
def test_place_malformed_json_vote(self, boto_mock):
# テスト対象のメソッドに入力するイベント
# "body" が JSON ではなく単なる文字列です
expected_event = {'httpMethod': 'POST', 'body': "Thor: Ragnarok"}
# テスト対象の関数を呼び出します
result = index.handler(expected_event, {})
# 期待される関数呼び出しがモックに対して発生していること、そしてレスポンスが正しいことを検証するために、単体テストのアサーションを実行
assert result.get('headers').get('Content-Type') == 'application/json'
assert result.get('statusCode') == 400
assert result.get('body') == 'Invalid input! Expecting a JSON.'
if __name__ == '__main__':
unittest.main()サンプルコードにはコメントを丁寧に残しているので、各単体テストが何をやっているかは分かりやすいと思います。成功条件をテストするとともに、ロジック内で正常に取り扱われる失敗のパスもテストしています。
この単体テストでは、モックライブラリの patch デコレータ (@patch) を使用しています。@patch は、呼び出したい関数をモックするのに役立ちます（今回の場合で言えば、botocore ライブラリの BaseClient クラスの _make_api_call 関数です）。
変更を commit する前に、ローカルでテストを動かしてみましょう。ターミナルで、再度テストを実行してください。もし、全ての単体テストがパスしたらこんな感じの結果を目にすると思われます。
AWS にアップロードする
テストが通ったということは、いよいよコードを commit してリポジトリに push するときがきたということです。
変更を add する
ターミナルでプロジェクトのフォルダに移動し、push するつもりの変更内容を確認するために次のコマンドを使用してください。
変更したファイルのみを add するため、次のコマンドを使用してください。
（訳注： git add の -u オプションは、新規追加されたファイルは除いて、変更のあった部分のみを add します）
変更を commit する
（コミットメッセージを付けて）変更を commit するために、次のコマンドを使用してください。
AWS CodeCommit に変更を push する
commit した変更を CodeCommit に push するために、次のコマンドを使用してください。
AWS CodeStar のコンソールで、変更内容がパイプラインを経由してデプロイされるのを確認することができます。また、AWS CodeStar のコンソールには、このプロジェクトの「ビルド実行」に遷移するリンクがあるので、 AWS CodeBuild 上でテストが実行されるのを見ることができます。Build Runs テーブルの下の最新のリンクをクリックするとログが表示されます。
デプロイが完了すると、AWS Lambda 関数と DynamoDB テーブルが作成されるとともにこのプロジェクトに同期され、 AWS CodeStar で表示されるようになります。AWS CodeStar のナビーションバーにある Project というリンクをクリックすると、このプロジェクトに紐付けられた AWS リソースが表示されます。
新しいデータベーステーブルなので中身は空っぽです。なので何度か投票してみましょう。Postman をダウンロードすれば、アプリケーション エンドポイントの POST と GET の呼び出しをテストすることができます。テストするエンドポイントは AWS CodeStar コンソールの Application endpoints の下に表示されている URL です。
Postman を開いて結果を見てみましょう。POST リクエストで何度か投票を行って見ましょう。今回の例では、正しい投票は A, B, C, または D のうちどれかの値を持ちます。正しい POST リクエストはこんなふうになります。
A, B, C, あるいは D 以外のなんらかの値を使った場合にどのようになるかがこちらです。
それでは、データベースから投票結果を取得するために GET リクエストを利用したいと思います。
以上で終了です。AWS Lambda, Amazon API Gateway, そして DynamoDB を使ってシンプルな投票ウェブサービスを作成しました。そして、ロジックを検証する単体テストを導入しました。そのおかげでグッドコードを ship することができましたね！
Happy coding!
翻訳は SA 畑史彦が担当しました。原文は AWS DevOps Blog の『Performing Unit Testing in an AWS CodeStar Project』です。