from collections.abc import Mapping, Sequence
from datetime import timedelta
from typing import Optional, Union
import httpx
import requests
from deprecated import deprecated
from furl import furl
from hamcrest import anything, described_as, has_entry
from hamcrest.core.base_matcher import BaseMatcher
from hamcrest.core.description import Description
from hamcrest.core.helpers.wrap_matcher import wrap_matcher
from hamcrest.core.matcher import Matcher
from hamcrest.core.string_description import StringDescription
from yarl import URL
from brunns.matchers.data import JsonStructure
from brunns.matchers.object import between
from brunns.matchers.utils import (
append_matcher_description,
describe_field_match,
describe_field_mismatch,
)
ANYTHING = anything()
[docs]
def is_response() -> "ResponseMatcher":
"""Matches a ``requests.Response`` or ``httpx.Response`` object.
This function returns a :class:`ResponseMatcher` which can be refined using builder methods
(e.g. ``.with_status_code(200)``).
:return: A matcher for HTTP responses.
"""
return ResponseMatcher()
ResponseType = Union[requests.Response, httpx.Response]
[docs]
class ResponseMatcher(BaseMatcher[ResponseType]):
"""Matches :class:`requests.Response` or :class:`httpx.Response`.
:param status_code: Expected status code.
:param body: Expected body text.
:param content: Expected raw binary content.
:param json: Expected JSON body (parsed).
:param headers: Expected headers dictionary.
:param cookies: Expected cookies dictionary.
:param elapsed: Expected elapsed time (timedelta).
:param history: Expected history sequence.
:param url: Expected URL.
:param encoding: Expected encoding string.
"""
def __init__(
self,
status_code: Union[int, Matcher[int]] = ANYTHING,
body: Union[str, Matcher[str]] = ANYTHING,
content: Union[bytes, Matcher[bytes]] = ANYTHING,
json: Union[JsonStructure, Matcher[JsonStructure]] = ANYTHING,
headers: Union[
Mapping[str, Union[str, Matcher[str]]],
Matcher[Mapping[str, Union[str, Matcher[str]]]],
] = ANYTHING,
cookies: Union[
Mapping[str, Union[str, Matcher[str]]],
Matcher[Mapping[str, Union[str, Matcher[str]]]],
] = ANYTHING,
elapsed: Union[timedelta, Matcher[timedelta]] = ANYTHING,
history: Union[
Sequence[
Union[
ResponseType,
Matcher[ResponseType],
]
],
Matcher[
Sequence[
Union[
ResponseType,
Matcher[ResponseType],
]
]
],
] = ANYTHING,
url: Union[furl, str, Matcher[Union[furl, str]]] = ANYTHING,
encoding: Union[Optional[str], Matcher[Optional[str]]] = ANYTHING,
) -> None:
super().__init__()
self.status_code: Matcher[int] = wrap_matcher(status_code)
self.body: Matcher[str] = wrap_matcher(body)
self.content: Matcher[bytes] = wrap_matcher(content)
self.json: Matcher[JsonStructure] = wrap_matcher(json)
self.headers: Matcher[Mapping[str, Union[str, Matcher[str]]]] = wrap_matcher(headers)
self.cookies: Matcher[Mapping[str, Union[str, Matcher[str]]]] = wrap_matcher(cookies)
self.elapsed = wrap_matcher(elapsed)
self.history = wrap_matcher(history)
self.url = wrap_matcher(url)
self.encoding = wrap_matcher(encoding)
def _matches(self, response: ResponseType) -> bool:
response_json = self._get_response_json(response)
return (
self.status_code.matches(response.status_code)
and self.body.matches(response.text)
and self.content.matches(response.content)
and self.json.matches(response_json)
and self.headers.matches(response.headers)
and self.cookies.matches(response.cookies)
and self.elapsed.matches(response.elapsed)
and self.history.matches(response.history)
and self.url.matches(response.url)
and self.encoding.matches(response.encoding)
)
@staticmethod
def _get_response_json(response: ResponseType) -> Optional[str]:
try:
return response.json()
except (ValueError, AttributeError):
return None
[docs]
def describe_to(self, description: Description) -> None:
description.append_text("response with")
append_matcher_description(self.status_code, "status code", description)
append_matcher_description(self.body, "body", description)
append_matcher_description(self.content, "content", description)
append_matcher_description(self.json, "json", description)
append_matcher_description(self.headers, "headers", description)
append_matcher_description(self.cookies, "cookies", description)
append_matcher_description(self.elapsed, "elapsed", description)
append_matcher_description(self.history, "history", description)
append_matcher_description(self.url, "url", description)
append_matcher_description(self.encoding, "encoding", description)
[docs]
def describe_mismatch(self, response: ResponseType, mismatch_description: Description) -> None:
mismatch_description.append_text("was response with")
describe_field_mismatch(self.status_code, "status code", response.status_code, mismatch_description)
describe_field_mismatch(self.body, "body", response.text, mismatch_description)
describe_field_mismatch(self.content, "content", response.content, mismatch_description)
describe_field_mismatch(self.json, "json", self._get_response_json(response), mismatch_description)
describe_field_mismatch(self.headers, "headers", response.headers, mismatch_description)
describe_field_mismatch(self.cookies, "cookies", response.cookies, mismatch_description)
describe_field_mismatch(self.elapsed, "elapsed", response.elapsed, mismatch_description)
describe_field_mismatch(self.history, "history", response.history, mismatch_description)
describe_field_mismatch(self.url, "url", response.url, mismatch_description)
describe_field_mismatch(self.encoding, "encoding", response.encoding, mismatch_description)
[docs]
def describe_match(self, response: ResponseType, match_description: Description) -> None:
match_description.append_text("was response with")
describe_field_match(self.status_code, "status code", response.status_code, match_description)
describe_field_match(self.body, "body", response.text, match_description)
describe_field_match(self.content, "content", response.content, match_description)
describe_field_match(self.json, "json", self._get_response_json(response), match_description)
describe_field_match(self.headers, "headers", response.headers, match_description)
describe_field_match(self.cookies, "cookies", response.cookies, match_description)
describe_field_match(self.elapsed, "elapsed", response.elapsed, match_description)
describe_field_match(self.history, "history", response.history, match_description)
describe_field_match(self.url, "url", response.url, match_description)
describe_field_match(self.encoding, "encoding", response.encoding, match_description)
[docs]
def with_status_code(self, status_code: Union[int, Matcher[int]]):
"""Matches if the response status code matches the given value or matcher.
:param status_code: The expected status code (e.g. 200) or a matcher (e.g. ``between(200, 299)``).
:return: Self, for chaining.
"""
self.status_code = wrap_matcher(status_code)
return self
[docs]
def and_status_code(self, status_code: Union[int, Matcher[int]]):
"""Matches if the response status code matches the given value or matcher.
A synonym for :meth:`with_status_code`.
:param status_code: The expected status code.
:return: Self, for chaining.
"""
return self.with_status_code(status_code)
[docs]
def with_body(self, body: Union[str, Matcher[str]]):
"""Matches if the response body text matches the given value or matcher.
:param body: The expected body string or matcher.
:return: Self, for chaining.
"""
self.body = wrap_matcher(body)
return self
[docs]
def and_body(self, body: Union[str, Matcher[str]]):
"""Matches if the response body text matches the given value or matcher.
A synonym for :meth:`with_body`.
:param body: The expected body string or matcher.
:return: Self, for chaining.
"""
return self.with_body(body)
[docs]
def with_content(self, content: Union[bytes, Matcher[bytes]]):
"""Matches if the response binary content matches the given value or matcher.
:param content: The expected bytes or matcher.
:return: Self, for chaining.
"""
self.content = wrap_matcher(content)
return self
[docs]
def and_content(self, content: Union[bytes, Matcher[bytes]]):
"""Matches if the response binary content matches the given value or matcher.
A synonym for :meth:`with_content`.
:param content: The expected bytes or matcher.
:return: Self, for chaining.
"""
return self.with_content(content)
[docs]
def with_json(self, json: Union[JsonStructure, Matcher[JsonStructure]]):
"""Matches if the response JSON body matches the given value or matcher.
The response body is parsed as JSON before matching.
:param json: The expected JSON structure (dict, list, etc.) or matcher.
:return: Self, for chaining.
"""
self.json = wrap_matcher(json)
return self
[docs]
def and_json(self, json: Union[JsonStructure, Matcher[JsonStructure]]):
"""Matches if the response JSON body matches the given value or matcher.
A synonym for :meth:`with_json`.
:param json: The expected JSON structure or matcher.
:return: Self, for chaining.
"""
return self.with_json(json)
[docs]
def and_headers(
self,
headers: Union[Mapping[str, Union[str, Matcher[str]]], Matcher[Mapping[str, Union[str, Matcher[str]]]]],
):
"""Matches if the response headers match the given value or matcher.
A synonym for :meth:`with_headers`.
:param headers: The expected headers dictionary or matcher.
:return: Self, for chaining.
"""
return self.with_headers(headers)
[docs]
def with_cookies(
self,
cookies: Union[Mapping[str, Union[str, Matcher[str]]], Matcher[Mapping[str, Union[str, Matcher[str]]]]],
):
"""Matches if the response cookies match the given value or matcher.
:param cookies: The expected cookies dictionary or matcher.
:return: Self, for chaining.
"""
self.cookies = wrap_matcher(cookies)
return self
[docs]
def and_cookies(
self,
cookies: Union[Mapping[str, Union[str, Matcher[str]]], Matcher[Mapping[str, Union[str, Matcher[str]]]]],
):
"""Matches if the response cookies match the given value or matcher.
A synonym for :meth:`with_cookies`.
:param cookies: The expected cookies dictionary or matcher.
:return: Self, for chaining.
"""
return self.with_cookies(cookies)
[docs]
def with_elapsed(self, elapsed: Union[timedelta, Matcher[timedelta]]):
"""Matches if the response elapsed time matches the given value or matcher.
:param elapsed: The expected timedelta or matcher.
:return: Self, for chaining.
"""
self.elapsed = wrap_matcher(elapsed)
return self
[docs]
def and_elapsed(self, elapsed: Union[timedelta, Matcher[timedelta]]):
"""Matches if the response elapsed time matches the given value or matcher.
A synonym for :meth:`with_elapsed`.
:param elapsed: The expected timedelta or matcher.
:return: Self, for chaining.
"""
return self.with_elapsed(elapsed)
[docs]
def with_history(
self,
history: Union[
Sequence[
Union[
ResponseType,
Matcher[ResponseType],
]
],
Matcher[
Sequence[
Union[
ResponseType,
Matcher[ResponseType],
]
]
],
],
):
"""Matches if the response history (redirects) matches the given sequence or matcher.
:param history: The expected sequence of responses/matchers or a sequence matcher.
:return: Self, for chaining.
"""
self.history = wrap_matcher(history)
return self
[docs]
def and_history(
self,
history: Union[
Sequence[Union[ResponseType, Matcher[ResponseType]]],
Matcher[Sequence[Union[ResponseType, Matcher[ResponseType]]]],
],
):
"""Matches if the response history (redirects) matches the given sequence or matcher.
A synonym for :meth:`with_history`.
:param history: The expected sequence or matcher.
:return: Self, for chaining.
"""
return self.with_history(history)
[docs]
def with_url(self, url: Union[furl, str, Matcher[Union[furl, str]]]):
"""Matches if the response URL matches the given value or matcher.
:param url: The expected URL string, object, or matcher.
:return: Self, for chaining.
"""
self.url = wrap_matcher(url)
return self
[docs]
def and_url(self, url: Union[furl, str, Matcher[Union[furl, str]]]):
"""Matches if the response URL matches the given value or matcher.
A synonym for :meth:`with_url`.
:param url: The expected URL string, object, or matcher.
:return: Self, for chaining.
"""
return self.with_url(url)
[docs]
def with_encoding(self, encoding: Union[Optional[str], Matcher[Optional[str]]]):
"""Matches if the response encoding matches the given value or matcher.
:param encoding: The expected encoding string or matcher.
:return: Self, for chaining.
"""
self.encoding = wrap_matcher(encoding)
return self
[docs]
def and_encoding(self, encoding: Union[Optional[str], Matcher[Optional[str]]]):
"""Matches if the response encoding matches the given value or matcher.
A synonym for :meth:`with_encoding`.
:param encoding: The expected encoding string or matcher.
:return: Self, for chaining.
"""
return self.with_encoding(encoding)
[docs]
def redirects_to(url_matcher: Union[str, Matcher[str], URL, Matcher[URL]]) -> Matcher[ResponseType]:
"""Is a response a redirect to a URL matching the supplied matcher?
Matches if the status code is between 300 and 399 and the ``Location`` header matches
the provided URL matcher.
:param url_matcher: The expected URL (string or matcher) found in the Location header.
:return: A matcher for redirect responses.
"""
return described_as(
str(StringDescription().append_text("redirects to ").append_description_of(url_matcher)),
is_response().with_status_code(between(300, 399)).and_headers(has_entry("Location", url_matcher)),
)
[docs]
@deprecated(version="2.3.0", reason="Use builder style is_response()")
def response_with(
status_code: Union[int, Matcher[int]] = ANYTHING,
body: Union[str, Matcher[str]] = ANYTHING,
content: Union[bytes, Matcher[bytes]] = ANYTHING,
json: Union[JsonStructure, Matcher[JsonStructure]] = ANYTHING,
headers: Union[Mapping[str, Union[str, Matcher[str]]], Matcher[Mapping[str, Union[str, Matcher[str]]]]] = ANYTHING,
) -> ResponseMatcher: # pragma: no cover
"""Matches a response with specific attributes.
.. deprecated:: 2.3.0
Use :func:`is_response` and its builder methods instead.
"""
return ResponseMatcher(status_code=status_code, body=body, content=content, json=json, headers=headers)