单元测试请求重试python

Question

如果端点在返回结果之前超时，我将尝试多次重试请求。这是代码：

def retry_request(self, params, max_retries=3):
    for i in xrange(max_retries):
        try:
            response = requests.get(params)
            break
        except requests.exceptions.Timeout as e:
            raise e

我想对重试进行单元测试以显示重试逻辑有效。有什么想法吗？

Answer 1

我通常不喜欢在测试中实际调用互联网：远程服务可能会关闭，或者您可能需要离线运行测试。最重要的是，您希望快速运行测试，而网络调用会显着减慢测试速度。

我还想确保重试逻辑能够实现我期望的重试，并且最终能够真正成功。

我尝试自己编写一个测试，但我很挣扎。我询问了互联网，但找不到任何可以满足我要求的东西。我深入研究了它的魔法世界urllib3，终于弄明白了它的真相，但这花了我一段时间。

由于这篇文章是在搜索时出现的，我将把我的解决方案留在这里供后代使用，试图节省其他人我花在尝试上的时间：

import urllib3
from http.client import HTTPMessage
from unittest.mock import ANY, Mock, patch, call

import requests


def request_with_retry(*args, **kwargs):
    session = requests.Session()
    adapter = requests.adapters.HTTPAdapter(max_retries=urllib3.Retry(
        raise_on_status=False,
        total=kwargs.pop("max_retries", 3),
        status_forcelist=[429, 500, 502, 503, 504],  # The HTTP response codes to retry on
        allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS"],  # The HTTP methods to retry on
    ))
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session.request(*args, **kwargs)


@patch("urllib3.connectionpool.HTTPConnectionPool._get_conn")
def test_retry_request(getconn_mock):
    getconn_mock.return_value.getresponse.side_effect = [
        Mock(status=500, msg=HTTPMessage()),
        Mock(status=429, msg=HTTPMessage()),
        Mock(status=200, msg=HTTPMessage()),
    ]

    r = request_with_retry("GET", "http://any.url/testme", max_retries=2)
    r.raise_for_status()

    assert getconn_mock.return_value.request.mock_calls == [
        call("GET", "/testme", body=None, headers=ANY),
        call("GET", "/testme", body=None, headers=ANY),
        call("GET", "/testme", body=None, headers=ANY),
    ]

（注意：如果您多次调用此方法，那么您可能只想初始化会话对象一次，而不是每次发出请求时都初始化！）

Answer 2

requests 库有自己的重试逻辑。

尝试这样的事情：

from requests import Session, exceptions
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from unittest import TestCase
import logging


class TestRequests(TestCase):

    def test_retries(self):
        session = Session()
        retries = Retry(total=5,
                        backoff_factor=0.1,
                        status_forcelist=[500, 502, 503, 504, 429],
                        method_whitelist=frozenset(['GET', 'POST']))

        session.mount(
            'https://', HTTPAdapter(max_retries=retries,
                                    pool_maxsize=30))

        try:
            result = session.get('https://httpbin.org/status/500',
                                 stream=True,
                                 timeout=2)
            print(result)
        except exceptions.HTTPError as e:
            logging.error('http', exc_info=e)
        except exceptions.RetryError as e:
            logging.error('retry', exc_info=e)