pytest で celery のタスクを monkeypatch する方法

• 概要
• 環境
• タスク
• メイン
• テストコード
• ポイント
• 参考サイト

概要

pytest の monkeypatch を使って celery の各タスクにパッチを当てる方法を紹介します
これを行うことでタスクのプロセスが実行していない状態でもテストすることができます

環境

• macOS 10.15.6
• Python 3.8.5
  
  • celery 4.4.7

タスク

import time
from celery import Celery

app = Celery('sub_tasks', backend='redis://localhost', broker='redis://localhost')

@app.task
def add(x, y):
    return x + y

メイン

from sub_tasks import add

def main():
    ret = add.delay(100, 1).get()
    return ret

テストコード

import pytest
from celery import Celery

from main import main
from sub_tasks import add, app

class DummyAsyncResult():
    def get(self):
        return 100

def test_main(monkeypatch):
    monkeypatch.setattr('sub_tasks.add.delay', lambda x,y: DummyAsyncResult())
    result = main()
    assert (result == 100)

ポイント

• delay を monkeypatch すること
• delay の返り値をダミーの AsyncResult にすること

参考サイト

• https://docs.celeryproject.org/en/stable/userguide/testing.html