Using python stdlib urllib to make requests and unit testing them

example.py

import json
from http import HTTPStatus
from urllib.request import HTTPError, Request, urlopen


def read_response_json(response):
    if "application/json" in response.headers["Content-Type"]:
        return json.loads(response.read().decode("utf8"))


def example():
    """Make a HTTP call with urllib."""
    data = json.dumps(dict(hello="world")).encode("utf8")
    request = Request(
        url="https://httpbin.org/post",
        method="POST",
        data=data,
        headers={
            "Content-Type": "application/json",
        },
    )

    try:
        response = urlopen(request)
    except HTTPError as exc:
        # HTTPError is a subclass of HTTPResponse, which can be useful
        response = exc
        
    return response, read_response_json(response)


if __name__ == "__main__":
    response, body = example()
    print(json.dumps(body, indent=4))

tests.py

# unit testing - the hard bit.
#
# HTTPResponse very not-sans-io, and requires a socket object to read from.
# Below is an example of how you might set up a mock response to a urlopen
# call, but still get a valid HTTPResponse object to use.
import io
import json
from collections import defaultdict
from contextlib import contextmanager
from datetime import datetime
from http import HTTPStatus, client
from unittest import mock

import example


class MockSocket:
    """Minimal socket api as used by HTTPResponse"""

    def __init__(self, data):
        self.stream = io.BytesIO(data)

    def makefile(self, mode):
        return self.stream


def create_http_response(status=HTTPStatus.OK, headers={}, body=None, method=None):
    """Create a minimal HTTP 1.1 response byte-stream to be parsed by HTTPResponse."""
    lines = [f"HTTP/1.1 {status.value} {status.phrase}"]
    lines.append(f"Date: {datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S')}")
    lines.append("Server: Test")
    for name, value in headers.items():
        lines.append(f"{name}: {value}")

    if body:
        lines.append(f"Content-Length: {len(body)}")
        lines.append("")
        lines.append("")

    data = ("\r\n".join(lines)).encode("ascii")
    if body:
        data += body.encode("utf8")

    sock = MockSocket(data)

    # HTTPResponse accepts method parameters and uses it to enforce correct
    # HEAD response parsing.
    response = client.HTTPResponse(sock, method=method)

    # parse and validate response early, to detect error in test setup
    response.begin()
    return response


class UrlopenResponses:
    """Simple responses-like interface for mocking."""

    def __init__(self):
        self.responses = defaultdict(list)

    def add_response(
        self, url, method="GET", status=HTTPStatus.OK, headers={}, body=None
    ):
        response = create_http_response(status, headers, body, method)
        key = (method, url)
        self.responses[key].append(response)

    def urlopen(self, request):
        """Replacement urlopen function."""
        key = (request.method, request.full_url)
        if key in self.responses:
            return self.responses[key].pop()
        else:
            response_list = "\n".join(f"{m} {u}" for m, u in self.responses)
            raise RuntimeError(
                f"{self.__class__.__name__}: Could not find matching response for "
                f"{request.method} {request.full_url}\n"
                f"Current responses:\n{response_list}"
            )

    @contextmanager
    def patch(self, patch_location="urllib.request.urlopen"):
        with mock.patch(patch_location, self.urlopen):
            yield


def test_example():
    responses = UrlopenResponses()
    body = json.dumps(dict(json=dict(hello="world")))
    responses.add_response(
        url="https://httpbin.org/post",
        method="POST",
        headers={"Content-Type": "application/json"},
        body=body,
    )

    with responses.patch("example.urlopen"):
        response, body = example.example()

    assert response.status == 200
    # note headers are based on email.message.EmailMessage semantics, i.e.
    # case insenstive lookup of first header instance via getitem, use
    # get_all() to get multiple instances of a header
    assert response.headers["Content-Type"] == "application/json"
    assert body["json"] ==  {"hello": "world"}


def test_error():
    responses = UrlopenResponses()
    body = json.dumps(dict(success="false"))
    responses.add_response(
        url="https://httpbin.org/post",
        method="POST",
        status=HTTPStatus.INTERNAL_SERVER_ERROR,
        headers={"Content-Type": "application/json"},
        body=body,
    )

    with responses.patch("example.urlopen"):
        response, body = example.example()

    assert response.status == 500
    assert body == {"success": "false"}