[django、TransactionTestCase]別スレッドを起動しているアプリの注意点

リクエストのタイムアウトの関係上、重い処理をdjangoのviewで走らせたい場合、予約して後程バッチ処理で実行したり、別スレッドを立ち上げて実行することがあるかと思います。ただ、別スレッドを立ち上げるビューのテストケースを書いたときに、TestCase(django.test.TestCase)を継承したテストケースだと期待したテスト結果を得られないことがありました。

この投稿ではビューから別スレッドを立ち上げた場合に、TestCase(django.test.TestCase)を継承したテストケースで起こり得る問題と、その解決方法を紹介します。

今回使ったdjangoのバージョンは1.8.3です。また、djangoのアプリが接続するDBはMySQLを使っています。

テスト対象のアプリ作成

まずはdjangoを使って、テスト対象のアプリを作成します。プロジェクト名はmyproject、アプリ名はmyappで作成します。

myapp/models.py

期待した通りに動かなかったのはDB周りだったので、モデルを作成します。

# coding: utf-8
from django.db import models


class Foo(models.Model):
    name = models.CharField(max_length=32)

myapp/views.py

別スレッドで、Fooにレコードを追加するビューを作成します。

# coding: utf-8
import threading
from django.views.generic import View
from django.http import HttpResponse
from myapp.models import Foo


class FooCreate(View):
    def get(self, request, *args, **kwds):
        class FooThread(threading.Thread):
            """別スレッドで起動させるクラスを定義します。"""
            def __init__(self, name, *args, **kwds):
                super().__init__(*args, **kwds)
                self.name = name

            def run(self):
                # Fooにデータを追加します。
                foo = Foo()
                foo.name = self.name
                foo.save()

        t = FooThread('foo')
        t.start()  # スレッドを起動します。
        return HttpResponse('OK')

myproject/urls.py

urls.pyも書き換え、作成したビューを呼べるようにします。

# coding: utf-8
from myapp import views as myapp_views
from django.conf.urls import url

urlpatterns = [
    url(r'create/', myapp_views.FooCreate.as_view(), name='myapp_foocreate'),
]

これで、テスト対象のアプリが完成しました。

テストケース(失敗)

テストケース作成

TestCase(django.test.TestCase)を継承して、テストケースを作成します。テストケースでは、ビューを叩いて、Fooが新規作成されていることを確認します。

# coding: utf-8
import time
from django.test import TestCase, client
from django.core.urlresolvers import reverse
from myapp.models import Foo


class TestFooCreate1(TestCase):
    def test_get(self):
        self.assertEqual(Foo.objects.count(), 0)  # 初期状態の確認です。
        c = client.Client()
        c.get(reverse('myapp_foocreate'))  # テスト対象を実行します。
        time.sleep(1)  # 別スレッドの終了を待つため、1秒間スリープさせます。
        self.assertEqual(Foo.objects.count(), 1)  # テスト結果を確認します。

テストケース実行

上記テストケースを実行してみます。すると、1回目は成功します。

$ ./manage.py test --keepdb myapp.tests.TestFooCreate1
.
----------------------------------------------------------------------
Ran 1 test in 1.100s

OK

しかし、2回目に実行すると、テストは失敗します。

$ ./manage.py test --keepdb myapp.tests.TestFooCreate1
F
======================================================================
FAIL: test_get (myapp.tests.TestFooCreate1)
...
    self.assertEqual(Foo.objects.count(), 0)  # 初期状態の確認です。
AssertionError: 1 != 0

----------------------------------------------------------------------
Ran 1 test in 0.097s

FAILED (failures=1)

初期状態の確認で失敗しているので、どうやらテストケース後にDBがリセットされていないみたいです。

※このテストケースだと「--keepdb」を外すだけで解決しますが、それは根本的な解決にはなりません。なお、「--keepdb」をつけると、テストケース終了後にDBを破棄しなくなります。

DBのデータがリセットされていない理由

djangoのドキュメントに書いてありますが、TestCaseを継承したテストケースでは、最初にトランザクションを開始しています。そのトランザクションの中で開発者が記述したテストケースが動きます。そして、1つのテストケースが終わった後にロールバックすることで、テストで作成等したデータを破棄しています。ちなみに、この動作はデータベースのリセットにかかる時間の速度を上げるための処置とドキュメントに記述されていました。

今回作成したビューでは、別スレッドでテーブルにレコードを追加しています。この別スレッドの処理はロールバックするトランザクションとは別なため、DBのデータがリセットされていないものと考えられます。

テストケース(解決)

DBのデータをリセットされていない問題を解決するには、TransactionTestCase(django.test.TransactionTestCase)を継承してテストケースを作成します。

TransactionTestCaseを継承したテストケースは、TestCaseとは違ってテストケースの最初でトランザクションを開始しません。各テストケースの最後にDBをリセットするときはflushコマンドを使っています。このコマンドで、MySQLなら各テーブルでTRUNCATEが実行されます。

テストケース作成

# coding: utf-8
import time
from django.test import TransactionTestCase, client
from django.core.urlresolvers import reverse
from myapp.models import Foo


class TestFooCreate2(TransactionTestCase):
    def test_get(self):
        self.assertEqual(Foo.objects.count(), 0)  # 初期状態の確認です。
        c = client.Client()
        c.get(reverse('myapp_foocreate'))  # テスト対象を実行します。
        time.sleep(1)  # 別スレッドの終了を待つため、1秒間スリープさせます。
        self.assertEqual(Foo.objects.count(), 1)  # テスト結果を確認します。

テストケース実行

$ ./manage.py test --keepdb myapp.tests.TestFooCreate2
.
----------------------------------------------------------------------
Ran 1 test in 1.295s

OK
$ ./manage.py test --keepdb myapp.tests.TestFooCreate2
.
----------------------------------------------------------------------
Ran 1 test in 1.295s

OK

TestCaseを継承したテストケースとは違い、何度実行してもテストは成功します。

※TestFooCreate1で消えなかったレコードが残っている場合は、そのレコードは削除しておかないとテストは失敗します。

備考

トランザクション分離レベルがSERIALIZEの場合

Webアプリではあまりないとは思いますが、トランザクション分離レベルをSERIALIZEにしていた場合、今回書いたテストケースの結果が変わります。

具体的には、SERIALIZEにしたことでファントムリードが抑制され、TestCaseを継承したテストケース(TestFooCreate1.test_get)は1回目から失敗します。

なお、TransactionTestCaseを継承したテストケース(TestFooCreate2.test_get)は、トランザクション分離レベルをSERIALIZABLEにしても成功します。

トランザクション分離レベルを変更

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': '...',
        'USER': '...',
        'PASSWORD': '...',
        'OPTIONS': {
            'init_command': 'SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE'
        },
    }
}

テストケース実行

TestCaseを継承したテストケースは失敗します。

$ ./manage.py test --keepdb myapp.tests.TestFooCreate1
F
======================================================================
FAIL: test_get (myapp.tests.TestFooCreate1)
    self.assertEqual(Foo.objects.count(), 1)  # テスト結果を確認します。
AssertionError: 0 != 1

----------------------------------------------------------------------
Ran 1 test in 1.092s

TransactionTestCaseを継承したテストケースは成功します。

$ ./manage.py test --keepdb myapp.tests.TestFooCreate2
.
----------------------------------------------------------------------
Ran 1 test in 1.199s

OK

この記事が役に立った場合、シェアしていただけると励みになります!!