Django ユニットテストでモックを使用して Celery タスクにパッチを当てる 質問する

Django ユニットテストでモックを使用して Celery タスクにパッチを当てる 質問する

私は、Python モック ライブラリを使用して、Django アプリでモデルが保存されるときに実行される Celery タスクにパッチを適用し、正しく呼び出されているかどうかを確認しようとしています。

基本的に、タスクは 内で定義されmyapp.tasks、次のように models.py ファイルの先頭にインポートされます。

from .tasks import mytask

...そして、save()を使用してモデル内で実行しますmytask.delay(foo, bar)。 ここまでは順調です。実際に Celeryd などを実行しているときにも問題なく動作します。

タスクをモックする単体テストを構築して、正しい引数で呼び出され、実際には Celery タスクを実行しようとしないことを確認したいと思います。

したがって、テスト ファイルでは、標準の TestCase 内に次のような内容が含まれています。

from mock import patch # at the top of the file

# ...then later
def test_celery_task(self):
    with patch('myapp.models.mytask.delay') as mock_task:
        # ...create an instance of the model and save it etc
        self.assertTrue(mock_task.called)

...しかし、これは呼び出されず、常に false になります。私はさまざまな実装を試しました (myapp.models.mytask代わりに にパッチを適用し、mock_task.delayが呼び出されたかどうかを確認しました。モックのドキュメントから、インポート パスが重要であることがわかりました。また、Google で検索すると、テスト対象のモジュール内で見られるパスが である必要があることがわかりました (正しく理解していれば、myapp.models.mytask.delayではなく にmyapp.tasks.mytask.delayなります)。

どこが間違っているのでしょうか? Celery タスクにパッチを適用するには、何か特別な困難があるのでしょうか?代わりにcelery.task(デコレータとして使用されるmytask) にパッチを適用できますか?

ベストアンサー1

あなたが抱えている問題は、これが Celery タスクであるという事実とは無関係です。間違ったものをパッチしているだけです。;)

具体的には、どのビューまたは他のファイルが「mytask」をインポートしているかを調べて、そこにパッチを適用する必要があります。関連する行は次のようになります。

with patch('myapp.myview.mytask.delay') as mock_task:

これにはさらにいくつかの特徴があります:

http://www.voidspace.org.uk/python/mock/patch.html#パッチを当てる場所

おすすめ記事