本記事は
基盤デザインウィーク
4日目の記事です。
🌈
3日目
▶▶ 本記事 ▶▶
5日目
💻
はじめに
はじめまして。基デザウィーク4日目を担当します、入社1年目の深瀬です。昨年8月に基盤デザイン事業部へ配属となり、インフラエンジニアの道を進むことになりました。最初は「インフラ!?うっ…頭が痛い。。」と思っていましたが、いざ配属されるとなんとなく思い描いていたのとは違い、今のところ楽しくインフラエンジニアライフを過ごせています。
本記事では、Amazon API Gateway（以下API Gateway）のテストをBoto3で行った体験記を綴っていきたいと思います。わたしは大のテスト苦手人間なので、読み返したときに思い出せるような、わかりやすい記事になっていれば幸いです。
pytest・Boto3・API Gatewayについて
まずはじめに、pytestやBoto3、API Gatewayについてざっくりと説明します。
pytestとは
pytestはPythonのテストフレームワークです。今回はpytestのAssertionを使用します。 Assertionは、特定の条件が満たされていることをチェックし、Trueを返すことでテストがPASSされる仕組みとなっています。
Boto3とは
次にBoto3とは、AWS SDK for Pythonの別称で、AWSの使用を迅速に開始できるライブラリです。 Boto3を使用することで、AWSの各種サービスをPythonで操作できるようになります。 ちなみに、Boto3の名前の由来はアマゾンカワイルカらしいです。 botoはアマゾンカワイルカの現地での呼ばれ方だそうで、短く、一風変わっており、アマゾンに何らかの関係がある単語という条件に見事合致したそうです。 開発者のネーミングセンスが光りまくっていて、なんだか羨ましいです。
API Gatewayとは
最後にAPI Gatewayとは、APIの作成、公開、保守、モニタリング、保護を行うことができるAWSのフルマネージド型サービスです。 API Gatewayを使用することで、Restful API（HTTP API, REST API）やWebSocket APIを作成することができます。 今回わたしはMockを作成したため、REST APIを使用しました。
やってみた
それでは、どんなAPI Gatewayを作成し、何をテストしたのかをまとめていきます。
API GatewayでMock作成
まず、API Gatewayに関してはAWS CDK for Pythonで、/hoge/fugaにGETリクエストを送るとJSON形式のレスポンス内容が返ってくるシンプルなMockを作成しました。
from aws_cdk import ( Stack, aws_apigateway as apigateway, ) from constructs import Construct import json class HogehogeStack(Stack): def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None: super().__init__(scope, construct_id, **kwargs) """ API Gateway作成 """ apigw = apigateway.RestApi( self, id="hogehoge", rest_api_name="hogehoge", endpoint_types=[apigateway.EndpointType.REGIONAL], ) """ リソース作成 """ hoge = apigw.root.add_resource("hoge") fuga = hoge.add_resource("fuga") """ レスポンス内容 """ fuga_responce_data = json.dumps( { "hoge": "fuga", } ) """ Mock（メソッド）作成 """ fuga.add_method( "GET", apigateway.MockIntegration( integration_responses=[ apigateway.IntegrationResponse( status_code="200", response_templates={ "application/json": fuga_responce_data, }, ), ], request_templates={ "application/json": '{ "statusCode": 200 }', }, ), method_responses=[apigateway.MethodResponse(status_code="200")], )
作成したMockのテストをBoto3で実施
次に、作成したMockのテストコードを作成しました。今回行ったのは、リソースパスが正しく作成されているか、レスポンス内容は正しいかといったテストです。テスト作成はBoto3公式ドキュメントを参考に進めていきました。
こちらが作成したテストコードの全体像です。
import json from boto3.session import Session # API Gatewayへのセッション def get_apigateway_client(): session = Session() apigateway_client = session.client( service_name="apigateway", region_name="ap-northeast-1", ) return apigateway_client # 期待するリソースパスのリスト expectation_resource_path_list = [ "/", "/hoge/fuga", "/hoge", ] def test_apigateway_resource_path(): """ Test: API Gatewayテスト Item: リソースパス確認 Expect: リソースパスが正しく設定されていること """ apigateway_client = get_apigateway_client() rest_api_id = apigateway_client.get_rest_apis()["items"][0]["id"] apigateway_resources = apigateway_client.get_resources( restApiId=rest_api_id, ) # リソースパスを格納するための空のリストを用意する apigateway_resource_path_list = [] apigateway_resource_items = apigateway_resources["items"] for i in apigateway_resource_items: apigateway_resource_path_list.append(i["path"]) assert apigateway_resource_path_list == expectation_resource_path_list # メソッドがあるリソースパスのリスト expectation_method_resource_path_list = [ "/hoge/fuga", ] # 期待するレスポンス内容のリスト expectation_response_list = [ { "hoge": "fuga", }, ] def test_apigateway_response(): """ Test: API Gatewayテスト Item: レスポンス内容確認 Expect: レスポンス内容が正しく設定されていること """ apigateway_client = get_apigateway_client() rest_api_id = apigateway_client.get_rest_apis()["items"][0]["id"] apigateway_resources = apigateway_client.get_resources( restApiId=rest_api_id, ) # リソースidとリソースパスを格納するための空の多次元リストを用意する apigateway_resource_id_path_list = [] for i in apigateway_resources["items"]: apigateway_resource_id_path_list.append([i["id"], i["path"]]) # レスポンスを格納するための空のリストを用意する apigateway_response_list = [] for resource_id, resource_path in apigateway_resource_id_path_list: for expectation_resource_path in expectation_method_resource_path_list: # apigateway_response_listに格納されているpathが期待するpathと合致する場合 if resource_path == expectation_resource_path: apigateway_response = apigateway_client.get_integration_response( restApiId=rest_api_id, resourceId=resource_id, httpMethod="GET", statusCode="200", ) apigateway_response_list.append( json.loads( apigateway_response["responseTemplates"]["application/json"] ) ) continue assert apigateway_response_list == expectation_response_list
細かく解説していきます。
作成したAPI Gatewayへのセッションはこちらのコードで行っています。
# API Gatewayへのセッション def get_apigateway_client(): session = Session() apigateway_client = session.client( service_name="apigateway", region_name="ap-northeast-1", ) return apigateway_client
service_nameはapigateway、region_nameは東京リージョンにデプロイしたのでap-northeast-1になっています。
apigateway_client = get_apigateway_client()
各テストコードの先頭で、作成したget_apigateway_client()を使用し、API Gatewayと接続しています。
リソースパスのテスト
リソースパスのテストコードでは、get_rest_apis()でREST API IDを取得し、その取得したREST API IDを使ってget_resources()でリソースを取得しています。
rest_api_id = apigateway_client.get_rest_apis()["items"][0]["id"] apigateway_resources = apigateway_client.get_resources( restApiId=rest_api_id, )
ここで、get_rest_apis()["items"][0]["id"]としているのは、itemsという配列の0番目のidを取得するためです。下記コードは公式ドキュメントに載っているget_rest_apis()の返り値です。
{ 'position': 'string', 'items': [ { 'id': 'string', 'name': 'string', 'description': 'string', 'createdDate': datetime(2015, 1, 1), 'version': 'string', 'warnings': [ 'string', ], 'binaryMediaTypes': [ 'string', ], 'minimumCompressionSize': 123, 'apiKeySource': 'HEADER'|'AUTHORIZER', 'endpointConfiguration': { 'types': [ 'REGIONAL'|'EDGE'|'PRIVATE', ], 'vpcEndpointIds': [ 'string', ] }, 'policy': 'string', 'tags': { 'string': 'string' }, 'disableExecuteApiEndpoint': True|False, 'rootResourceId': 'string' }, ] }
リソース名を取得したい場合はget_rest_apis()["items"][0]["name"]となるということですね。
また、こちらのコードでは空のリストapigateway_resource_path_listに取得したリソースパスを格納していき、あらかじめ用意しておいた期待するリソースパスのリストexpectation_resource_path_listと比較しています。
# リソースパスを格納するための空のリストを用意する apigateway_resource_path_list = [] apigateway_resource_items = apigateway_resources["items"] for i in apigateway_resource_items: apigateway_resource_path_list.append(i["path"]) assert apigateway_resource_path_list == expectation_resource_path_list
apigateway_resourcesには先ほどget_resources()で獲得したリソースが代入されています。get_resources()の返り値には、求めているリソースパスpathが含まれているため、for文で取得しています。i["path"]はapigateway_resources["items"][i]["path"]という意味です。
テストの結果がこちらです。
(.venv) root@add59a1b2c52:~/cdk_study# pytest -v tests/integration -vvs -k test_apigateway_resource_path ============================================================== test session starts =============================================================== platform linux -- Python 3.10.13, pytest-7.4.4, pluggy-1.3.0 -- /root/cdk_study/.venv/bin/python3 cachedir: .pytest_cache rootdir: /root/cdk_study plugins: typeguard-2.13.3 collected 2 items / 1 deselected / 1 selected tests/integration/test_hogehoge_stack.py::test_apigateway_resource_path PASSED ======================================================== 1 passed, 1 deselected in 0.53s =========================================================
無事PASSすることができました。
レスポンス内容のテスト
レスポンス内容のテストコードでも、get_rest_apis()でREST API IDを取得し、その取得したREST API IDを使ってget_resources()でリソースを取得しています。今回は空の多次元リストapigateway_resource_id_path_listに取得したリソースIDとリソースパスを格納していきます。
rest_api_id = apigateway_client.get_rest_apis()["items"][0]["id"]
apigateway_resources = apigateway_client.get_resources(
restApiId=rest_api_id,
)
# リソースidとリソースパスを格納するための空の多次元リストを用意する
apigateway_resource_id_path_list = []
for i in apigateway_resources["items"]:
apigateway_resource_id_path_list.append([i["id"], i["path"]])
取得したリソースパスが期待するリソースパスと同じ場合、取得したREST API IDやリソースIDを用いてget_integration_response()でレスポンスの詳細を取得します。
# レスポンスを格納するための空のリストを用意する apigateway_response_list = [] for resource_id, resource_path in apigateway_resource_id_path_list: for expectation_resource_path in expectation_method_resource_path_list: # apigateway_response_listに格納されているpathが期待するpathと合致する場合 if resource_path == expectation_resource_path: apigateway_response = apigateway_client.get_integration_response( restApiId=rest_api_id, resourceId=resource_id, httpMethod="GET", statusCode="200", ) apigateway_response_list.append( json.loads( apigateway_response["responseTemplates"]["application/json"] ) ) continue assert apigateway_response_list == expectation_response_list
ここで、apigateway_response["responseTemplates"]["application/json"]によってレスポンス内容を抽出するのですが、JSON形式のレスポンスとなっているため、json.loads()でJSON文字列からPythonオブジェクトに変換し、空のリストexpectation_response_listに格納しています。
テストの結果がこちらです。
(.venv) root@add59a1b2c52:~/cdk_study# pytest -v tests/integration -vvs -k test_apigateway_response ===================================================================== test session starts ====================================================================== platform linux -- Python 3.10.13, pytest-7.4.4, pluggy-1.3.0 -- /root/cdk_study/.venv/bin/python3 cachedir: .pytest_cache rootdir: /root/cdk_study plugins: typeguard-2.13.3 collected 2 items / 1 deselected / 1 selected tests/integration/test_hogehoge_stack.py::test_apigateway_response PASSED =============================================================== 1 passed, 1 deselected in 1.03s ================================================================
こちらも無事PASSすることができました。
おわりに
今回はBoto3を用いてAPI Gatewayのテストを行いました。わたしが以前業務で行ったAPI Gatewayの構築とテストをこのブログ用に再実施したのですが、所々で躓き、まだまだ実力不足であることを痛感しました。今回は特にprint()にありがたみを感じました。リストの中身をprint()で確認できるため、テストをPASSさせる過程でとても助かりました。このテストコードにはまだまだ改善の余地があると思いますが、一旦は良かったです。
また、今まではテストに対して苦手意識があったのですが、それはどう書けばいいかわからないことが最大の原因だったと思います。今回はBoto3を用いてテストを作成しましたが、公式ドキュメントを確認しながら作成したことで理解が深まり、苦手意識が少し薄まったように思います。
それでは、拙い文章ではありましたが、最後まで読んでいただきありがとうございました。
