こんなのがあったのか!? Moto - Mock AWS Services

はじめに

AWS-SDKを使って各AWS Serviceを呼び出しているプログラムに対して、テストを行う場合、各々のAWS Serviceに対して、 Mockするのはとても大変です。

なにか良いライブラリやサービスはないかと探しているとMotoと呼ばれるAWS Serviceに対してMockしてくれるライブラリを見つけた＆使ってみて感動したので記事を書くことにしました。

Moto - Mock AWS Services

詳細は、以下GitHubのリンク先を参照ください。

注意として、Motoは全てのAWS Serviceの呼び出し（関数）に対して、Mockしてくれるわけではありません。
以下にMotoが対応している関数の対応表がありますので、Motoを使用する時は、以下対応表を参照して、Motoを使ってMockできるかを確認しましょう。

もし、対応していない関数に遭遇した場合は、自力でMockするか、localstack等の別のツールを使いましょう。

使い方（例）

ここでは、以下2つの例を使ってMockの方法を説明したいと思います。

• EC2のインスタンスIDを取得する処理に対して、Mockしたい。
• S3のバケットからデータを取得する処理に対して、Mockしたい。

EC2のインスタンスIDを取得する処理に対して、Mockしたい。

hoge.py(テスト対象のコード)

import boto3

def get_ec2_instance_id():
    # EC2からインスタンスIDを取得
    ec2 = boto3.client('ec2', region_name='ap-northeast-1')

    instance    = ec2.describe_instances()
    instance_id = instance['Reservations'][0]['Instances'][0]['InstanceId']

    return instance_id

test_hoge.py(テストコード)

import boto3
import os
import pytest
from moto.ec2.models import AMIS
from moto import mock_ec2

os.environ['AMI_IMAGE'] = AMIS[0]['ami_id']

@mock_ec2
def test_ec2():
    # EC2インスタンスを立ち上げる（AMIは、Moto側で用意してあるものを使用します)
    ec2 = boto3.client('ec2', region_name='ap-northeast-1')
    ec2.run_instances(ImageId=os.environ['AMI_IMAGE'], MinCount=2, MaxCount=2)

    # 立ち上げたEC2インスタンスからインスタンスIDを取得する
    all_instance_info = ec2.describe_instances()
    instance_id       = all_instance_info['Reservations'][0]['Instances'][0]['InstanceId']

    # テスト対象のコードと、インスタンスIDを比較、一致を確認する
    assert instance_id == hoge.get_ec2_instance_id()

boto3.client('ec2')に対して、Motoを使ってMockしたい場合、from moto import mock_ec2を宣言して、テストメソッドの上に@mock_ec2と記述するだけです。

Motoは、boto3.client('ec2')に対してMockしているため、test_hoge.pyだけでなく、hoge.py上使うboto3に対しても、Mockされます。

EC2からインスタンスIDを取得するには、事前にEC2インスタンスを立ち上げておく必要があります。
そちらに対しても、自動でMockしてくれており、おまけにMotoにはAMIのサンプルが付属しているので、サンプルAMIをベースにEC2インスタンスを立ち上げるといった事もできます。

そのため、test_hoge.pyで事前にEC2インスタンスを立ち上げておき、hoge.pyでEC2インスタンスからインスタンスIDを取得できるよね。といったテストができます。

S3のバケットからデータを取得する処理に対して、Mockしたい。

hoge.py(テスト対象のコード)

import boto3
import json
import os

def get_s3_file():
    # バケットからデータを取得する
    s3     = boto3.resource('s3')
    s3_obj = s3.Object(os.environ['S3_BUCKET_NAME'], os.environ['BACKUP_FILE_KEY']).get()

    return json.load(s3_obj['Body'])

test_hoge.py(テストコード)

import boto3
import json
import os
import pytest
from moto import mock_s3

os.environ['S3_BUCKET_NAME']  = 's3_backet_name'
os.environ['BACKUP_FILE_KEY'] = 'backup_file_key'

@mock_s3
def test_s3():
    # S3のバケットを作成、適当なデータをバケットに入れる。
    s3 = boto3.resource('s3')
    s3.create_bucket(Bucket = os.environ['S3_BUCKET_NAME'])

    s3_obj = s3.Object(os.environ['S3_BUCKET_NAME'], os.environ['BACKUP_FILE_KEY'])

    test_json = {'key': 'value'}
    s3_obj.put(Body = json.dumps(test_json))    

    # テスト対象のコードと、バケットから取得したデータを比較
    assert test_json == hoge.get_s3_file()

