Django BackGround Tasksのテスト方法を考える

前提

background　tasksについてはこちらをご参照いただけると良いと思います。

koty.hatenablog.com

ざっくり言うと非同期処理したいけどcelery入れるのはちょっと大変だしpipだけで完結すると楽だよねって時に使っています。
ここら辺の基本的なことは知っている前提で話しを進めます

環境

Python == 3.7
Django==2.2.6
django-background-tasks==1.2.0

処理を書く

こんな処理があったとします。

view.py

from django.http import HttpResponse
from app.tasks import honne


def index(request):

    param = request.GET.get('q', 'ヨシ！')

    if param == 'ヨシ！':
        honne('ヨシ！')
    else:
        honne('シらない')

    return HttpResponse('ヨシ！')

どんなリクエストが来てもクライアントにはヨシ！と返します

合わせてクエリパラメータを見て”ヨシ！”か”シらない”をデータベースに登録します。

from background_task import background

from app.models import Emotion


@background(queue='honnne', schedule=5)
def honne(emotion):

    Emotion.objects.create(emotion=emotion)

もちろん本音はクライアントに返す必要がないので非同期処理で 心の内 データベースに登録します。

テストをどうする？

テストを書きましょう

class TestYoshi(TestCase):

    def test(self):
        client = Client()
        # ヨシ！を送る
        result = client.get('/app/', dict(q='ヨシ！'))
        # Responseがヨシ！であることを確認する
        self.assertEqual(result.content.decode('utf-8'), 'ヨシ！')
        
        # 非同期処理を上げていないとモデルが作られない
        self.assertEqual(Emotion.objects.all().count(), 1)
        e = Emotion.objects.all().first()
        self.assertEqual(e.emotion, 'ヨシ')

テストを実行しましょう

Failure
Traceback (most recent call last):
  File "*****/DjangoBackgroundTasksTest/app/tests.py", line 12, in test
    self.assertEqual(Emotion.objects.all().count(), 1)
AssertionError: 0 != 1

非同期処理が実行されないのでモデルが登録されずに落ちました！

どうする？

裏でBackGroundTasksを上げる?

テストの度に別processを上げなければならない。 test用のデータベースはsettingsで指定されたものにtest_が付いたものになるので、それ用のsettingsを用意する必要がある。
→面倒だし個人的にあまりやりたくない方法なので割愛します。

個人的解決案

まず、tasks.pyを修正します。

@background(queue='honnne', schedule=5)
def honne(emotion: str):
    _honne(emotion) # 引数を渡すだけ


def _honne(emotion: str):
　#　今まで honnne でやっていた処理をそのまま
    Emotion.objects.create(emotion=emotion)

次にテストを直します

import json

from background_task.models import Task # 今回新たにimportしたところ
from django.test import TestCase, Client

    def test_resove(self):
        client = Client()
        result = client.get('/app/', dict(q='ヨシ！'))
        self.assertEqual(result.content.decode('utf-8'), 'ヨシ！')
        # 非同期処理を呼んだときは裏ではTaskというモデルにレコードが作られている
        tasks = Task.objects.all()
        self.assertEqual(tasks.count(), 1)
        params = json.loads(tasks[0].task_params)
        # 非同期処理を呼ぶ際のパラメーターが正しく渡っていることを確認する
        self.assertEqual(params[0][0],  'ヨシ！')

後はhonneの中身をテストします。

class TestHonne(TestCase):

    def test(self):
        # _honneの中のテストはここで行う。
        # 登録されていることを確認する
        _ = _honne('ヨシ！') # 同期処理なので普通に呼べる
        e = Emotion.objects.all().first()
        self.assertEqual(e.emotion, 'ヨシ！')

どうでしょう？　こうすれば非同期処理の中もテストできますし、テストを動かしたいときにいちいち別processを立ち上げる手間も省けます。

あくまで業務中に思いついたネタなのでもっとよりよい方法があったら(優しく)コメント頂ければと思います。

ソースはこちらになります。

github.com