diff options
Diffstat (limited to 'tests/unit/helpers.py')
-rw-r--r-- | tests/unit/helpers.py | 66 |
1 files changed, 66 insertions, 0 deletions
diff --git a/tests/unit/helpers.py b/tests/unit/helpers.py new file mode 100644 index 0000000..33a7c78 --- /dev/null +++ b/tests/unit/helpers.py @@ -0,0 +1,66 @@ +import datetime +import io +import json +from typing import Optional + +import requests + + +# NOTE: The function `httmock_response` and the class `Headers` is taken from +# https://github.com/patrys/httmock/ which is licensed under the Apache License, Version +# 2.0. Thus it is allowed to be used in this project. +# https://www.apache.org/licenses/GPL-compatibility.html +class Headers(object): + def __init__(self, res): + self.headers = res.headers + + def get_all(self, name, failobj=None): + return self.getheaders(name) + + def getheaders(self, name): + return [self.headers.get(name)] + + +def httmock_response( + status_code: int = 200, + content: str = "", + headers=None, + reason=None, + elapsed=0, + request: Optional[requests.models.PreparedRequest] = None, + stream: bool = False, + http_vsn=11, +) -> requests.models.Response: + res = requests.Response() + res.status_code = status_code + if isinstance(content, (dict, list)): + content = json.dumps(content).encode("utf-8") + if isinstance(content, str): + content = content.encode("utf-8") + res._content = content + res._content_consumed = content + res.headers = requests.structures.CaseInsensitiveDict(headers or {}) + res.encoding = requests.utils.get_encoding_from_headers(res.headers) + res.reason = reason + res.elapsed = datetime.timedelta(elapsed) + res.request = request + if hasattr(request, "url"): + res.url = request.url + if isinstance(request.url, bytes): + res.url = request.url.decode("utf-8") + if "set-cookie" in res.headers: + res.cookies.extract_cookies( + requests.cookies.MockResponse(Headers(res)), + requests.cookies.MockRequest(request), + ) + if stream: + res.raw = io.BytesIO(content) + else: + res.raw = io.BytesIO(b"") + res.raw.version = http_vsn + + # normally this closes the underlying connection, + # but we have nothing to free. + res.close = lambda *args, **kwargs: None + + return res |