対象の読者

PythonベースのWebアプリケーションのテストコードを書く方
- 特に、Djangoを用いている方
- 特に、Databaseに関するボリュームテストを実装する予定の方

概要

Webアプリケーションを運用する上では、Databaseに大量にデータが入ってもサービスが正常に機能するかをテストしたい場合があります。
- 例えば、1000万件レコードが入ったときに、特定のレコードn件を抽出する処理時間など
- 特に、バッチ処理にかかる時間を見誤ると、正常にサービスを提供することが困難になる場合もあります。
本記事では、こういったテスト(=ボリュームテスト)をPython/Djangoで実施する際に、押さえておくと便利なTipsを紹介します。

前提条件

以下の条件でテストをしています。
- Django==1.10.2
- Python 3.5.2
bashのコマンドは、Mac(macOS Sierra)で動作確認をしています。
プロジェクトのディレクトリ構成は次の通りとします。

project_root/
    manage.py
    mysite/
        __init__.py
        settings.py
        urls.py
        wsgi.py
    myapp/
        models.py
        views.py
        ...
        tests.py

本文

テストコードの基本構成と問題点

DjangoのUnittestフレームワークでボリュームテストを実装する場合、テストコードの基本構成は次のようになります。

project_root/myapp/tests.py

from django.test import TestCase

class VolumeTest(TestCase):
    def setUp(self):
        self.setup_data()

    def setup_data(self):
        ... # テストデータを作成する処理

    def test_volume_test(self):
        ... # テスト内容。例えば、n件 SELECT してその時間を計測するなど。

しかしながら、次のような問題に出くわすでしょう。
- 単体テストの度にボリュームテストが実行され、時間を大きく消費する。
- ボリュームテスト用に大量のデータを作成するプログラムを書く必要がある。
- ボリュームテストの時間が長すぎてコンソールがタイムアウトする。
このような問題に対するTipsをこれから紹介したいと思います。

Tips1: 他のテストケースと実行を分離する

Unittestフレームワークには、特定条件が指定された時のみテストを実行する/スキップする機能があります。
これを用いて、ボリュームテストが通常テスト時に実行されないようにしましょう。
Djangoでは、テスト用のsettings.pyと、skipUnlessデコレータを用いることで、実現できます。
まず、テスト用のsettingsである settings_volume_test.py を、settings.pyと同じディレクトリに作成します。
- 通常の settings.py の中身を継承し、ボリュームテスト用フラグ定数のみを含むような実装にします。

project_root/mysite/settings_volume_test.py

from mysite.settings import *
ENABLE_VOLUME_TEST = True

次に、上記settingsに定義した定数を用いて、テストケースをskipUnlessデコレータでデコレートします。

project_root/myapp/tests.py

from django.test import TestCase
from django.conf import settings  # new

from unittest import skipUnless  # new

@skipUnless(condition=getattr(settings, 'ENABLE_VOLUME_TEST', False),  # new
            reason='Volume test is not enabled in current settings.')
class VolumeTest(TestCase):
    def setUp(self):
    ...

これで、通常はボリュームテストがスキップされるようになります。
ボリュームテストを実行する場合は、プロジェクトルートで、次のように明示的にsettingsを指定します。

./manage.py test myapp.tests --settings mysite.settings_volume_test

Tips2: ライブラリでテストデータを自動生成する

ボリュームテストをする際には、大量のデータを生成する必要があり、少なくとも以下のような手間が必要になります。

データを自動生成するためのコードを書く。
- 文字列, 整数, 時刻など、型ごとに自動生成するコードを書く。
- ユニーク制約がある属性に対しては、シーケンスなども定義する必要がある。
Databaseなどに記録するためのコードを書く。

このような処理を代替してくれるfactory-boyというライブラリがpythonにはありますので、簡単に紹介したいと思います。

factory-boyを用いたデータの自動生成

前提条件

factory-boyで自動生成するデータの構造は、DjangoのModelとして、以下のように定義されているとします。

project_root/myapp/models.py

from django.db import models


class MyAppUser(models.Model):

    display_name = models.CharField(max_length=32, unique=True)
    e_mail = models.EmailField()
    gems = models.IntegerField()  # a virtual money in this app.
    created_time = models.DateTimeField()
    updated_time = models.DateTimeField()

    def __str__(self):  # print時にモデルの中身を確認するために定義する。
        return self.display_name + '@' + str(self.created_time.date()) +\
               ', gems:' + str(self.gems) + ', mail:' + self.e_mail

Step1: 前準備

ライブラリをインストールします。

pip install factory-boy==2.10.0

ライブラリを使うにあたり、必要な引数を生成するため、以下のライブラリもインストールします。

pip install pytz==2018.3

Step2: Factoryの定義

