Eu estou tentando usar a biblioteca de simulação de Python para corrigir uma tarefa de aipo que é executada quando um modelo é salvo no meu aplicativo django, para ver que está sendo chamado corretamente.

Basicamente, a tarefa é definida dentromyapp.tasks, e é importado no topo do meu arquivo models.py assim:

from .tasks import mytask

... e, em seguida, é executado emsave() dentro do modelo usandomytask.delay(foo, bar). Até aí tudo bem - funciona bem quando estou realmente correndo Celeryd etc.

Eu quero construir um teste de unidade que zomba da tarefa, apenas para verificar se ele é chamado com os argumentos corretos e realmente não tenta executar a tarefa de aipo nunca.

Então, no arquivo de teste, eu tenho algo parecido com isso dentro de um TestCase padrão:

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)

... mas nunca é chamado / é sempre falso. Eu tentei várias encarnações (remendandomyapp.models.mytask em vez disso, e verificar semock_task.delay foi chamado em vez disso. Eu juntei dos documentos simulados que o caminho de importação é crucial, e googling me diz que este deve ser o caminho como é visto dentro do módulo em testes (o que seriamyapp.models.mytask.delay ao invés demyapp.tasks.mytask.delay, se eu entendi corretamente).

Onde estou indo errado aqui? Existe alguma dificuldade específica em consertar as tarefas do Aipo? Eu poderia corrigircelery.task (que é usado como decorador paramytask) em vez de?