django - Python: Mock doesn't work inside celery task -

this question has answer here:

• why python mock patch doesn't work? 2 answers

i want use python mock library testing django application sends email.

test code:

# tests.py django.test import testcase  class mytestcase(testcase):      @mock.patch('django.core.mail.mail_managers')     def test_canceled_wo_claiming(self, mocked_mail_managers):         client = client()         client.get('/')         print(mocked_mail_managers.called)         mocked_mail_managers.assert_called_with('hi, managers!', 'message body') 

first example - without tasks

# views.py django.views.generic import view django.core.mail import mail_managers  class myview(view):      def get(self, request):         mail_managers('hi, managers!', 'message body')         return httpresponse('hello!') 

second example - tasks

# views.py django.views.generic import view . import tasks  class myview(view):     def get(self, request):         tasks.notify.apply_async()         return httpresponse('hello!')   # tasks.py celery import shared_task django.core.mail import mail_managers  @shared_task def notify():     mail_managers('hi, managers!', 'message body') 

the first example works normal, second example fails, not called exception.

my settings:

# celery broken_url = 'memory://' broker_backend = 'memory'  celery_always_eager = true celery_eager_propagates_exceptions = true test_runner = 'djcelery.contrib.test_runner.celerytestsuiterunner' 

is possible perform such integrated test or way solve problem split test two?

testing behavior of asynchronous tasks directly code uses them can tricky. 1 reason test may execute assertions before task runs, give false positives. in cases break testing 2 steps:

1. mocking task , testing called when has called , expected arguments.
2. test task standalone function , executing normal function, i.e. without need of celery server.

to illustrate, like:

# views.py path.to.tasks import my_task   def my_view(requtest):     # stuff     my_task.delay('foo', 'bar')     return httpresponse('whatever')   # test_my_task.py views import my_view path.to.tasks import my_task   class mytest(testcase):     @mock.patch('path.to.tasks.my_task')     def test_my_task_is_called(self, mocked_task):         client = client()         client.get('/')         my_task.assert_called_with('foo', 'bar')      def test_my_task_works(self):         my_task('foo', 'bar')  # note don't use .delay(...), .apply_async(...), etc         assert 'my task did expected do' 

this way can test implementing code behaves correctly in regards task , task behaves correctly once it's called expected.

i hope it's useful! :)