API Gateway (REST) + LamndaをServerless Frameworkで構築し、pytestを使ってテストしてみる
データアナリティクス事業本部のueharaです。
今回は、API Gateway (REST) + LamndaをServerless Frameworkで構築し、pytestを使ってテストしてみたいと思います。
前提条件
今回はServerless Frameworkを利用してAWS環境にデプロイを行います。
まだ準備がおすみでない方は、以下を参考にインストールください。
また、Serverless Frameworkのserverless-python-requirementsプラグインも利用します。
Serverless Frameworkのインストール後、以下でインストールしてください。
$ sls plugin install -n serverless-python-requirements
構成図
今回作成する構成は、簡単ですが次の通りです。
実装
今回、デプロイのために用意するフォルダ構成は以下の通りです。
. ├ handler │ └ sample.py ├ requirements.txt └ serverless.yaml
Lamnda (Python) の実装
API GWの後ろにあるLamnda関数の実装を行います。
今回は非常にシンプルに、ステータスコード200で"Test Response"というレスポンスを返す処理を記載します。
また、今回は使用しませんが、requirements.txtによるLambda Layer構築のテストもしたいので、redshift_connectorをインポートする処理を記載しておきます。
import json
import logging
# テスト用
import redshift_connector
logger = logging.getLogger()
level = logging.getLevelName("INFO")
logger.setLevel(level)
def lambda_handler(event, context):
logger.info("event: {}".format(json.dumps(event)))
logger.info("redshift_connector version: {}".format(redshift_connector.__version__))
response = {
"statusCode": 200,
"body": json.dumps("Test Response")
}
return response
requirements.txt
先にも記載した通り、今回使用する想定はありませんがテスト用にredshift_connectorを記載します。
redshift_connector
serverless.yaml
今回本題となるserverless.yamlです。
公式ドキュメントやこちらのブログを参考に、ざっと以下のように記載してみました。
service: my-apigw-test # Cloudformationのstack nameを設定
frameworkVersion: '3'
provider:
name: aws
stage: dev
runtime: python3.9
lambdaHashingVersion: 20201221
region: ap-northeast-1
endpointType: REGIONAL # API Gateway REST APIのエンドポイントタイプ: edgeまたはregional (デフォルト: edge)
apiGateway:
resourcePolicy:
- Effect: Allow
Principal: '*'
Action: execute-api:Invoke
Resource:
- execute-api:/*/*/*
apiKeys:
- free: # 使用プラン
- name: ${self:service}-key # key名
value: (YOUR_API_KEY) # 30-128文字の任意の英数字
usagePlan:
- free:
quota:
limit: 100 # APIの呼び出しを行える最大回数
offset: 0 # APIの呼び出し回数の初期値（通常は0回を指定する）
period: DAY # DAY or WEEK or MONTH
throttle:
rateLimit: 2 # 1秒あたりに処理できる API リクエスト数
burstLimit: 3 # 同時に処理できる最大リクエスト数
functions:
redshift_select:
name: ${self:service}-handler # lambda関数名
handler: handler/sample.lambda_handler # 実行される関数を指定
role: LambdaRole # lambdaに紐づけられるロール
memorySize: 128 # lambdaのメモリサイズ
timeout: 30 # lambdaのタイムアウト時間
layers: # lambdaに紐づくレイヤーを指定
- Ref: PythonRequirementsLambdaLayer # Layerを参照
events: # lambda関数のトリガーを指定
- http:
path: /sample # このendpointのパス
method: post
private: true # リクエストの`x-api-key`ヘッダーにAPIキー値を追加することを要求
custom:
accountid: ${AWS::AccountId}
pythonRequirements:
dockerizePip: false # python以外で作られているライブラリを使用する時はtrueに
usePipenv: false # Pipenvを使用する場合にtrueに
layer: true # ライブラリからLambda Layerを作成するオプション
useDownloadCache: true # pipがパッケージをコンパイルするために必要なダウンロードをキャッシュするダウンロードキャッシュ
useStaticCache: true # requirements.txtのすべてをコンパイルした後にpipの出力をキャッシュする静的キャッシュ
plugins:
- serverless-python-requirements
resources:
Description: Test role for Lamnda
Resources:
LambdaRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: "2012-10-17"
Statement:
- Action:
- sts:AssumeRole
Effect: Allow
Principal:
Service:
- "lambda.amazonaws.com"
ManagedPolicyArns:
- arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
package:
patterns:
- "!.vscode/**"
- "!.git/**"
- "!.gitignore"
- "!.serverless"
- "!.serverless/**"
- "!README.md"
- "!package*.json"
- "!requirements.txt"
- "!node_modules/**"
- "!__pycache__"
- "!yarn.lock"
上記で設定していることを簡単に説明すると、/sampleのエンドポイントに対してpostリクエストがあると、handler/sample.pyで設定されるLambda関数を起動するようになっています。
なお、private: trueに設定していることにより、APIキーの指定が無いリクエストはステータスコード403を返すようになります。
(YOUR_API_KEY)という部分には、ご自身がAPIキーとして利用したい30-128文字の任意の英数字を記載してください。
その他、functionsプロパティでは作成するLamnda関数の設定、customeプロパティでは作成するLambda Layerに関する設定を行っています。
デプロイ
serverless.yamlがあるフォルダ上で以下のコマンドを実行します。
$ AWS_SDK_LOAD_CONFIG=true AWS_PROFILE=(AWS環境にアクセスするProfile) sls deploy
上記コマンド実行後、AWSマネジメントコンソールからCloudFormationにアクセスし、serverless.yamlで指定したサービス名のスタックが作成されていれば成功です。
同じくAWSマネジメントコンソールからAPI Gatewayにアクセスし、作成したAPIのステージを確認すると、エンドポイントポイントを確認することができます。
pytestによるテスト
まず、今回テストに必要なPythonモジュールをインストールします。
$pip install requests pytest
インストールが完了したら、次のテストファイルを作成します。
import requests
# 正常ケース
def test_endpoint_accept():
endpoint = "https://(restapi_id).execute-api.ap-northeast-1.amazonaws.com/dev/sample"
headers = {"x-api-key": "(YOUR_API_KEY)"}
res = requests.post(endpoint, headers=headers)
assert res.status_code == 200
assert res.text == '"Test Response"'
# ヘッダでAPIキーを指定しないケース
def test_endpoint_reject():
endpoint = "https://(restapi_id).execute-api.ap-northeast-1.amazonaws.com/dev/sample"
res = requests.post(endpoint)
assert res.status_code == 403
上記では、正常ケースとヘッダでAPIキーが指定されていないケースをテストします。
具体的には、正常ケースではステータスコード200とレスポンスBodyが想定したものになっていること、ヘッダでAPIキーを指定しないケースではステータスコード403が返るか確認しています。
endpointには、先程マネジメントコンソール上で確認したご自身のエンドポイントを入力して下さい。
また、ヘッダのx-api-keyにはご自身で指定したAPIキーを指定して下さい。（マネジメントコンソールからも確認可能です。）
テスト実行
テストの実行は以下コマンドで可能です。
$ pytest -v test_apigw.py
以下の通り、2つのテスト共にPASSしていれば期待通り動作しています。
Lambdaの実行ログを確認すると、今回テスト目的で実施したredshift_connectorのインポートもきちんとできていることが分かります。
最後に
今回は、API Gateway (REST) + LamndaをServerless Frameworkで構築し、pytestを使ってテストしてみました。
参考になりましたら幸いです。
