Счетчик вызовов не работает с функцией asyn c - PullRequest
Новая
       43

Счетчик вызовов не работает с функцией asyn c

1 голос
/ 03 августа 2020

У меня есть функция с декоратором @retry, которая повторяет выполнение функции, если произошло определенное исключение. Я хочу проверить, что эта функция выполняет правильное количество раз, для чего у меня работает следующий код: 

@pytest.mark.asyncio
async def test_redis_failling(mocker):
    sleep_mock = mocker.patch.object(retry, '_sleep')
    with pytest.raises(ConnectionError):
        retry_store_redis()
    assert sleep_mock.call_count == 4

@retry(ConnectionError, initial_wait=2.0, attempts=5)
def retry_store_redis():
    raise ConnectionError()

Но, если я изменю retry_store_redis() на asyn c функцию , возвращаемое значение sleep_mock.call_count - 0.

1 Ответ

2 голосов
/ 03 августа 2020

Итак, вы определяете «повтор» как функцию. Затем вы определяете тест, затем вы определяете код, который использует @ retry.

@ retry в качестве декоратора вызывается во время импорта. Таким образом, порядок операций:

  1. объявить повтор
  2. объявить тест
  3. повторить вызов с retry_store_redis в качестве аргумента
  4. начать тест
  5. повторная попытка исправления
  6. вызов функции, которую вы определили на шаге 3

, поэтому «повтор» вызывается один раз (во время импорта), ваш макет вызывается ноль раз. Чтобы получить желаемое поведение (гарантируя, что повторная попытка фактически повторно вызывает базовую функцию), я бы сделал 

@pytest.mark.asyncio
async def test_redis_failling(mocker):
    fake_function = MagicMock(side_effect=ConnectionError)

    decorated_function = retry(ConnectionError, initial_wait=2.0, attempts=5)(fake_function)

    with pytest.raises(ConnectionError):
        decorated_function()
    assert fake_function.call_count == 4

, если бы вы хотели проверить это как построенное (вместо теста специально для декоратора) вам придется имитировать исходную функцию внутри декорированной функции, что будет зависеть от того, как вы реализовали декоратор. Способ по умолчанию (без каких-либо библиотек) означает, что вам нужно будет проверить атрибут «закрытие». Вы можете построить объект, чтобы сохранить ссылку на исходную функцию, хотя вот пример 

def wrap(func):
    class Wrapper:
        def __init__(self, func):
            self.func = func

        def __call__(self, *args, **kwargs):
            return self.func(*args, **kwargs)
    return Wrapper(func)


@wrap
def wrapped_func():
  return 42

в этом сценарии вы можете исправить обернутую функцию на wrapped_func.func

...