使用自动重试、退避和抖动对 celery 任务进行单元测试时出现 AssertionError

Question

使用 celery 4.3.0。我尝试为以下任务编写单元测试。

    from django.core.exceptions import ObjectDoesNotExist

    @shared_task(autoretry_for=(ObjectDoesNotExist,), max_retries=5, retry_backoff=10)
    def process_something(data):
        product = Product()
        product.process(data)

单元测试：

@mock.patch('proj.tasks.Product')
@mock.patch('proj.tasks.process_something.retry') 
def test_process_something_retry_failed_task(self, process_something_retry, mock_product):

    mock_object = mock.MagicMock()
    mock_product.return_value = mock_object
    mock_object.process.side_effect = error = ObjectDoesNotExist()

    with pytest.raises(ObjectDoesNotExist):
        process_something(self.data)

    process_something_retry.assert_called_with(exc=error)

这是我运行测试后得到的错误：

    @wraps(task.run)
    def run(*args, **kwargs):
        try:
            return task._orig_run(*args, **kwargs)
        except autoretry_for as exc:
            if retry_backoff:
                retry_kwargs['countdown'] = \
                    get_exponential_backoff_interval(
                        factor=retry_backoff,
                        retries=task.request.retries,
                        maximum=retry_backoff_max,
                        full_jitter=retry_jitter)
>           raise task.retry(exc=exc, **retry_kwargs)
E           TypeError: exceptions must derive from BaseException

我知道这是因为异常。我用Exception 替换了所有地方的ObjectDoesNotExist。运行测试后，我收到此错误：

    def assert_called_with(self, /, *args, **kwargs):
        """assert that the last call was made with the specified arguments.

            Raises an AssertionError if the args and keyword args passed in are
            different to the last call to the mock."""
        if self.call_args is None:
            expected = self._format_mock_call_signature(args, kwargs)
            actual = 'not called.'
            error_message = ('expected call not found.\nExpected: %s\nActual: %s'
                    % (expected, actual))
            raise AssertionError(error_message)

        def _error_message():
            msg = self._format_mock_failure_message(args, kwargs)
            return msg
        expected = self._call_matcher((args, kwargs))
        actual = self._call_matcher(self.call_args)
        if expected != actual:
            cause = expected if isinstance(expected, Exception) else None
>           raise AssertionError(_error_message()) from cause
E           AssertionError: expected call not found.
E           Expected: retry(exc=Exception())
E           Actual: retry(exc=Exception(), countdown=7)

请告诉我如何解决这两个错误。

Answer 1

我在进行测试以确保 celery 重试逻辑覆盖我的特定场景时遇到了类似的问题。对我有用的是使用explicit retry 而不是autoretry_for 参数。

我已根据我的解决方案调整了您的代码。虽然我的解决方案没有使用 shared_task 我认为它应该也可以工作。测试于celery==5.1.2

任务：

from django.core.exceptions import ObjectDoesNotExist

@shared_task(bind=True, max_retries=5, retry_backoff=10)
def process_something(self, data):
   try:
      product = Product()
      product.process(data)
   except ObjectDoesNotExist as exc:
      raise self.retry(exc=exc)

测试：

from proj.tasks import Product # I assume the Product class is located here
from django.core.exceptions import ObjectDoesNotExist
import celery

@mock.patch.object(Product, "__init__", Mock(return_value=None)) # just mocking the init method
@mock.patch.object(Product, "process")
@mock.patch('proj.tasks.process_something.retry')
def test_process_something_retry_failed_task(self, retry_mock, process_mock):
   exc = ObjectDoesNotExist()
   process_mock.side_effect = exc
   retry_mock.side_effect = celery.exceptions.Retry
   
   with pytest.raises(celery.exceptions.Retry):
      process_something(self.data)

   retry_mock.assert_called_with(exc=exc)

在我的问题中，我还使用了自定义异常。有了这个解决方案，我不需要更改我的异常类型。