celenicDjango任务在mock中不起作用.了解如何获取在celenic任务中执行的DB的更新值

Question

我正在开发django应用程序，我制作了一个名为service.py的服务文件，在该文件中我调用了一个芹菜任务，即

# 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

当我正常执行该任务时，该任务运行良好。通过使用命令TestClass.call_celery_task((的代码

但是遇到测试用例的问题，我正在编写测试用例，因为我想调用我的芹菜任务，并想获取芹菜任务的DB的更新值，我这样做，即

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 ?

在这里，我没有得到更新的名字值，这项工作是由芹菜任务完成的。如果这是一个异步调用，那么请建议我在这里获得更新的值，谢谢。

我不想在测试用例中手动调用芹菜任务。我只想使用mock.patch.

Answer 1

以上测试的问题是您嘲笑了Celery应用程序。因此，当调用app.send_task时，它并没有执行实际任务。您甚至可以打印app.send_task来查看其值。此外，在芹菜任务中添加pdb.set_trace()，以验证在运行单元测试时是否未调用该方法。这就是为什么不更新名称的值。

Answer 2

这里我有这个问题的解决方案，上面的问题被芹菜应用程序嘲笑了，这是错误的，你要做的是你只需要嘲笑你函数中的任务。只需导入芹菜任务-从apps.myapp.tasks导入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

这就是我在测试用例中所做的，我只是做了所有与工作相关的测试用例，当我的任务出现在画面中时，我嘲笑它并同步调用它。而且很管用谢谢。