AssertionError при модульном тестировании задачи сельдерея с автоповтором, откатом и джиттером - PullRequest
1 голос
/ 08 мая 2020

Использование сельдерея 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

Я так понимаю, это исключение. Я заменил ObjectDoesNotExist везде на Exception. После запуска теста я получаю следующую ошибку:

    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)

Пожалуйста, дайте мне знать, как я могу исправить обе ошибки.

...