単体テストのフラスコビューセロリのタスクを模擬して

Question

私は、セロリのタスクをキューに追加し、ユーザーに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エラーで終了します。どんな援助もありがとう！

Answer 1

私はまた、任意の@patchデコレータを試してみましたが、それが を動作しませんでしたそして、私は同じようsetUpでモックを見つけました：（documentationを参照）@taskデコレータはTaskオブジェクトに機能を置き換える

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

。あなたが自分自身を模擬するならば、のオブジェクトをMagicMockに置き換え、タスクをまったくスケジュールしません。代わりに、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)