前提条件で定義したモデル MyAppUser に対応するファクトリを以下のように作成します。
- 実装内容に関する解説は後述します。

project_root/myapp/utils/factory.py

from factory.django import DjangoModelFactory
from factory.fuzzy import FuzzyInteger, FuzzyText, FuzzyDateTime
from factory import Sequence

from myapp.models import MyAppUser
from myapp.utils.datetime import get_good_old_time  # 定義は後述

_default_start_dt = get_good_old_time()


class MyAppUserFactory(DjangoModelFactory):

    class Meta:
        model = MyAppUser

    display_name = Sequence(lambda n: 'user{0}'.format(n))
    e_mail = FuzzyText(length=12, suffix='@example.com')
    gems = FuzzyInteger(low=0, high=99999)

    # end_dt will be datetime.now()
    created_time = FuzzyDateTime(start_dt=_default_start_dt, force_microsecond=0)
    updated_time = FuzzyDateTime(start_dt=_default_start_dt, force_microsecond=0)

解説 (大枠をまず把握したい方は読み飛ばしてください)

import部

factory-boyはfactoryという名前でインポートできます。
DjangoのModelに対応するFactoryを作成するため、DjangoModelFactoryという基底クラスをimportしています。
FuzzyIntegerといったFuzzy系は、ランダムな値を生成するためのオブジェクトです。
- 整数以外にも、文字列、時刻、浮動小数点、選択肢から1つ選ぶといったオブジェクトもあります。
- 公式ドキュメントはこちら
Sequence というオブジェクトは、シーケンスを生成するためのオブジェクトです。
- 一意性制約がある属性の値生成に使います。
MyAppUserは、先ほど定義した、自動生成する対象のオブジェクトです。

get_good_old_timeは、1970年1月1日のタイムゾーン付きの時刻情報を返す自作のヘルパメソッドです。

FuzzyDateTimeの引数に必要になるため作成しています。
ちなみに実装は以下のようになっています。

project_root/myapp/utils/datetime.py

from datetime import datetime
import pytz


def get_good_old_time():
    time_zone = pytz.timezone('UTC')
    dt = datetime(year=1970, month=1, day=1,
                  hour=0, minute=0, second=0, microsecond=0,
                  tzinfo=time_zone)
    return dt

ファクトリ定義部

MyAppUser のインスタンスを自動生成するファクトリ MyAppUserFactory は、DjangoModelFactoryを継承して作ります。
クラス内に Meta という名前のクラスを定義し、変数 model に自動生成対象のクラスへの参照を代入します。
- ここで、自動生成対象のオブジェクトを指定しています。
上記の定義が済んだら、MyAppUserの属性に対応する変数を一つずつ定義し、Fuzzy/Sequenceオブジェクトへの参照を代入します。
- これは、属性毎に、ランダム値の生成に用いる生成器を指定している事に等しいです。
- 基本的には、ユニーク制約のある属性に対してはSequenceを、それ以外には、Fuzzyオブジェクトを指定します。
  - 今回は、ユニーク制約のあるdisplay_name属性にSequenceを指定しています。
- FuzzyTextでは、文字列を生成する際、ランダム部分の文字列長をlength、固定値部をprefix, suffixとして指定できます。
- FuzzyIntegerでは、生成する値の範囲を指定することができます。
- FuzzyDateTimeでは、生成する時刻の範囲を指定することができます。(最小値に関しては指定必須)
  - time_zone付きのDatetimeオブジェクトを渡す必要があります。
    - 上記サンプルコードでは、ヘルパメソッドで生成した値を渡しています。
  - また、生成した値のマイクロ秒以下の値を切り捨てるなどの制御に関する指定も可能です。

Step3: Factoryを用いたテストコードを書く

上記のMyAppUserFactoryを用いて、ようやくボリュームテストが作成できます。基本的には以下の流れになります。
- Unittestのsetupメソッド内で、MyAppUserFactoryのcreate_batchメソッドを呼び出し、テストデータをDatabase上に作成する。
- テストメソッド内で、テストデータに関する何らかの操作をし、その処理時間を計測する。
  - 実際には、メモリ使用量/CPU使用率/DiskIOなども計測したほうがよいですが、本記事では省略します。
テストコードの実装例を以下に示します。

project_root/myapp/tests.py

from django.test import TestCase
from django.conf import settings
from unittest import skipUnless

from myapp.models import MyAppUser
from myapp.utils.factory import MyAppUserFactory

from myapp.utils.stopwatch import display_spent_time  # 自作のコンテキストマネージャ。詳細は後述

# Create your tests here.


@skipUnless(condition=getattr(settings, 'ENABLE_VOLUME_TEST', False),
            reason='Volume test is not enabled in current settings.')
