Unit test Flask view mocking out celery tasks

Question

So, I have a flask view, which adds a celery task to a queue, and returns a 200 to the user.

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

The issue is with testing the following, I don't want to have to have a celery instance etc. etc. I just want to know that after all the verification is ok, it returns 200 to the user. The celery launch_task() will be tested elsewhere.

Therefore I'm keen to mock out that launch_task() call so essentially it does nothing, making my unittest independent of the celery instance.

I've tried various incarnations of:

@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)

But can't seem to get it to work, my view just exits with a 500 error. Any assistance would be appreciated!

Answer 1

I tried also any @patch decorator and it didn't work And I found mock in setUp like:

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()

Answer 2

The @task decorator replaces the function with a Task object (see documentation). If you mock the task itself you'll replace the (somewhat magic) Task object with a MagicMock and it won't schedule the task at all. Instead mock the Task object's run() method, like so:

# 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)