import json import re from typing import Optional, Union, Any from re import Pattern import httpx from pytest_httpx._httpx_internals import _proxy_url from pytest_httpx._options import _HTTPXMockOptions def _url_match( url_to_match: Union[Pattern[str], httpx.URL], received: httpx.URL ) -> bool: if isinstance(url_to_match, re.Pattern): return url_to_match.match(str(received)) is not None # Compare query parameters apart as order of parameters should not matter received_params = dict(received.params) params = dict(url_to_match.params) # Remove the query parameters from the original URL to compare everything besides query parameters received_url = received.copy_with(query=None) url = url_to_match.copy_with(query=None) return (received_params == params) and (url == received_url) class _RequestMatcher: def __init__( self, options: _HTTPXMockOptions, url: Optional[Union[str, Pattern[str], httpx.URL]] = None, method: Optional[str] = None, proxy_url: Optional[Union[str, Pattern[str], httpx.URL]] = None, match_headers: Optional[dict[str, Any]] = None, match_content: Optional[bytes] = None, match_json: Optional[Any] = None, match_data: Optional[dict[str, Any]] = None, match_files: Optional[Any] = None, match_extensions: Optional[dict[str, Any]] = None, is_optional: Optional[bool] = None, is_reusable: Optional[bool] = None, ): self._options = options self.nb_calls = 0 self.url = httpx.URL(url) if url and isinstance(url, str) else url self.method = method.upper() if method else method self.headers = match_headers self.content = match_content self.json = match_json self.data = match_data self.files = match_files self.proxy_url = ( httpx.URL(proxy_url) if proxy_url and isinstance(proxy_url, str) else proxy_url ) self.extensions = match_extensions self.is_optional = not options.assert_all_responses_were_requested if is_optional is None else is_optional self.is_reusable = options.can_send_already_matched_responses if is_reusable is None else is_reusable if self._is_matching_body_more_than_one_way(): raise ValueError( "Only one way of matching against the body can be provided. " "If you want to match against the JSON decoded representation, use match_json. " "If you want to match against the multipart representation, use match_files (and match_data). " "Otherwise, use match_content." ) if self.data and not self.files: raise ValueError( "match_data is meant to be used for multipart matching (in conjunction with match_files)." "Use match_content to match url encoded data." ) def expect_body(self) -> bool: matching_ways = [ self.content is not None, self.json is not None, self.files is not None, ] return sum(matching_ways) == 1 def _is_matching_body_more_than_one_way(self) -> bool: matching_ways = [ self.content is not None, self.json is not None, self.files is not None, ] return sum(matching_ways) > 1 def match( self, real_transport: Union[httpx.HTTPTransport, httpx.AsyncHTTPTransport], request: httpx.Request, ) -> bool: return ( self._url_match(request) and self._method_match(request) and self._headers_match(request) and self._content_match(request) and self._proxy_match(real_transport) and self._extensions_match(request) ) def _url_match(self, request: httpx.Request) -> bool: if not self.url: return True return _url_match(self.url, request.url) def _method_match(self, request: httpx.Request) -> bool: if not self.method: return True return request.method == self.method def _headers_match(self, request: httpx.Request) -> bool: if not self.headers: return True encoding = request.headers.encoding request_headers = {} # Can be cleaned based on the outcome of https://github.com/encode/httpx/discussions/2841 for raw_name, raw_value in request.headers.raw: if raw_name in request_headers: request_headers[raw_name] += b", " + raw_value else: request_headers[raw_name] = raw_value return all( request_headers.get(header_name.encode(encoding)) == header_value.encode(encoding) for header_name, header_value in self.headers.items() ) def _content_match(self, request: httpx.Request) -> bool: if self.content is not None: return request.content == self.content if self.json is not None: try: # httpx._content.encode_json hard codes utf-8 encoding. return json.loads(request.content.decode("utf-8")) == self.json except json.decoder.JSONDecodeError: return False if self.files: if not ( boundary_matched := re.match(b"^--([0-9a-f]*)\r\n", request.content) ): return False # Ensure we re-use the same boundary for comparison boundary = boundary_matched.group(1) # Prevent internal httpx changes from impacting users not matching on files from httpx._multipart import MultipartStream multipart_content = b"".join( MultipartStream(self.data or {}, self.files, boundary) ) return request.content == multipart_content return True def _proxy_match( self, real_transport: Union[httpx.HTTPTransport, httpx.AsyncHTTPTransport] ) -> bool: if not self.proxy_url: return True if real_proxy_url := _proxy_url(real_transport): return _url_match(self.proxy_url, real_proxy_url) return False def _extensions_match(self, request: httpx.Request) -> bool: if not self.extensions: return True return all( request.extensions.get(extension_name) == extension_value for extension_name, extension_value in self.extensions.items() ) def should_have_matched(self) -> bool: """Return True if the matcher did not serve its purpose.""" return not self.is_optional and not self.nb_calls def __str__(self) -> str: if self.is_reusable: matcher_description = f"Match {self.method or 'every'} request" else: matcher_description = "Already matched" if self.nb_calls else "Match" matcher_description += f" {self.method or 'any'} request" if self.url: matcher_description += f" on {self.url}" if extra_description := self._extra_description(): matcher_description += f" with {extra_description}" return matcher_description def _extra_description(self) -> str: extra_description = [] if self.headers: extra_description.append(f"{self.headers} headers") if self.content is not None: extra_description.append(f"{self.content} body") if self.json is not None: extra_description.append(f"{self.json} json body") if self.data is not None: extra_description.append(f"{self.data} multipart data") if self.files is not None: extra_description.append(f"{self.files} files") if self.proxy_url: extra_description.append(f"{self.proxy_url} proxy URL") if self.extensions: extra_description.append(f"{self.extensions} extensions") return " and ".join(extra_description)