Python Robust Requests

python_robust_requests.py

from dataclasses import dataclass
from typing import List

import pytest
import requests
from requests import Session
from requests.adapters import HTTPAdapter
from urllib3 import Retry


class APIRequestError(ValueError):
    pass


@dataclass(frozen=True)
class ADummyUser:
    id: int
    email: str
    first_name: str
    last_name: str
    avatar: str


class ADummyAPIService(object):
    endpoint_url: str = "https://reqres.in/api/users"
    request_session: Session = None

    def __init__(self):
        self.request_session = retry_session()
        # client.HTTPConnection.debuglevel = 1

    def get_all_users(self):
        try:
            response = self.request_session.get(self.endpoint_url, timeout=2)
        except Exception as x:
            print('It failed :(', x.__class__.__name__)
            raise APIRequestError
        else:
            return response.json()['data']


class ADummyDataGenerator(object):
    all_users: List[ADummyUser] = []

    def __init__(self, api_service):
        self.api_service = api_service
        for user in api_service.get_all_users():
            self.all_users.append(ADummyUser(**user))

    def get_total_users(self):
        return len(self.all_users)


def retry_session():
    retry_strategy = Retry(total=3,
                           backoff_factor=1,
                           status_forcelist=[429, 500, 502, 503, 504],
                           method_whitelist=frozenset(['GET', 'POST']))

    adapter = HTTPAdapter(max_retries=retry_strategy)

    session = Session()
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


_api_service = ADummyAPIService()
_generator = ADummyDataGenerator(_api_service)


@pytest.fixture
def a_dummy_api_service():
    return _api_service


@pytest.fixture
def a_dummy_generator():
    return _generator


def test_request_a_url_without_timeout_successfully():
    response = requests.get('https://postman-echo.com/status/200')
    assert response.status_code == 200


def test_request_a_url_with_infinite_timeout():
    response = requests.get('https://postman-echo.com/delay/10', timeout=None)
    assert response.status_code == 200


def test_request_a_url_with_a_long_timeout_successfully():
    response = requests.get('https://postman-echo.com/status/200', timeout=5)
    assert response.status_code == 200


def test_request_a_url_with_a_very_short_timeout_and_fails():
    with pytest.raises(requests.exceptions.Timeout):
        requests.get('https://postman-echo.com/status/200', timeout=0.001)


def test_retry_session_a_successful_url_and_success():
    session = retry_session()
    response = session.get("https://postman-echo.com/status/200")
    assert response.status_code == 200


def test_retry_session_a_non_existing_url_n_times_and_fails(caplog):
    session = retry_session()
    with pytest.raises(requests.exceptions.ConnectionError):
        session.get("http://this-url-does-not-exist.bar")

    assert "Retry(total=0, connect=None, read=None, redirect=None, status=None)" in caplog.records[2].message
    assert "Retry(total=1, connect=None, read=None, redirect=None, status=None)" in caplog.records[1].message
    assert "Retry(total=2, connect=None, read=None, redirect=None, status=None)" in caplog.records[0].message


def test_get_a_url_with_response_status_500(a_dummy_api_service):
    endpoint_url = "https://postman-echo.com/status/500"

    with pytest.raises(requests.exceptions.RetryError):
        a_dummy_api_service.request_session.get(endpoint_url, timeout=2)


def test_get_all_users_successfully(a_dummy_generator):
    assert a_dummy_generator.get_total_users() == 6