test_requests.py

#!/usr/bin/env python3

import http.server
import requests
import socketserver
import threading

from requests.adapters import HTTPAdapter

# Set this variable to cause the assert to either pass or fail:
PLEASE_WORK = False

if PLEASE_WORK:
    from requests.packages import urllib3
else:
    from requests.packages.urllib3.util import Retry


def main():
    # Create a ThreadingTCPServer that will send one 500 response then a
    # 200 response:
    server = socketserver.ThreadingTCPServer(
        ("localhost", 0), Handler)
    server_address = server.server_address[0]
    listen_port = server.server_address[1]
    server_url = 'http://%s:%s' % (server_address, listen_port)
    server_thread = threading.Thread(target=server.serve_forever)
    # Exit the server thread when the main thread terminates
    server_thread.daemon = True
    server_thread.start()

    # Create a requests session with a Retry configured to retry 500 responses
    # twice:
    session = requests.Session()
    if PLEASE_WORK:
        retries = urllib3.util.Retry(
            total=2,
            method_whitelist=['GET', 'POST', 'PATCH'],
            status_forcelist=[500, 502, 503, 504]
        )
    else:
        retries = Retry(
            total=2,
            method_whitelist=['GET', 'POST', 'PATCH'],
            status_forcelist=[500, 502, 503, 504]
        )
    retryable_adapter = HTTPAdapter(max_retries=retries)
    session.mount('http://', retryable_adapter)
    response = session.get(server_url)

    assert response.status_code == 200, response

    # Clean up thread:
    server.shutdown()
    server.server_close()
    server_thread.join()


class Handler(http.server.BaseHTTPRequestHandler):

    call_count = 0

    def do_GET(self):
        if Handler.call_count < 1:
            Handler.call_count += 1
            self.send_error(500, "Oops, something went wrong!")
        else:
            response = b"Hello Requests"
            self.send_response(200)
            self.send_header("Content-Length", str(len(response)))
            self.end_headers()
            self.wfile.write(response)


if __name__ == '__main__':
    main()