boto3+mockでテスト

はじめに

boto3を使って、AWSのS3に対して操作をするようなコードを書きました。
テストコードを書いてみようと思い、mockとunittestを用いて作成することにしました。

準備

まずは環境の準備です。
適当なフォルダを作成してvirtualenvを使って環境を作ります。
今回はAWS Lambdaに乗せることを前提としています。
後、boto3とmockをpipしときます。

環境準備

$ mkdir s3operation
$ cd s3operation
$ virtualenv .
$ source bin/activate
$ pip install boto3
$ pip install moto

S3への操作するサンプルクラス

$ pwd
~/s3operation/
$ vi s3access.py

s3access.py

# -*- coding:utf-8 -*-
import boto3
import os.path

class S3Access:
    s3 = None
    def __init__(self):
        self.s3 = boto3.client('s3')

    def upload(self, bucketname, uploadfilepath, s3folder):
        """
        S3へアップロード
        @param bucketname S3バケット名
        @param uploadfilepath アップロードするファイルのパス
        @param s3folder S3バケットの下のフォルダパス（ファイル名は含まない）
        """
        basefilename = os.path.basename(uploadfilepath)
        s3filepath   = s3folder + '/' + basefilename
        self.s3.upload_file(uploadfilepath, bucketname, s3filepath)

テストコード

今回は、アップロードだけのプログラムにしました。
では、次にテストコードを作成します。まずは、環境構成を。

環境構成

今回のファイル／フォルダ構成です。
テストコードだけ、別のフォルダ(今回はtestフォルダ)に作成したかったので、以下のような感じにしています。

~/s3operation/
    s3access.py
    test/
        testfile.txt
        tests3access.py

テストコード

テストコードは以下のような感じです。

tests3access.py

# -*- coding:utf-8 -*-
import boto3
import unittest

from moto import mock_s3

import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + '/../')
from s3access import S3Access

class TestS3Access(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    @mock_s3
    def test_s3_upload(self):
        #初期化
        bucketname = 'testbucket001'
        prefix     = 'firstfolder'

        ##テスト準備(バケット、オブジェクト作成）
        s3test = boto3.client('s3')
        s3test.create_bucket(Bucket=bucketname)
        s3test.put_object(Bucket=bucketname, Key=prefix)

        #テスト対象のオブジェクト生成
        s3 = S3Access()
        #対象メソッド実行
        s3.upload('./test.md', bucketname, prefix)

        #実行結果確認
        ##アップロード済みの情報をS3から読み込む
        keyname  = prefix + '/test.md'
        response = s3test.get_object(Bucket=bucketname, Key=keyname)
        body     = response['Body'].read().decode('utf-8').encode('utf-8')
        ##アップロードしたファイルを読み込む
        fp       = open(filename, 'r')
        readStr  = fp.read()
        fp.close
        ##結果確認
        assert body2 == readStr

テスト実行

テストコードもできたので、実際にテストを実行します。

$ pwd
~/s3operation/test/
$ python -m unittest discover ./
.
----------------------------------------------------------------------
Ran 1 tests in 0.472s

OK

これで、一つのテストが実行されて成功したことになります。

失敗した場合

何かプログラムにエラーが存在する場合、以下のような感じになります。
エラー情報から解析して後は直すだけです。

python -m unittest discover ./
E
======================================================================
ERROR: test_s3_upload (tests3access.TestS3Access)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "~/s3operation/lib/python2.7/site-packages/moto/core/models.py", line 70, in wrapper
    result = func(*args, **kwargs)
  File "~/s3operation/test/tests3access.py", line 43, in test_s3_upload
    response = s3test.get_object(Bucket=bucketname, Key=keyname)
  File "~/s3operation/lib/python2.7/site-packages/botocore/client.py", line 310, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "~/s3operation/lib/python2.7/site-packages/botocore/client.py", line 599, in _make_api_call
    raise error_class(parsed_response, operation_name)
NoSuchKey: An error occurred (NoSuchKey) when calling the GetObject operation: The specified key does not exist.

----------------------------------------------------------------------
Ran 6 tests in 0.519s

FAILED (errors=1)

さいごに

今回の例では、S3の一部だけの評価を書きましたが、他にも以下のAWSリソースにも対応しています。
- API Gateway
- AutoScaling
- CloudFormation
- CloudWatch
- DataPipeline
- DynamoDB
- EC2
- ECS
- ELB
- EMR
- Glacier
- IAM
- Lambda
- Kinesis
- KMS
- RDS
- Redshift
- Route53
- S3
- SES
- SNS
- SQS
- SSM
- STS
- SWF

普段からテストコード作成をクセづけておき、テストコードがオールグリーンになれば開発完了になる感じになっていけばなと思いました。
まずはそのとっかかりにでも。