celery Django task is not working in mock.patch how to get updated value of DB which is being performed in the celery task

Question

I am working on django app, I made a service file which is called service.py and in that file i am calling a celery task i.e.

# start celery task in the apps>myapp>tasks.py
@app.task()
def update_user(user_id):
    try:
        User.objects.filter(id=user_id).update(first_name="")
    except Exception as e:
        logging.exception(e)
# end celery task in the apps>myapp>tasks.py




# start  test class in which i am calling celery task apps>myapp>service.py
from config.celery import app

class TestClass:
    def call_celery_task(self):
        app.send_task(
            "apps.myapp.tasks.update_user",
            [1],
        )
# end test class in which i am calling celery task apps>myapp>service.py

this task is working well when i execute it normally Via code with command TestClass.call_celery_task()

but getting issue with test cases here i am writing test cases and in that i want to call my celery task and want to fetch updated value of DB of celery task i am doing this way i.e.

from unittest import mock
from django.test import TestCase
from django.contrib.auth.models import User


class TestTestClass(TestCase):
    def setUp(self):
        self.user = User.objects.create(username="test",first_name="test")


    @mock.patch("apps.myapp.service.app")
    def test_call_celery_task(self, mock_app):
        mock_app.send_task.assert_called_with(
            "apps.myapp.tasks.update_user",
            [self.user.id],
        )
        self.assertTrue(mock_app.send_task.called) #this is returning True 
        self.user.refresh_from_db()
        self.assertEqual(self.user.first_name, "") #this is returning old first name why ?

here I am not getting updated first name value which job is done by celery task. if this is an asynchronous call then pls suggest me to get the updated value here thanks.

I don't want to manually call the celery task in the test case. I just only want to use the mock.patch.

Answer 1

Here i have got solution for this which is in above question was was mocked the celery app which is wrong what you have to do is you just have to mock only the task in your function. just import the celery task - from apps.myapp.tasks import update_user

from django.test import TestCase
from django.contrib.auth.models import User
from apps.myapp.tasks import update_user


class TestTestClass(TestCase):
    def setUp(self):
        self.user = User.objects.create(username="test",first_name="test")


    @mock.patch("apps.myapp.tasks.update_user")
    def test_call_celery_task(self, mock_app):
        update_user.apply(args=(what ever args you have)).get()
        self.user.refresh_from_db()
        self.assertEqual(self.user.first_name,"")
        #here you will get updated value of user
        

so here's what I did in my test case I just did my all job related test case and when my task comes into the picture i mocked it and called it synchronously. and it works Thanks.