# 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"}