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をインポートする処理を記載しておきます。

sample.py

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を記載します。

requirements.txt

redshift_connector

serverless.yaml

今回本題となる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

インストールが完了したら、次のテストファイルを作成します。

test_apigw.py

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を使ってテストしてみました。

参考になりましたら幸いです。

参考文献

• Serverless Framework Documentation
• API Gateway + REST + Lambda 非プロキシ(カスタム)統合をServerless Frameworkで実装してみた
• pytest: helps you write better programs