#Mocking Celery objects issue.

1 messages · Page 1 of 1 (latest)

copper ledge
#

Hello, everyone.
I desperately need a help with my attempts to mock celery object.
So, here is more context.

  1. in my project I have a Celery functionality, final point of which is this piece of a code on a view level
def create(self, request, *args, **kwargs):
    response = super().create(request, *args, **kwargs)
    overdue_borrowing_notifications.delay()
    return response
  1. and I am trying to mock Celery object within my test function:
@patch("celery.app.task.Task.delay", return_value=1)
@patch("celery.app.task.Task.apply_async", return_value=1)
def test_borrowing_create_not_allowed_if_zero_inventory(self, *args, **kwargs):
    book = self.book_1
    self.borrowing_1 = Borrowing.objects.create(
        expected_return_date=datetime.now().date() + timedelta(days=25),
        book=book,
        user=self.user
    )

    book.inventory -= 1

    book.save()

    book.refresh_from_db()

    self.borrowing_2 = Borrowing.objects.create(
        expected_return_date=datetime.now().date() + timedelta(days=30),
        book=book,
        user=self.user2
    )

    book.inventory -= 1

    book.save()

    book.refresh_from_db()

    payload = {
        "expected_return_date": datetime.now().date() + timedelta(days=10),
        "book": book.id,
        "user": self.user3.id
    }

    response = self.client.post(BORROWING_URL, payload)

    self.assertEquals(response.status_code, status.HTTP_400_BAD_REQUEST)
    self.assertEquals(
        response.data["non_field_errors"][0],
        "There are no books in inventory to borrow."
    )
#

As you can see I created 2 Mock objects by usibng patches and obviously it must work properly.
But in fact it failed! It showed the following error message:

raise ConnectionError(str(exc)) from exc
kombu.exceptions.OperationalError: [WinError 10061] No connection could be made because the target machine actively refused it

And btw, I don’t want to run celery while doing testing – it should mock Celery objects properly. As was desired.
What I really did as of now – tried to improve code in this way:

@patch("celery.app.task.Task.delay", return_value=1)
@patch("celery.app.task.Task.apply_async", return_value=1)
def test_borrowing_create_not_allowed_if_zero_inventory(self, mock_delay, mock_apply_async):
    # Create a mock AsyncResult object
    mock_result = MagicMock()
    mock_result.id = "mocked_id"
    mock_delay.return_value = mock_result
    mock_apply_async.return_value = mock_result

    book = self.book_1
    self.borrowing_1 = Borrowing.objects.create(
        expected_return_date=datetime.now().date() + timedelta(days=25),
        book=book,
        user=self.user
    )

    book.inventory -= 1

    book.save()

    book.refresh_from_db()

    self.borrowing_2 = Borrowing.objects.create(
        expected_return_date=datetime.now().date() + timedelta(days=30),
        book=book,
        user=self.user2
    )

    book.inventory -= 1

    book.save()

    book.refresh_from_db()

    payload = {
        "expected_return_date": datetime.now().date() + timedelta(days=10),
        "book": book.id,
        "user": self.user3.id
    }

    response = self.client.post(BORROWING_URL, payload)

    self.assertEquals(response.status_code, status.HTTP_400_BAD_REQUEST)
    self.assertEquals(
        response.data["non_field_errors"][0],
        "There are no books in inventory to borrow."
    )
#

Unfortunately, the same mistake.
Print outputs showed the following:

print("Patching Task.delay:", mock_delay) -> <MagicMock name='apply_async' id='1913129939488'>
print("Patching Task.apply_async:", mock_apply_async) -> <MagicMock name='delay' id='1913132075568'>
#

So, how can I solve this issue effectively? Please help me whomever can really do it. Thank you in advance.

opaque sun
#

You typically don't patch celery at all. Typically you enable eager_mode which can be done via the setting CELERY_TASK_ALWAYS_EAGER which emulates the worker and calls the task function normally.

Your test can remain the same with the exception of removing the mocking code

copper ledge
#

I mean as far as I understood you I should remove my patches at all and just add ELERY_TASK_ALWAYS_EAGER=True to settings.py?

#

alonside with CELERY_TASK_EAGER_PROPAGATES = True ?

copper ledge
opaque sun
#

Can you share the current test and your celery settings

copper ledge
copper ledge