モックが無くてLambdaのテストが辛い?!無ければ自分で作ればいい!! Greengrassのユニットテストが辛かったのでmotoを拡張してGreengrassのモックを実装してみた

Greengrass周りのユニットテストが辛かったので、motoを拡張してGreengrassのモックを実装してみました。
2019.04.23

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

サーバーレス開発部@大阪の岩田です。 最近Greengrassを活用したシステムのバックエンド開発をやっていのですが、ローカル環境でのユニットテストがとても辛いです。いや、辛かったです。 DynamoDBのようなメジャーサービスであればDynamoDB LocalやLocalStackを利用することで、比較的簡単にユニットテストを回していくことができますが、GreengrassにはDynamoDB Localのようなモックツールが存在せず、LocalStackもGreengrassには未対応となっています。

解決策としてmotoを拡張してGreengrassのモックを作成したので手順についてご紹介します。 以前紹介した下記ブログでは既にmoto本体に実装済みのDynamoDBのモックにTransactWriteItemsの機能を追加しましたが、今回はmoto本体にモックが実装されていないサービスへのモックの実装となります。

AWSのサービスをモックするライブラリmotoを拡張してDynamoDBのTransactWriteItemsを実装する

なおソースコードはGitHubで公開しています。

環境

以下の環境で開発中のプロジェクトにGreengrassのモックを追加しました。

  • OS: Mac OS X 10.14.2
  • Python: 3.6.5
  • boto3: 1.9.71
  • botocore: 1.12.71
  • moto: 1.3.7

motoを拡張する手順

今回はtestsディレクトリにmotoというディレクトリを作成し、さらにその中のgreengrassというディレクトリに諸々の処理を実装しました。

/tests/
├── moto
│   └── greengrass
│       ├── __init__.py
│       ├── exceptions.py
│       ├── models.py
│       ├── responses.py
│       └── urls.py

ざっくりと各ファイルの役割は下記の通りです。

  • __init__.py モック用のモジュールを初期化するファイル
  • exceptions.py Greengrassの独自例外を定義するファイル
  • models.py モックのメインロジックを実装するファイル
  • responses.py boto3から受け付けたAPIリクエストを受け付けてmodels.pyの適切な処理に振り分けるファイル
  • urls.py GreengrassのAPIリクエスト先のURLを定義するファイル

各ファイルの詳細を見ていきながら、boto3のGreengrass Clientにcreate_groupのモックを実装する手順を解説します。

init.py

__init__.py

from __future__ import unicode_literals
from .models import greengrass_backends
from moto.core.models import base_decorator


greengrass_backend = greengrass_backends['us-east-1']
mock_greengrass = base_decorator(greengrass_backends)

このファイルでやることはシンプルです。mock_greengrassという名前でGreengrassバックエンドのモックを準備します。 これでgreengrassモジュールをimportしたファイルからはmock_greengrassという名前でモックにアクセスできるようになります。

urls.py

from __future__ import unicode_literals
from .responses import GreengrassResponse

url_bases = [
    "https?://greengrass.(.+).amazonaws.com",
]

response = GreengrassResponse()

url_paths = {
    '{0}/.*$': response.dispatch,
}

このファイルではGreengrassAPIのエンドポイントURLを定義します。エンドポイントの一覧はGreengrassのAPI Referenceに記載されています。が、Greengrassのエンドポイントはシンプルなので単にurl_baseshttps?://greengrass.(.+).amazonaws.comを入れておけばOKです。

ここで定義したURLに対してboto3がリクエストを発行した際にmotoのBaseResponseで定義されたdispatchメソッドに処理が渡り、処理がモックに置き換わるという流れです。dispatchメソッドは後述するresponses.pyで定義した処理にリクエストをルーティングします。

exceptions.py

このファイルではboto3のClientErrorのエラーコード毎に独自例外を定義します。 boto3では例外をClientErrorというクラスで表現しています。例えばこんな感じです。

>>> client.create_group(Name='')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/xxxxxx/site-packages/botocore/client.py", line 357, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/Users/xxxxxx/site-packages/botocore/client.py", line 661, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidContainerDefinition) when calling the CreateGroup operation: InvalidContainerDefinitionException: Group name is missing in the input

このClientErrorクラスはresponseというプロパティを持っており、このプロパティの中でさらに詳細なエラーメッセージやエラーコードを表現しています。responseの中身はこんな感じで、先ほど発生させた例外であればエラーコードはInvalidContainerDefinitionとなります。

{'Error': {'Message': 'InvalidContainerDefinitionException: Group name is missing in the input', 'Code': 'InvalidContainerDefinition'}, 'ResponseMetadata': {'RequestId': '5fb1d2f6-56a6-11e9-b19b-bb2f7987dd15', 'HTTPStatusCode': 400, 'HTTPHeaders': {'date': 'Thu, 04 Apr 2019 06:53:38 GMT', 'content-type': 'application/json', 'content-length': '132', 'connection': 'keep-alive', 'x-amzn-requestid': '5fb1d2f6-56a6-11e9-b19b-bb2f7987dd15', 'x-amzn-greengrass-trace-id': 'Root=1-5ca5a9f2-4eb3d60f600f734096d4be3f', 'x-amz-apigw-id': 'Xmd91HzptjMFfHw=', 'x-amzn-trace-id': 'Root=1-5ca5a9f2-995be03e61145afc4c2569b2;Sampled=0'}, 'RetryAttempts': 0}}

