CODEWITHSUNDEEP
pythonpython-3.xpytest
Punter Vicky
Punter Vicky

Reputation: 16992

Unit Test Retry functionality provided by Python

I am using the code below to perform exponential retries. I have set the reties to 5. As part of my unit test, I am mocking the response and returning status code 503. However I do not see 5 retries being performed when I use this unit test. What changes should I make to my unit test to validate that session.get has retried 5 times?

try:
    max_retries = Retry(total=retries, backoff_factor=5, status_forcelist=[500, 502, 503, 504, 429])
    self.session.mount('https://', HTTPAdapter(max_retries=max_retries))
    # Perform GET request
    response = self.session.get(url, verify=verify)
except Exception as e:
    print(f" Exception occured {e}")
return response

Unit Test

def test_my_function_retries(self):    
    responses.add(responses.GET,
                  'https://requesturl',
                  json={}, status=503)
    request.get()

Upvotes: 7

Views: 6028

Answers (4)

Enrico Marchesin
Enrico Marchesin

Reputation: 5150

I usually prefer not to make actual calls out to the internet in my tests: the remote service could be down or you might need to run your tests offline. Most importantly you want to run your tests FAST, and the network call can slow them down significantly.

I also want to make sure that the retry logic actually makes the retries I expect, and that it can actually succeeds eventually. And do not like the idea of using log messages as a proxy for that.

I could not find a proper solution online, so I'll leave my solution here for posterity, trying to save someone else the hours I've spent trying:

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),
    ]

(Note: if you are making several calls to this method, then you might want to initialize the session object only once, not every time you make a request!)

Upvotes: 1

orjames
orjames

Reputation: 71

I was able to test the urllib3.util.retry Retry package was working with the following pytest code:

# define retrying strategy
retries = Retry(
            total=5,
            backoff_factor=0.4,
            status_forcelist=[429, 500, 501, 502, 503, 504, 505, 506, 507, 509, 510, 511],
)
# define the session
session = requests.Session()
# Add retries to requests to all https:// addresses
session.mount("https://", HTTPAdapter(max_retries=self.retries))

def test_retry_session(caplog):
    with pytest.raises(requests.exceptions.ConnectionError):
        session.get("https://postman-echo.com/status/200")

    assert "Retry(total=0, " in caplog.records[4].message
    assert "Retry(total=1, " in caplog.records[3].message
    assert "Retry(total=2, " in caplog.records[2].message
    assert "Retry(total=3, " in caplog.records[1].message
    assert "Retry(total=4, " in caplog.records[0].message

Upvotes: 0

Punter Vicky
Punter Vicky

Reputation: 16992

I was able to use httpretty to test the retries

@httpretty.activate
def test_my_function(self):
    httpretty.register_uri(
        httpretty.GET,
        url,           
        responses=[
            httpretty.Response(
                body='{}',
                status=503,
            ),
        ]
    )

    my_class = MyClass()
    my_class.my_function()
    print(len(httpretty.latest_requests()))

Upvotes: 1

Tomer Cohen
Tomer Cohen

Reputation: 332

I'm not familiar with retry library you use but it looks like what you need can be found in this library:

https://pypi.org/project/retry/

when I use retries I just extract to functionality I want to perform the retry on to a different method and use the retry decorator like this:

from retry import retry

@retry(exceptions=IllegalStateException, tries=2, delay=5)
def flaky_function():
    ...

It also keeps the code much cleaner

Upvotes: 0

Related Questions