この記事は、OthloTech Advent Calendar 2018 10日目の記事として書かれています。
TL;DR
- Pythonでboto3(AWS SDK)を使う時にはmotoを使おう。
- 機能が足りない時はISSUEを書いたり、余裕があればプルリクエストを送ろう。
AWS SDKとは
AWS SDKとはAWSの機能をプログラムから制御するために利用するライブラリを指します。AWSコンソールを利用して行うことは概ね行うことができるのでアプリケーションがAWSを直接的に利用する際等(S3のファイルにアクセス、インスタンスを立てる等)に利用されます。リンク先に記述があるように多くの言語で利用することが可能です。
自分はLambdaでAWSリソースを制御したい時に使うことが多いです。また、AWS-CLIがPythonやSDKであるboto3に依存していて安定していそうな気がするということからPythonを使うことが多いです。したがって本記事ではPythonのAWS SDKであるboto3についてのみ言及しています。
AWS SDKの開発をしていて辛かったこと
AWS SDKではリソースを作成するようなプログラムを作成することがそれなりにありましたが、課金対象になってしまうのであまりテストプログラムを書こうとは思えませんでした。また、VMを立てるようなプログラムの場合仮にテストを行ったとしてもかなり時間がかかってしまいます。CIで実行するならまだましですが、手元で実行することもあるので待ち時間があるのは厳しいです。
さらに言えば個人的には開発手法としてのテスト駆動開発のような細かい目標地点がわかりやすいテストが好きなこと、CIツールにアラートを出されるとなんとなく守られてる気がするのが嬉しいというのもありテストは書きたかったです。
Motoとは
OSSでboto3向けに開発されているモックライブラリです。所定の宣言を行うことでboto3を実行した時にそれっぽい値を返却するようになります。また、その時AWSにはアクセスしません。リポジトリは下記です。
なお、他の言語でもある程度モックプログラムが開発されているらしいので興味がある人は探してみてくださいね。
使い方
前準備
パッケージのインストールをしましょう。環境によって動かし方は異なると思います。基本Python3の方が良いと思います。
pip3 install moto
テスト対象
例えば下記の2つの関数のテストがしたいとします。
import boto3
def get_s3_file_body():
bucket_name = 'hoge'
file_name = 'fuga.txt'
s3 = boto3.resource('s3')
response = s3.Object(bucket_name, file_name).get()
return response['Body'].read().decode('utf-8')
def stop_instance(instance_ids):
ec2_client = boto3.client('ec2')
ec2_client.stop_instances(InstanceIds=instance_ids)
get_s3_file_bodyがs3のhogeというバケットのfuga.txtの中身を取得するプログラムです。stop_instanceが引数で与えられたInstanceIdのインスタンスを停止させるプログラムです。まあ、なんとなくイメージしやすいプログラムなんじゃないでしょうか。
テストプログラム
サクッとunittestを使って作ることにします。Pipenvが出て、テスト専用のライブラリを導入しやすくなりましたが、標準で含まれているunittestは使う分には楽でいいですね。
import os
import unittest
import boto3
import moto
from moto.ec2.models import AMIS
from boto3_example import get_s3_file_body, stop_instance
os.environ['AWS_DEFAULT_REGION'] = 'ap-northeast-1'
class TestBoto3Example(unittest.TestCase):
@moto.mock_s3
def test_get_s3_file_body(self):
bucket_name = 'hoge'
file_name = 'fuga.txt'
expected = 'foobar'
s3 = boto3.resource('s3')
s3.create_bucket(Bucket=bucket_name)
s3.Object(bucket_name, file_name).put(Body=expected)
self.assertEqual(expected, get_s3_file_body())
@moto.mock_ec2
def test_stop_instance(self):
# make instance
ec2_client = boto3.client('ec2')
ec2_client.run_instances(ImageId=AMIS[0]['ami_id'], MinCount=2, MaxCount=2)
# get ids
full_info = ec2_client.describe_instances()
instance_ids = []
for r in full_info['Reservations']:
for i in r['Instances']:
# check running
self.assertEqual('running', i['State']['Name'])
instance_ids.append(i['InstanceId'])
# stop instances
stop_instance(instance_ids)
# check stopping
full_info = ec2_client.describe_instances()
for r in full_info['Reservations']:
for i in r['Instances']:
# check running
self.assertEqual('stopped', i['State']['Name'])
self.assertTrue(i['InstanceId'] in instance_ids)
if __name__ == '__main__':
unittest.main()
さて、リージョンについてですが、ec2ではデフォルトのリージョンが設定されていない場合にはエラーが発生するため普段環境変数に持たせていたり、Lambdaでの運用が前提の場合にはテストプログラムで環境変数を使うといいと思います。
利用する際には関数の前に利用する機能デコレータを設定することでboto3はAWSにアクセスするのではなく、motoを見に行くようになります。テストしたい関数に合わせて設定しましょう。特別なコードとしてはこれだけで、あとは普通にテストプログラムを各要領でプログラムを書けばOKです。
test_get_s3_file_bodyではS3にアップロードして、それをget_s3_file_bodyで読み込み、書き込んだ内容と比較しています。test_stop_instanceではインスタンスを作成して、そのIDを確認するとともに動作していることを確認しています。次にstop_instanceを動作させます。最後に状態として、停止状態にあることとInstanceIdに変更がないことを確認しています。
実行してみる
普通にunittestなのでそのまま実行します。
➜ python -m unittest discover
..
----------------------------------------------------------------------
Ran 2 tests in 0.511s
OK
正常に動作することが確認できました！
注意事項
CloudWatchのメトリクス取得部分などまだまだ実装されていない機能も多く存在します。気になるかたISSUEで連絡したり、OSSということでご自身で実装するのも面白いかと思います。
まとめ
motoを使うとboto3のテストが大きな手間なく行えます！
motoを使ってboto3を使ってプログラムでもテストプログラムを書こう！