流れとしては、EC2と同じです。
boto3.resource('s3')に対しても、from moto import mock_s3と宣言して、テストメソッドの上に@mock_s3と記述するだけでMotoを使ってMockできます。

EC2のインスタンスを事前に立ち上げておくのと同様、S3のバケットを事前に作成（こちらも自動でMockしてくており、バケットもkey名も架空の名前を使用します）して、データをS3のバケットに格納します。

これで、S3のバケットからデータを取得することができるよね。といったテストができます。

何故、AWS-SDK for Python(boto3)のMockが難しいのか？ 自力でMockはできないのか？

ここに関しては、正直、悩みました。（+ _ +)
しかし、悩んだ末になんとなく見えてきた部分もあるので、備忘録として書いておこう思います。（もし、内容に誤り等がありましたら、ご指摘ください。）

boto3のMockが難しい理由は、インスタンス化する時の宣言にあります。
boto3.client('ec2')やboto3.resource('s3')でインスタンス化すると思いますが、このec2やs3（AWS Service名）によって、インスタンス化される内容が変わります。

このインスタンス化される内容が変わる部分ですが、Pythonプログラムに関数が直接書かれていると思いきや、そうではなく、Jsonファイルに関数が書かれていました。
さすがのMockもファイルから読み込まれる関数（実態のない関数）に対してMockできない・・・というわけですね。(= = ;;)

言葉だと分かり辛いので、図でも表現してみます。

仕組みはなんとなく理解できました。
では、自力でMockは無理なのか？と調べて＆考えたら、以下の方法でMockできました。

SSMを使ってEC2インスタンスからkernelのバージョン情報を取得・表示するプログラムです。
Motoを使用せずにMockしています。

hoge.py(テスト対象のコード)

import boto3
import json
import os

def ec2_kernel_version():
    # SSMのsend_commandを使用して、EC2インスタンスへUnixコマンドを送信する
    ssm = boto3.client('ssm', region_name='ap-northeast-1')
    send_command_response = ssm.send_command(
        InstanceIds  = 'ec2_instance_id',
        DocumentName = 'AWS-RunShellScript',
        Comment      = 'Command uname -r',
        Parameters   = {
            'commands': [
                'uname -r',
            ]
        }
    )

    # SSMのsend_command終了まで待機する
    time.sleep(2)

    # SSMのsend_commandの結果を取得するため、CommandIDを取得する
    command_id = send_command_response['Command']['CommandId']

    # SSMのlist_command_invocationsを使用して、send_commandの結果を取得する
    list_command_response = ssm.list_command_invocations(
        CommandId = command_id,
        Details=True
    )

    # SSMのlist_command_invocationsからkernelのversion情報を取得する
    ec2_kernel_version = list_command_response['CommandInvocations'][0]['CommandPlugins'][0]['Output'].rstrip('\n')

    return ec2_kernel_version

test_hoge.py(テストコード)

import boto3
import json
import os
import pytest

# ① SSMのsend_command、list_command_invocationsの出力結果を記載する
def make_api_call(self, operation_name, kwarg):
    if operation_name == 'SendCommand' :
        return {
            'Command': {
                'CommandId': 'test_command_id'
            }
        }
    if operation_name == 'ListCommandInvocations' :
        return {
            'CommandInvocations': [
                {
                    'CommandPlugins': [
                        {
                            'Output': 'hoge'
                        }
                    ]
                }
            ]
        }

    return 'NG'

def test_ssm(mocker):
    # ② botocore.client.BaseClient._make_api_callの関数を①の関数で上書きする
    mocker.patch('botocore.client.BaseClient._make_api_call', new=make_api_call)

    assert 'hoge' == hoge.ec2_kernel_version()

ポイントはbotocore.client.BaseClient._make_api_callに対して、Mockのpatchを当てているところです。
_make_api_call関数には、Jsonファイルから関数を取得する処理が書かれています。
そのため、この関数に対してMock関数で上書きを行います。
上書く方法ですが、operation_nameにはJsonファイルから取得した関数名が入ってくるので、Mockしたい関数名を拾って、出力する値を設定してあげればOKです。

これで、SSMのsend_commandとlist_command_invocationsに対して、Mockすることができました。

長い間、悩み＆考えましたが、一旦自分の中ではすっきりしたので、良い学びになりました。（= o =)