問題描述

我正在嘗試使用 python 模擬庫來修補在我的 django 應用程序中保存模型時運行的 Celery 任務，以查看它是否被正確調用.

I'm trying to use the python mock library to patch a Celery task that is run when a model is saved in my django app, to see that it's being called correctly.

基本上，任務是在 myapp.tasks 中定義的，并像這樣在我的 models.py 文件的頂部導入:

Basically, the task is defined inside myapp.tasks, and is imported at the top of my models.py-file like so:

from .tasks import mytask

...然后使用 mytask.delay(foo, bar) 在模型內部的 save() 上運行.到目前為止一切順利 - 當我實際運行 Celeryd 等時效果很好.

...and then runs on save() inside the model using mytask.delay(foo, bar). So far so good - works out fine when I'm actually running Celeryd etc.

我想構建一個模擬任務的單元測試，只是為了檢查它是否被正確的參數調用，并且實際上并沒有嘗試運行 Celery 任務.

I want to construct a unit test that mocks the task, just to check that it gets called with the correct arguments, and doesn't actually try to run the Celery task ever.

所以在測試文件中，我在標準 TestCase 中有這樣的內容:

So in the test file, I've got something like this inside of a standard 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)

...但它永遠不會被調用/總是錯誤的.我嘗試了各種化身(修補 myapp.models.mytask ，并檢查是否調用了 mock_task.delay .我從模擬文檔中收集到導入路徑至關重要，谷歌搜索告訴我它應該是在被測模塊內部看到的路徑(應該是 myapp.models.mytask.delay 而不是 myapp.tasks.mytask.delay，如果我理解正確的話).

...but it never gets called/is always false. I've tried various incarnations (patching myapp.models.mytask instead, and checking if mock_task.delay was called instead. I've gathered from the mock docs that the import path is crucial, and googling tells me that it should be the path as it is seen inside the module under tests (which would be myapp.models.mytask.delay rather than myapp.tasks.mytask.delay, if I understand it correctly).

我哪里錯了?修補 Celery 任務是否有一些特定的困難?我可以修補 celery.task(用作 mytask 的裝飾器)嗎?

Where am I going wrong here? Is there some specific difficulties in patching Celery tasks? Could I patch celery.task (which is used as a decorator to mytask) instead?

推薦答案

您遇到的問題與這是 Celery 任務這一事實無關.你只是碰巧修補了錯誤的東西.;)

The issue that you are having is unrelated to the fact that this is a Celery task. You just happen to be patching the wrong thing. ;)

具體來說，您需要找出哪個視圖或其他文件正在導入mytask"并在那里對其進行修補，因此相關行如下所示:

Specifically, you need to find out which view or other file is importing "mytask" and patch it over there, so the relevant line would look like this:

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

這里有更多的味道:

http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch

這篇關于在 Django 單元測試中使用 mock 修補 celery 任務的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，也希望大家多多支持html5模板網！