対象の読者
- Web APIを利用するプログラムを開発している人
- Pythonを用いたプログラム開発している人
記事概要
- vcrpyは、プログラム内で実行されたHTTPリクエスト/レスポンスをファイルに記録し再生するためのpythonライブラリである。
- これを使うと、APIサーバの状態に依らず、テストを実行できるようになる。
前提条件
- Python 3.5.1
- vcrpy 1.10.5
はじめに
さまざまなAPIがWeb上に公開され、それと連携するサービスやプログラムも年々増えています。それに伴い、こう言ったAPIを用いるプログラムやサービスを作る機会も同様に増えて来ているかと思います。
今回取り扱う問題
- APIと連携するプログラムをテストしようとすると、単純なテスト実装ではAPIサーバーの状態の影響を受けてしまう。
- 影響を受けないようモックを作成すると、APIの数が増えると作成/管理が大変になる。
サンプルプログラム
実際にWebAPIをから情報を取得し、抽出結果を返す簡単なクラスを作成してみます。
from http import client
import json
class SampleApiClient(object):
def __init__(self, base_url):
"""
An Api Client that accesses to resources under specific url.
:param base_url: root url of API server that contains several resources (String)
"""
self.base_url = base_url
def get(self, resource, key='id', value=None):
"""
An method to get entry of specific resources that satisfy following searching option.
:param resource: a relative path from base_url that correspond to resource you want to access.(String)
:param key: an attribute name of resource you want to filter by. default: id (String)
:param value: a value of an attribute you want to filter by. (String)
:return: filtered_data: a result of operation. (Array of Dictionary)
"""
# create connection, and get raw data from API server
conn = client.HTTPConnection(self.base_url, port=80)
conn.request(method='GET', url=('/' + resource))
response = conn.getresponse()
raw_body = response.read()
json_body = json.loads(raw_body.decode(encoding='utf-8'))
# filter if value is specified.
if value is not None:
filtered_data = []
for entry in json_body:
if entry[key] == value:
filtered_data.append(entry)
else:
filtered_data = json_body
return filtered_data
上記は、下記のfake Web APIから、情報を取得するクラスです。
今回テストするのは、上記クラスの持つメソッド、getです。
これは、上記APIから、特定の条件にマッチした要素を抽出するメソッドです。
サンプルプログラムのテストコード
このメソッドをテストする、テストコードを次のように定義します。
このテストでは、リソース'todos'の取得結果を特定のタイトルで絞り込めるかどうかをテストしています。
(対象リソース): http://jsonplaceholder.typicode.com/todos/
from unittest import TestCase
from bin.sample_api_client import SampleApiClient
# Test Case Definition Starts from here.
class TestSampleApiClient(TestCase):
def test_get_todo_by_title(self):
client = SampleApiClient(base_url='jsonplaceholder.typicode.com')
# a free online REST service that produces some fake JSON data.
result = client.get(resource='todos', key='title', value="delectus aut autem")
expected = [
{
"userId": 1,
"id": 1,
"title": "delectus aut autem",
"completed": False
}
]
self.assertEqual(expected, result)
実際に実行してみます。
コマンド
python -m unittest test_sample_api_client.py
実行結果
----------------------------------------------------------------------
Ran 1 test in 0.311s
OK
このように、無事テストが通りました。
一方で、APIサーバの状態が変化すると、テストに失敗するようになる場合があります。
例えば、APIサーバまでの通信経路に障害が発生すると、テストの実行結果は次のようなエラーになります。
======================================================================
ERROR: test_get_todo_by_title (tests.test_sample_api_client.TestSampleApiClient)
----------------------------------------------------------------------
Traceback (most recent call last):
~ 省略 ~
OSError: [WinError 10065] 到達できないホストに対してソケット操作を実行しようとしました。
----------------------------------------------------------------------
Ran 1 test in 21.026s
FAILED (errors=1)
問題点
単体テストは、プログラムの単体での正当性を確認するものですが、このようなテスト実装では結合先(APIサーバ)や通信経路の影響を受けてしまいます。
他にも、別のユーザが、同じタイトルのtodoを投稿すると、ヒット件数が2件になり、実装自体が正常にも関わらずテストが失敗するようになってしまいます。
この問題を回避する方法としてよくあるものは、モックを作成し、下位モジュールをモックで置換し、戻り値を常に固定にしてしまうことです。(本記事では詳細は割愛します)
しかし、モックを作成するとなると、下位モジュールの戻り値を、連携するAPIやリソース毎に定義し管理する必要があり、連携する数が増えると大変な作業になります。
解決法
前置きが長くなりましたが、この問題を解決する手段として、"vcrpy"というライブラリを紹介します。
このライブラリを用いると、APIサーバに対して行われたHTTPリクエスト/レスポンスをファイルに記録し再生できるようになり、モックの作成の手間が省けるようになります。
vcrpyを用いたテストコード
まずは、以下のコマンドでvcrpyモジュールを環境にインストールします。
pip install vcrpy
次に、テストコードを、vcrpyを用いるように書き換えます。
from unittest import TestCase
import vcr
from bin.sample_api_client import SampleApiClient
# Instantiate VCR in order to Use VCR in test scenario.
vcr_instance = vcr.VCR( # Following option is often used.
cassette_library_dir='vcr/cassettes/', # A Location to storing VCR Cassettes
decode_compressed_response=True, # Store VCR content (HTTP Requests / Responses) as a Plain text.
serializer='json', # Store VCR Record as a JSON Data
)
# Test Case Definition Starts from here.
class TestSampleApiClient(TestCase):
@vcr_instance.use_cassette
def test_get_todo_by_title(self):
client = SampleApiClient(base_url='jsonplaceholder.typicode.com')
# a free online REST service that produces some fake JSON data.
result = client.get(resource='todos', key='title', value="delectus aut autem")
expected = [
{
"userId": 1,
"id": 1,
"title": "delectus aut autem",
"completed": False
}
]
self.assertEqual(expected, result)
変更点は大きく以下の2点となります。
- テストメソッドの実行前に、vcrpyモジュール内のVCRオブジェクトをインスタンス化する処理を記載する(7~11行目)
- テストメソッドに、@インスタンス名.use_cassetteとつける(17行目)
これだけです。この状態でテストを実行すると、8行目のVCRのオプションに指定したディレクトリに、HTTPリクエスト/レスポンス内容がファイルとして記録されます。
再度、このテストを実行すると、APIサーバーに対する通信は発生しなくなり、代わりにこのファイルからHTTPレスポンスが読まれるようになります。
実際に、再度ネットワークを不通にした上で、再度テストを実行してみます。
実行結果
----------------------------------------------------------------------
Ran 1 test in 0.012s
OK
まとめ
このように、vcrpyを用いる事で、単体テストにおいて、対向装置(APIサーバ)の状態の影響を受けにくくすることができます。
手軽に実装できるので、是非試してみてください。