エラーコード:InvalidContainerDefinitionExceptionを表現する場合はこんな感じです。

from __future__ import unicode_literals
from moto.core.exceptions import JsonRESTError


class GreengrassClientError(JsonRESTError):
    code = 400


class InvalidContainerDefinitionException(GreengrassClientError):
    def __init__(self, msg):
        self.code = 400
        super(InvalidContainerDefinitionException, self).__init__(
            "InvalidContainerDefinitionException",
            msg
        )

まずmoto.core.exceptions.JsonRESTErrorを継承したGreengrassClientErrorを定義し、さらにGreengrassClientErrorを継承したInvalidContainerDefinitionExceptionを定義します。 例外クラスのコンストラクタではcodeにHTTPのステータスコードを設定します。大体400か404です。設定すべきステータスコードは実際にboto3で例外を発生させて確認して下さい。

responses.py

このファイルではboto3から発行されたリクエストを自作したモックのバックエンドに引き渡す中継処理を実装します。

responses.py

from __future__ import unicode_literals

import json

from moto.core.responses import BaseResponse
from tests.moto.greengrass.models import greengrass_backends


class GreengrassResponse(BaseResponse):
    SERVICE_NAME = "greengrass"

    @property
    def greengrass_backend(self):
        return greengrass_backends[self.region]

    def create_group(self):
        name = self._get_param("Name")
        initial_version = self._get_param("InitialVersion")
        res = self.greengrass_backend.create_group(name=name, initial_version=initial_version)
        return 201, {"status": 201}, json.dumps(res.to_dict())

まずBaseResponseクラスを継承した独自のResponseクラスを作成します。 作成したらSERVICE_NAMEという変数にAWSのサービス名を設定します。今回の例だとgreengrassです。 次にgreengrass_backendというプロパティを定義し、 ここまでできたら後はboto3クライアントの各メソッドに対応したメソッドを定義していきます。今回の例ではcreate_groupを使います。この処理の中では

  • BaseResponseクラスの_get_paramメソッドを利用してboto3の各メソッドに渡された名前付き引数の値を取得
  • モックのバックエンド処理実行
  • レスポンスの返却

といった操作を行います。

models.pyの追加

このファイルにモックのメイン処理を実装していきます。

Fakexxxxxクラスの追加

対象AWSサービスが管理するエンティティを表現するためにFakexxxxxというクラスを自作します。 今回はcreate_groupを実装してGreengrassのGroupを作成するので、GreengrassのGroupを表現するエンティティを作成します。

models.py

class FakeGroup(BaseModel):
    def __init__(self, region_name, name):
        self.region_name = region_name
        self.group_id = str(uuid.uuid4())
        self.name = name
        self.arn = "arn:aws:greengrass:%s:1:/greengrass/groups/%s" % (self.region_name, self.group_id)
        self.created_at_datetime = datetime.utcnow()
        self.last_updated_datetime = datetime.utcnow()
        self.latest_version = ''
        self.latest_version_arn = ''

    def to_dict(self):
        res = {
            "Arn": self.arn,
            "CreationTimestamp": self.created_at_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
            "Id": self.group_id,
            "LastUpdatedTimestamp": self.last_updated_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
            "Name": self.name,
        }

        if self.latest_version:
            res["LatestVersion"] = self.latest_version
            res["LatestVersionArn"] = self.latest_version_arn

        return res


class FakeGroupVersion(BaseModel):
    def __init__(self, region_name, group_id, definition):
        self.region_name = region_name
        self.group_id = group_id
        self.version = str(uuid.uuid4())
        self.arn = "arn:aws:greengrass:%s:1:/greengrass/groups/%s/versions/%s" \
                   % (self.region_name, self.group_id, self.version)
        self.created_at_datetime = datetime.utcnow()
        self.definition = definition

    def to_dict(self):

        return {
            "Arn": self.arn,
            "CreationTimestamp": self.created_at_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
            "Definition": self.definition,
            "Id": self.group_id,
            "Version": self.version

        }

Fakexxxxxクラスには

  • コンストラクタ
  • APIのモックレスポンス返却用に各メンバ変数を辞書形式で返却する処理

を実装します。

GreengrassBackendクラスの実装

ここまでできたら、モックのメインを作成します。 moto.core.BaseBackendを継承してGreengrassBackendというクラスを作成し、諸々の処理を実装していきます。

models.py

