From c3c17e1e8e419b055b7ec10e22d4cdd0b4bfe521 Mon Sep 17 00:00:00 2001 From: Tiara Rodney Date: Sat, 6 Jun 2026 14:40:43 +0200 Subject: [PATCH] test: add http client unit tests --- .../unit/byteb4rb1e/utils/http/test_client.py | 217 ++++++++++++++++++ 1 file changed, 217 insertions(+) create mode 100644 tests/unit/byteb4rb1e/utils/http/test_client.py diff --git a/tests/unit/byteb4rb1e/utils/http/test_client.py b/tests/unit/byteb4rb1e/utils/http/test_client.py new file mode 100644 index 0000000..8eed589 --- /dev/null +++ b/tests/unit/byteb4rb1e/utils/http/test_client.py @@ -0,0 +1,217 @@ +"""Tests for the generic HTTP client.""" + +import email.message +import io +import urllib.error +import urllib.parse +import urllib.request +from types import TracebackType +from typing import Dict, List, Optional, Tuple, Type, Union + +import pytest + +from byteb4rb1e.utils.http.client import HttpResponse, HttpSession + + +class _FakeRawResponse: + """Stands in for the object returned by OpenerDirector.open().""" + + def __init__( + self, + status: int = 200, + headers: Optional[Dict[str, str]] = None, + data: bytes = b"", + ) -> None: + self._status = status + self._headers = headers or {} + self._data = data + + def getcode(self) -> int: + return self._status + + def getheaders(self) -> List[Tuple[str, str]]: + return list(self._headers.items()) + + def read(self) -> bytes: + return self._data + + def __enter__(self) -> "_FakeRawResponse": + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + tb: Optional[TracebackType], + ) -> None: + return None + + +class _FakeOpener: + """Records requests and replays canned responses.""" + + def __init__( + self, + responses: Optional[ + List[Union[_FakeRawResponse, Exception]] + ] = None, + ) -> None: + self.requests: List[urllib.request.Request] = [] + self._responses = list(responses or [_FakeRawResponse()]) + + def open( + self, + req: urllib.request.Request, + timeout: Optional[int] = None, + ) -> _FakeRawResponse: + self.requests.append(req) + response = self._responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +def _http_error( + code: int = 404, + data: bytes = b"", + headers: Optional[Dict[str, str]] = None, +) -> urllib.error.HTTPError: + hdrs = email.message.Message() + for key, value in (headers or {}).items(): + hdrs[key] = value + return urllib.error.HTTPError( + "http://testserver/", code, "error", hdrs, io.BytesIO(data), + ) + + +class TestHttpResponse: + + def test_json(self) -> None: + resp = HttpResponse(200, {}, b'{"a": 1}') + assert resp.json() == {"a": 1} + + def test_text(self) -> None: + resp = HttpResponse(200, {}, b"hello") + assert resp.text == "hello" + + def test_text_replaces_invalid_utf8(self) -> None: + resp = HttpResponse(200, {}, b"\xff\xfe") + assert "�" in resp.text + + def test_reason_defaults_to_none(self) -> None: + resp = HttpResponse(200, {}, b"") + assert resp.reason is None + + def test_frozen(self) -> None: + resp = HttpResponse(200, {}, b"") + with pytest.raises(Exception): + resp.status_code = 500 + + +class TestHttpSession: + + def test_opener_has_cookie_processor(self) -> None: + session = HttpSession() + processors = [ + h for h in session._opener.handlers + if isinstance(h, urllib.request.HTTPCookieProcessor) + ] + assert len(processors) == 1 + assert processors[0].cookiejar is session._jar + + def test_get(self) -> None: + opener = _FakeOpener([ + _FakeRawResponse(200, {"X-Foo": "bar"}, b"body"), + ]) + session = HttpSession() + session._opener = opener + + resp = session.get("http://testserver/page") + + assert resp.status_code == 200 + assert resp.data == b"body" + assert resp.headers == {"X-Foo": "bar"} + assert opener.requests[0].get_method() == "GET" + assert opener.requests[0].full_url == "http://testserver/page" + + def test_get_with_params(self) -> None: + opener = _FakeOpener() + session = HttpSession() + session._opener = opener + + session.get("http://testserver/page", params={"a": "1", "b": "x y"}) + + assert opener.requests[0].full_url == ( + "http://testserver/page?a=1&b=x+y" + ) + + def test_default_headers_sent(self) -> None: + opener = _FakeOpener() + session = HttpSession(default_headers={"User-Agent": "test"}) + session._opener = opener + + session.get("http://testserver/") + + assert opener.requests[0].get_header("User-agent") == "test" + + def test_request_headers_override_defaults(self) -> None: + opener = _FakeOpener() + session = HttpSession(default_headers={"X-Token": "default"}) + session._opener = opener + + session.get("http://testserver/", headers={"X-Token": "override"}) + + assert opener.requests[0].get_header("X-token") == "override" + + def test_post_form_encodes_data(self) -> None: + opener = _FakeOpener() + session = HttpSession() + session._opener = opener + + session.post("http://testserver/login", data={"user": "u", "pass": "p"}) + + req = opener.requests[0] + assert req.get_method() == "POST" + assert isinstance(req.data, bytes) + assert dict(urllib.parse.parse_qsl(req.data.decode())) == { + "user": "u", + "pass": "p", + } + assert req.get_header("Content-type") == ( + "application/x-www-form-urlencoded" + ) + + def test_post_keeps_explicit_content_type(self) -> None: + opener = _FakeOpener() + session = HttpSession() + session._opener = opener + + session.post( + "http://testserver/", + data={"a": "1"}, + headers={"Content-Type": "text/plain"}, + ) + + assert opener.requests[0].get_header("Content-type") == "text/plain" + + def test_post_without_data(self) -> None: + opener = _FakeOpener() + session = HttpSession() + session._opener = opener + + session.post("http://testserver/") + + assert opener.requests[0].data is None + + def test_http_error_returned_as_response(self) -> None: + opener = _FakeOpener([ + _http_error(404, b"missing", {"X-Err": "yes"}), + ]) + session = HttpSession() + session._opener = opener + + resp = session.get("http://testserver/nope") + + assert resp.status_code == 404 + assert resp.data == b"missing" + assert resp.headers["X-Err"] == "yes"