Testing views with pytest, how to mock celery task

Viewed 1109

I faced a problem while mocking celery task. Hope someone will be able to help me

conftest.py

import pytest

@pytest.fixture
def api_client():
   from rest_framework.test import APIClient
   return APIClient()

views.py

class SomeViewSet(ModelViewSet):

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer_class()(data=request.data)
        if serializer.is_valid():
            serializer.save()
            my_celery_task.delay(serializer.instance.pk)
            return Response(serializer.data, status=status.HTTP_201_CREATED)

tests.py

@patch('requests.post')
@pytest.mark.django_db
def test_create(api_client):    
    response = api_client.post(reverse('withdraw-act-list'), data=request_data, format='json')
    assert response.status_code == 201

I have tried severals @patch('requests.post') @patch.object(my_celery_task, 'delay'), but everytime got AssertionError: assert <MagicMock name='post.post().status_code' id='140584951434880'> == 201

How to fix this? Is there a way to mock the celery task and get valid response in the test

2 Answers

Most likely Celery runs in another thread and your mock in current thread is useless, you need to enable synchronous operations by this option in your test environment:

CELERY_ALWAYS_EAGER = True

and possibly this to get exceptions:

CELERY_EAGER_PROPAGATES = True

Decision

@patch.object(my_celery_task, 'delay')
@pytest.mark.django_db
def test_create(mock_delay, api_client):    
    response = api_client.post(reverse('withdraw-act-list'), data=request_data, format='json')
    assert response.status_code == 201

I should pass mock_delay as parameter into my test function (as first parameter, it's important)

Related