Làm thế nào để kiểm tra Thử lại trong ứng dụng Celery bằng Python?

Question

Tôi đang cố kiểm tra xem ứng dụng có đang thử lại không.

@celery.task(bind=False, default_retry_delay=30) 
def convert_video(gif_url, webhook): 
    // doing something 
    VideoManager().convert(gif_url) 
    return 
    except Exception as exc: 
     raise convert_video.retry(exc=exc) 

Và tôi chế giễu các thử nghiệm

@patch('src.video_manager.VideoManager.convert') 
@patch('requests.post') 
def test_retry_failed_task(self, mock_video_manager, mock_requests): 
    mock_video_manager.return_value= {'webm':'file.webm', 'mp4':'file.mp4', 'ogv' : 'file.ogv', 'snapshot':'snapshot.png'} 
    mock_video_manager.side_effect = Exception('some error') 

    server.convert_video.retry = MagicMock() 

    server.convert_video('gif_url', 'http://www.company.com/webhook?attachment_id=1234') 
    server.convert_video.retry.assert_called_with(ANY) 

Và tôi nhận được lỗi này

TypeError: exceptions must be old-style classes or derived from BaseException, not MagicMock

Đó là hiển nhiên, nhưng tôi không biết làm thế nào để làm điều đó bằng cách khác để kiểm tra nếu phương thức đang được gọi.

Answer 1

Tôi không làm cho nó hoạt động chỉ bằng cách sử dụng tính năng thử đã được tạo sẵn vì vậy tôi phải sử dụng một mô hình có tác dụng phụ của Thử lại thực, điều này giúp bạn có thể bắt nó trong thử nghiệm. tôi đã thực hiện nó như thế này:

from celery.exceptions import Retry 
from mock import MagicMock 
from nose.plugins.attrib import attr 

# Set it for for every task-call (or per task below with @patch) 
task.retry = MagicMock(side_effect=Retry) 

#@patch('task.retry', MagicMock(side_effect=Retry) 
def test_task(self): 
    with assert_raises(Retry): 
     task() # Note, no delay or things like that 

# and the task, I don't know if it works without bind. 
@Celery.task(bind=True) 
def task(self): 
    raise self.retry() 

Nếu có ai biết làm thế nào tôi có thể thoát khỏi bước bổ sung trong chế giễu Retry "ngoại lệ" Tôi rất vui khi được nghe nó!

Answer 2

from mock import patch 

import pytest  


@patch('tasks.convert_video.retry') 
@patch('tasks.VideoManager') 
def test_retry_on_exception(mock_video_manger, mock_retry): 
    mock_video_manger.convert.side_effect = error = Exception() 
    with pytest.raises(Exception): 
     tasks.convert_video('foo', 'bar') 
    mock_retry.assert_called_with(exc=error) 

bạn cũng đang thiếu một số nội dung trong nhiệm vụ của bạn:

@celery.task(bind=False, default_retry_delay=30) 
def convert_video(gif_url, webhook): 
    try: 
     return VideoManager().convert(gif_url) 
    except Exception as exc: 
     convert_video.retry(exc=exc)