class GreengrassBackend(BaseBackend):
    def __init__(self, region_name):
        super(GreengrassBackend, self).__init__()
        self.region_name = region_name
        self.groups = OrderedDict()
        self.group_versions = OrderedDict()

    def reset(self):
        region_name = self.region_name
        self.__dict__ = {}
        self.__init__(region_name)

    def create_group(self, name, initial_version):

        if name == "":
            raise InvalidContainerDefinitionException(
                "Group name is missing in the input"
            )

        group = FakeGroup(self.region_name, name)
        self.groups[group.group_id] = group

        if initial_version:
            self.create_group_version(group.group_id, initial_version)

        return group

    def create_group_version(self, group_id, definition):
        # TODO Implement validation
        group_ver = FakeGroupVersion(self.region_name, group_id, definition)
        group_vers = self.group_versions.get(group_ver.group_id, {})
        group_vers[group_ver.version] = group_ver
        self.group_versions[group_ver.group_id] = group_vers
        self.groups[group_id].latest_version_arn = group_ver.arn
        self.groups[group_id].latest_version = group_ver.version
        return group_ver

まずはコンストラクタです。 モックのバックエンドクラスではAWSサービスの各エンティティをOrderedDictで管理します。コンストラクトの中にメンバ変数をOrderedDictで初期化するロジックを入れていきましょう。

コンストラクタが実装できたら後は実際のバックエンド処理を実装します。 ココが一番の腕の見せ所です。boto3の挙動をよーく観察し、Greengrassのサービスが裏で何をしているのか思いを馳せながら適切な処理を実装します。

最後に他のファイルからimportしてもらうためにgreengrass_backendsを定義します。

models.py

available_regions = boto3.session.Session().get_available_regions("greengrass")
greengrass_backends = {region: GreengrassBackend(region) for region in available_regions}

これでモックの作成は終了です!

自作したモックを使ってみる

モックが自作できたの使ってみましょう! 今回のPJではpytestのfixture内でAWSサービスのモック開始・停止を行なっています。 greengrassのモック処理だけ自作したファイルからimportしてこんな感じで使います。

import boto3
import moto
import pytest
from unittest import mock

from tests.moto.greengrass import mock_greengrass

@pytest.fixture()
def start_moto_mock():
    moto.mock_iot().start()
    moto.mock_sts().start()
    moto.mock_iotdata().start()
    moto.mock_dynamodb2().start()
    moto.mock_s3().start()
    mock_greengrass().start()
    moto.mock_lambda().start()

    yield

    moto.mock_iot().stop()
    moto.mock_sts().stop()
    moto.mock_iotdata().stop()
    moto.mock_dynamodb2().stop()
    moto.mock_s3().stop()
    mock_greengrass().stop()
    moto.mock_lambda().stop()

#...略

テスト対象となるLambda関数のコードです。boto3を利用して入力値を元にGreengrassグループを作成するだけの簡単な処理です。

import boto3
import json


gg_client = boto3.client("greengrass", region_name="ap-northeast-1")


def handler(event, context):

    grp_name = event['body']['grp_name']
    res = gg_client.create_group(Name=grp_name)

    return json.dumps({
        "statusCode": 201,
        "body": res
    })

テストコード本体です

from botocore.exceptions import ClientError
import json
import pytest

from src.create_gg_group import handler


@pytest.mark.usefixtures('start_moto_mock')
class TestClass(object):

    def test_create_gg_group(self):

        event = {
            "body": {
                "grp_name": "test_grp"
            }
        }

        res = handler(event, {})
        data = json.loads(res)
        assert data["statusCode"] == 201
        assert data["body"]["Name"] == "test_grp"

    def test_empty_group_name(self):

        event = {
            "body": {
                "grp_name": ""
            }
        }

        with pytest.raises(ClientError) as ex:
            handler(event, {})
        assert ex.value.response["Error"]["Message"] == "Group name is missing in the input"
        assert ex.value.response["Error"]["Code"] == "InvalidContainerDefinitionException"

実際にpytestでGreengrass周りのテストを実行したログです

platform darwin -- Python 3.6.5, pytest-4.4.1, py-1.8.0, pluggy-0.9.0 -- /Users/xxxxxxx/.venv/bin/python
cachedir: .pytest_cache
rootdir: /Users/xxxxxxx/
collected 2 items

tests/unit/test_create_gg_group.py::TestClass::test_create_gg_group PASSED                                                                                                                                                                                        [ 50%]
tests/unit/test_create_gg_group.py::TestClass::test_empty_group_name PASSED

バッチリ実行できています。

まとめ

自力でモック作るのはかなりパワーがかかりましたが、その甲斐あってユニットテストはかなり楽になりました。きっとこの頑張りがPJの中盤ぐらいからボディブローのように効いてくるはず。 また、モックを実装するに当たってGreengrassというサービスが内部的にどうのようにデータを保持しているのか、各エンティティの関連はどうなっているのか?ということを分析したので、Greengrassというサービスにもそこそこ詳しくなれたかなと感じています。これもモックを自力実装して良かった点です。 自作したモックはまだ品質的に不安な面もあるので、もう少しプロジェクトを進めながらブラッシュアップした後でmoto本体にプルリクを上げたいと思います。

今回紹介した手順の大枠は他のAWSサービスにも流用できるはずなので、AWSサービスを使用するアプリのユニットテストでお困りの方がいれば、motoの自力拡張を検討してみてはいかがでしょうか?