那么,我有一个flask视图,它将芹菜任务添加到队列中,并向用户返回200。
from flask.views import MethodView
from app.tasks import launch_task
class ExampleView(MethodView):
def post(self):
# Does some verification of the incoming request, if all good:
launch_task(task, arguments)
return 'Accepted', 200
问题是测试以下,我不想有一个芹菜实例等。我只是想知道,在所有的验证是ok的,它返回200给用户。芹菜launch_task()
将在其他地方进行测试。
因此,我热衷于模拟launch_task()
调用,因此本质上它什么也不做,使我的unittest独立于芹菜实例。
我试过各种形式的
@mock.patch('app.views.launch_task.delay'):
def test_launch_view(self, mock_launch_task):
mock_launch_task.return_value = None
# post a correct dictionary to the view
correct_data = {'correct': 'params'}
rs.self.app.post('/launch/', data=correct_data)
self.assertEqual(rs.status_code, 200)
@mock.patch('app.views.launch_task'):
def test_launch_view(self, mock_launch_task):
mock_launch_task.return_value = None
# post a correct dictionary to the view
correct_data = {'correct': 'params'}
rs.self.app.post('/launch/', data=correct_data)
self.assertEqual(rs.status_code, 200)
但是似乎不能让它工作,我的视图只是退出一个500错误。任何帮助将不胜感激!
我也尝试过任何@patch
装饰器,它没有工作我在setUp
中发现了mock:
import unittest
from mock import patch
from mock import MagicMock
class TestLaunchTask(unittest.TestCase):
def setUp(self):
self.patcher_1 = patch('app.views.launch_task')
mock_1 = self.patcher_1.start()
launch_task = MagicMock()
launch_task.as_string = MagicMock(return_value = 'test')
mock_1.return_value = launch_task
def tearDown(self):
self.patcher_1.stop()
@task
装饰器用Task
对象替换函数(参见文档)。如果您模拟任务本身,您将用MagicMock
替换(有点神奇的)Task
对象,并且它根本不会调度任务。相反,模拟Task
对象的run()
方法,如下所示:
# With CELERY_ALWAYS_EAGER=True
@patch('monitor.tasks.monitor_user.run')
def test_monitor_all(self, monitor_user):
"""
Test monitor.all task
"""
user = ApiUserFactory()
tasks.monitor_all.delay()
monitor_user.assert_called_once_with(user.key)