class VolumeTest(TestCase):

    test_size = 10000  # Import test_size from settings.py is a better idea.

    def setUp(self):
        self.setup_data()

    def setup_data(self):
        factory = MyAppUserFactory

        print('Creating records for volume test...')
        factory.create_batch(size=self.test_size)

    def test_volume_test(self):
        print('Executing a volume test..')
        # Test total times of querying and evaluating 5 latest users who registered to this app.
        with display_spent_time(operation_name='get latest 5 Users'):
            new_users = MyAppUser.objects.order_by('-created_time')[:5]
            print([user for user in new_users])

上記テストケースでは、MyAppUserを10000件登録した時に、最新のユーザ5件を取得し出力する時間を計測しています。
- 変数test_sizeの値を増減することで、件数に関する条件を変えることができます。
- 例では10000という値をハードコードしていますが、settingsに記載し読み込むようにすると、簡単に変更できて便利になります。

また、コード中のdisplay_spent_timeという自作のコンテキストマネージャは、withブロック内の処理にかかった時間を出力する機能を持ちます。

実装は以下のようになっています。

project_root/myapp/utils/stopwatch.py

from datetime import datetime
from contextlib import contextmanager


@contextmanager
def display_spent_time(operation_name="An operation"):

    start_time = datetime.now()
    yield  # Execute Inner code of "with" statement.
    end_time = datetime.now()

    spent_time = end_time - start_time
    print('{0} took {1}'.format(operation_name, spent_time))

あとは、テストを実行するのみです。次のような結果が出力されます。

Creating test database for alias 'default'...
Creating records for volume test...
Executing a volume test..
[<MyAppUser: user9999@1970-01-01, gems:5647, mail:dcwPZQCMWUqx@example.com>, <MyAppUser: user9998@1970-01-01, gems:84250, mail:xfGGKqKgvdBf@example.com>, <MyAppUser: user9997@1970-01-01, gems:63671, mail:RPvZyuFpkQxw@example.com>, <MyAppUser: user9996@1970-01-01, gems:40592, mail:dVcCSxptJUmq@example.com>, <MyAppUser: user9995@1970-01-01, gems:79379, mail:YVLgvdEtTlDd@example.com>]
get latest 5 Users took 0:00:00.007239
.
----------------------------------------------------------------------
Ran 1 test in 4.226s

OK
Destroying test database for alias 'default'...

ただし、自動生成したデータを用いたボリュームテストでは、本番サービスと同じような性能特性が得られない場合があります。
- 注意事項として、本記事の最後に記述いたしましたので、併せてご確認ください。

Tips3: バックグラウンドでテストを実行する

以下のコマンドで、テストを実行できますが、実行に時間がかかるテストではコンソールがタイムアウトする場合があります。
- 特にリモートサーバ上で実行する場合によく発生します。

./manage.py test myapp.tests --settings mysite.settings_volume_test

nohupコマンドを用いることで、完全にバックグラウンドで実行することができます。
- 結果はテキストファイルに出力されるため、後から確認できます。

nohup 上記コマンド > volume_test_result.txt &

また、以前に中断したテストがある場合、テスト用に生成されたDataBaseが残っており、yesと入力しなければテストが続行できない場合があります。
それによりテストが妨害されないようにするためには、以下のようにコマンドを工夫します。

nohup bash -c 'yes yes | head -n 1 | 上記コマンド' > volume_test_result.txt &

まとめ

本記事では筆者がつまづいた経験を元に、ボリュームテストに関するTipsを紹介しました。
これらのTipsがみなさまのお役に立てると幸いです。

注意事項

(重要)テストデータの自動生成に関する注意点

本来の性能特性が得られない場合がある

上記テストコードでは、MyAppUserの作成時刻を完全にランダムに生成しています。したがって、大量にレコードを生成した場合、ほぼ同時刻のデータが物理的に記憶装置の各所に散らばります。
一方、実際のサービス環境では、生成時刻がほぼ同時刻のMyAppUserは、物理的に記憶装置上で近い位置に配置される可能性が高いです。
したがって、時刻の範囲を指定するクエリを投げるようなテストを記載した場合、実際のサービス環境よりI/Oが増え、大きく性能が落ちる場合があります。
このように意図せずテスト条件が実際のサービスのものと乖離する場合があるため、以下のような属性はランダム値の代わりにシーケンスを用いることをおすすめします。
- Index が設定される属性
- データの物理配置に規則性があり、かつ検索条件となりうる属性

例えば、上記MyAppUserFactoryの以下の時間属性

project_root/myapp/utils/factory.py

created_time = FuzzyDateTime(start_dt=_default_start_dt, force_microsecond=0)

は

project_root/myapp/utils/factory.py

from datetime import timedelta

created_time = Sequence(lambda n: _default_start_dt + timedelta(seconds=n))

として生成したほうが、より本番環境に近い性能特性が得られます。