diff --git a/CHANGELOG.md b/CHANGELOG.md index f279eee..cf57b10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.26.0] - 2023-09-18 +### Added +- Added `proxy_url` parameter which allows matching on proxy URL. + ## [0.25.0] - 2023-09-11 ### Changed - Requires [`httpx`](https://www.python-httpx.org)==0.25.\* @@ -283,7 +287,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - First release, should be considered as unstable for now as design might change. -[Unreleased]: https://github.com/Colin-b/pytest_httpx/compare/v0.25.0...HEAD +[Unreleased]: https://github.com/Colin-b/pytest_httpx/compare/v0.26.0...HEAD +[0.26.0]: https://github.com/Colin-b/pytest_httpx/compare/v0.25.0...v0.26.0 [0.25.0]: https://github.com/Colin-b/pytest_httpx/compare/v0.24.0...v0.25.0 [0.24.0]: https://github.com/Colin-b/pytest_httpx/compare/v0.23.1...v0.24.0 [0.23.1]: https://github.com/Colin-b/pytest_httpx/compare/v0.23.0...v0.23.1 diff --git a/README.md b/README.md index ad895de..006123d 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Build status Coverage Code style: black -Number of tests +Number of tests Number of downloads

@@ -146,6 +146,26 @@ def test_head(httpx_mock: HTTPXMock): ``` +#### Matching on proxy URL + +`proxy_url` parameter can either be a string, a python [re.Pattern](https://docs.python.org/3/library/re.html) instance or a [httpx.URL](https://www.python-httpx.org/api/#url) instance. + +Matching is performed on the full proxy URL, query parameters included. + +Order of parameters in the query string does not matter, however order of values do matter if the same parameter is provided more than once. + +```python +import httpx +from pytest_httpx import HTTPXMock + + +def test_proxy_url(httpx_mock: HTTPXMock): + httpx_mock.add_response(proxy_url="http://test_proxy_url?a=1&b=2") + + with httpx.Client(proxies={"https://": "http://test_proxy_url?a=2&b=1"}) as client: + response = client.get("https://test_url") +``` + #### Matching on HTTP headers Use `match_headers` parameter to specify the HTTP headers to reply to. @@ -586,6 +606,8 @@ You can add criteria so that requests will be returned only in case of a more sp Matching is performed on the full URL, query parameters included. +Order of parameters in the query string does not matter, however order of values do matter if the same parameter is provided more than once. + #### Matching on HTTP method Use `method` parameter to specify the HTTP method (POST, PUT, DELETE, PATCH, HEAD) of the requests to retrieve. @@ -594,6 +616,14 @@ Use `method` parameter to specify the HTTP method (POST, PUT, DELETE, PATCH, HEA Matching is performed on equality. +#### Matching on proxy URL + +`proxy_url` parameter can either be a string, a python [re.Pattern](https://docs.python.org/3/library/re.html) instance or a [httpx.URL](https://www.python-httpx.org/api/#url) instance. + +Matching is performed on the full proxy URL, query parameters included. + +Order of parameters in the query string does not matter, however order of values do matter if the same parameter is provided more than once. + #### Matching on HTTP headers Use `match_headers` parameter to specify the HTTP headers executing the callback. @@ -606,6 +636,14 @@ Use `match_content` parameter to specify the full HTTP body executing the callba Matching is performed on equality. +##### Matching on HTTP JSON body + +Use `match_json` parameter to specify the JSON decoded HTTP body executing the callback. + +Matching is performed on equality. You can however use `unittest.mock.ANY` to do partial matching. + +Note that `match_content` cannot be provided if `match_json` is also provided. + ## Do not mock some requests By default, `pytest-httpx` will mock every request. diff --git a/pytest_httpx/__init__.py b/pytest_httpx/__init__.py index 540d631..b579fa9 100644 --- a/pytest_httpx/__init__.py +++ b/pytest_httpx/__init__.py @@ -51,7 +51,7 @@ def httpx_mock( "_transport_for_url", lambda self, url: real_sync_transport(self, url) if url.host in non_mocked_hosts - else _PytestSyncTransport(mock), + else _PytestSyncTransport(real_sync_transport(self, url), mock), ) # Mock asynchronous requests real_async_transport = httpx.AsyncClient._transport_for_url @@ -60,7 +60,7 @@ def httpx_mock( "_transport_for_url", lambda self, url: real_async_transport(self, url) if url.host in non_mocked_hosts - else _PytestAsyncTransport(mock), + else _PytestAsyncTransport(real_async_transport(self, url), mock), ) yield mock mock.reset(assert_all_responses_were_requested) diff --git a/pytest_httpx/_httpx_internals.py b/pytest_httpx/_httpx_internals.py index a953bea..e12b933 100644 --- a/pytest_httpx/_httpx_internals.py +++ b/pytest_httpx/_httpx_internals.py @@ -1,3 +1,4 @@ +import base64 from typing import ( Union, Dict, @@ -6,8 +7,10 @@ Iterable, AsyncIterator, Iterator, + Optional, ) +import httpcore import httpx # TODO Get rid of this internal import @@ -36,3 +39,34 @@ async def __aiter__(self) -> AsyncIterator[bytes]: AsyncIteratorByteStream.__init__(self, stream=Stream()) IteratorByteStream.__init__(self, stream=Stream()) + + +def _to_httpx_url(url: httpcore.URL, headers: list[tuple[bytes, bytes]]) -> httpx.URL: + for name, value in headers: + if b"Proxy-Authorization" == name: + return httpx.URL( + scheme=url.scheme.decode(), + host=url.host.decode(), + port=url.port, + raw_path=url.target, + userinfo=base64.b64decode(value[6:]), + ) + + return httpx.URL( + scheme=url.scheme.decode(), + host=url.host.decode(), + port=url.port, + raw_path=url.target, + ) + + +def _proxy_url( + real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport] +) -> Optional[httpx.URL]: + if isinstance(real_transport, httpx.HTTPTransport): + if isinstance(real_pool := real_transport._pool, httpcore.HTTPProxy): + return _to_httpx_url(real_pool._proxy_url, real_pool._proxy_headers) + + if isinstance(real_transport, httpx.AsyncHTTPTransport): + if isinstance(real_pool := real_transport._pool, httpcore.AsyncHTTPProxy): + return _to_httpx_url(real_pool._proxy_url, real_pool._proxy_headers) diff --git a/pytest_httpx/_httpx_mock.py b/pytest_httpx/_httpx_mock.py index 975c531..425f420 100644 --- a/pytest_httpx/_httpx_mock.py +++ b/pytest_httpx/_httpx_mock.py @@ -1,117 +1,21 @@ import copy import inspect -import json -import re -from typing import List, Union, Optional, Callable, Tuple, Pattern, Any, Dict, Awaitable +from typing import Union, Optional, Callable, Any, Awaitable import httpx from pytest_httpx import _httpx_internals - - -class _RequestMatcher: - def __init__( - self, - url: Optional[Union[str, Pattern[str], httpx.URL]] = None, - method: Optional[str] = None, - match_headers: Optional[Dict[str, Any]] = None, - match_content: Optional[bytes] = None, - match_json: Optional[Any] = None, - ): - 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 - if match_content is not None and match_json is not None: - 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. Otherwise, use match_content." - ) - self.content = match_content - self.json = match_json - - def match(self, request: httpx.Request) -> bool: - return ( - self._url_match(request) - and self._method_match(request) - and self._headers_match(request) - and self._content_match(request) - ) - - def _url_match(self, request: httpx.Request) -> bool: - if not self.url: - return True - - if isinstance(self.url, re.Pattern): - return self.url.match(str(request.url)) is not None - - # Compare query parameters apart as order of parameters should not matter - request_params = dict(request.url.params) - params = dict(self.url.params) - - # Remove the query parameters from the original URL to compare everything besides query parameters - request_url = request.url.copy_with(query=None) - url = self.url.copy_with(query=None) - - return (request_params == params) and (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 None and self.json is None: - return True - if self.content is not None: - return request.read() == self.content - try: - # httpx._content.encode_json hard codes utf-8 encoding. - return json.loads(request.read().decode("utf-8")) == self.json - except json.decoder.JSONDecodeError: - return False - - def __str__(self) -> str: - matcher_description = f"Match {self.method or 'all'} requests" - if self.url: - matcher_description += f" on {self.url}" - if self.headers: - matcher_description += f" with {self.headers} headers" - if self.content is not None: - matcher_description += f" and {self.content} body" - elif self.json is not None: - matcher_description += f" and {self.json} json body" - elif self.content is not None: - matcher_description += f" with {self.content} body" - elif self.json is not None: - matcher_description += f" with {self.json} json body" - return matcher_description +from pytest_httpx._pretty_print import RequestDescription +from pytest_httpx._request_matcher import _RequestMatcher class HTTPXMock: def __init__(self) -> None: - self._requests: List[httpx.Request] = [] - self._callbacks: List[ - Tuple[ + self._requests: list[ + tuple[Union[httpx.BaseTransport, httpx.AsyncBaseTransport], httpx.Request] + ] = [] + self._callbacks: list[ + tuple[ _RequestMatcher, Callable[ [httpx.Request], @@ -148,6 +52,8 @@ def add_response( :param url: Full URL identifying the request(s) to match. Can be a str, a re.Pattern instance or a httpx.URL instance. :param method: HTTP method identifying the request(s) to match. + :param proxy_url: Full proxy URL identifying the request(s) to match. + Can be a str, a re.Pattern instance or a httpx.URL instance. :param match_headers: HTTP headers identifying the request(s) to match. Must be a dictionary. :param match_content: Full HTTP body identifying the request(s) to match. Must be bytes. :param match_json: JSON decoded HTTP body identifying the request(s) to match. Must be JSON encodable. @@ -185,6 +91,8 @@ def add_callback( :param url: Full URL identifying the request(s) to match. Can be a str, a re.Pattern instance or a httpx.URL instance. :param method: HTTP method identifying the request(s) to match. + :param proxy_url: Full proxy URL identifying the request(s) to match. + Can be a str, a re.Pattern instance or a httpx.URL instance. :param match_headers: HTTP headers identifying the request(s) to match. Must be a dictionary. :param match_content: Full HTTP body identifying the request(s) to match. Must be bytes. :param match_json: JSON decoded HTTP body identifying the request(s) to match. Must be JSON encodable. @@ -199,6 +107,8 @@ def add_exception(self, exception: Exception, **matchers: Any) -> None: :param url: Full URL identifying the request(s) to match. Can be a str, a re.Pattern instance or a httpx.URL instance. :param method: HTTP method identifying the request(s) to match. + :param proxy_url: Full proxy URL identifying the request(s) to match. + Can be a str, a re.Pattern instance or a httpx.URL instance. :param match_headers: HTTP headers identifying the request(s) to match. Must be a dictionary. :param match_content: Full HTTP body identifying the request(s) to match. Must be bytes. :param match_json: JSON decoded HTTP body identifying the request(s) to match. Must be JSON encodable. @@ -213,11 +123,12 @@ def exception_callback(request: httpx.Request) -> None: def _handle_request( self, + real_transport: httpx.BaseTransport, request: httpx.Request, ) -> httpx.Response: - self._requests.append(request) + self._requests.append((real_transport, request)) - callback = self._get_callback(request) + callback = self._get_callback(real_transport, request) if callback: response = callback(request) @@ -225,16 +136,18 @@ def _handle_request( return _unread(response) raise httpx.TimeoutException( - self._explain_that_no_response_was_found(request), request=request + self._explain_that_no_response_was_found(real_transport, request), + request=request, ) async def _handle_async_request( self, + real_transport: httpx.AsyncBaseTransport, request: httpx.Request, ) -> httpx.Response: - self._requests.append(request) + self._requests.append((real_transport, request)) - callback = self._get_callback(request) + callback = self._get_callback(real_transport, request) if callback: response = callback(request) @@ -244,56 +157,29 @@ async def _handle_async_request( return _unread(response) raise httpx.TimeoutException( - self._explain_that_no_response_was_found(request), request=request + self._explain_that_no_response_was_found(real_transport, request), + request=request, ) - def _explain_that_no_response_was_found(self, request: httpx.Request) -> str: + def _explain_that_no_response_was_found( + self, + real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport], + request: httpx.Request, + ) -> str: matchers = [matcher for matcher, _ in self._callbacks] - headers_encoding = request.headers.encoding - expect_headers = set( - [ - # httpx uses lower cased header names as internal key - header.lower().encode(headers_encoding) - for matcher in matchers - if matcher.headers - for header in matcher.headers - ] - ) - expect_body = any( - [ - matcher.content is not None or matcher.json is not None - for matcher in matchers - ] - ) - request_description = f"{request.method} request on {request.url}" - if expect_headers: - present_headers = {} - # Can be cleaned based on the outcome of https://github.com/encode/httpx/discussions/2841 - for name, lower_name, value in request.headers._list: - if lower_name in expect_headers: - name = name.decode(headers_encoding) - if name in present_headers: - present_headers[name] += f", {value.decode(headers_encoding)}" - else: - present_headers[name] = value.decode(headers_encoding) - - request_description += f" with {present_headers} headers" - if expect_body: - request_description += f" and {request.read()} body" - elif expect_body: - request_description += f" with {request.read()} body" + message = f"No response can be found for {RequestDescription(real_transport, request, matchers)}" matchers_description = "\n".join([str(matcher) for matcher in matchers]) - - message = f"No response can be found for {request_description}" if matchers_description: message += f" amongst:\n{matchers_description}" return message def _get_callback( - self, request: httpx.Request + self, + real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport], + request: httpx.Request, ) -> Optional[ Callable[ [httpx.Request], @@ -303,7 +189,7 @@ def _get_callback( callbacks = [ (matcher, callback) for matcher, callback in self._callbacks - if matcher.match(request) + if matcher.match(real_transport, request) ] # No callback match this request @@ -321,19 +207,25 @@ def _get_callback( matcher.nb_calls += 1 return callback - def get_requests(self, **matchers: Any) -> List[httpx.Request]: + def get_requests(self, **matchers: Any) -> list[httpx.Request]: """ Return all requests sent that match (empty list if no requests were matched). :param url: Full URL identifying the requests to retrieve. Can be a str, a re.Pattern instance or a httpx.URL instance. :param method: HTTP method identifying the requests to retrieve. Must be an upper-cased string value. + :param proxy_url: Full proxy URL identifying the requests to retrieve. + Can be a str, a re.Pattern instance or a httpx.URL instance. :param match_headers: HTTP headers identifying the requests to retrieve. Must be a dictionary. :param match_content: Full HTTP body identifying the requests to retrieve. Must be bytes. :param match_json: JSON decoded HTTP body identifying the requests to retrieve. Must be JSON encodable. """ matcher = _RequestMatcher(**matchers) - return [request for request in self._requests if matcher.match(request)] + return [ + request + for real_transport, request in self._requests + if matcher.match(real_transport, request) + ] def get_request(self, **matchers: Any) -> Optional[httpx.Request]: """ @@ -342,6 +234,8 @@ def get_request(self, **matchers: Any) -> Optional[httpx.Request]: :param url: Full URL identifying the request to retrieve. Can be a str, a re.Pattern instance or a httpx.URL instance. :param method: HTTP method identifying the request to retrieve. Must be an upper-cased string value. + :param proxy_url: Full proxy URL identifying the request to retrieve. + Can be a str, a re.Pattern instance or a httpx.URL instance. :param match_headers: HTTP headers identifying the request to retrieve. Must be a dictionary. :param match_content: Full HTTP body identifying the request to retrieve. Must be bytes. :param match_json: JSON decoded HTTP body identifying the request to retrieve. Must be JSON encodable. @@ -364,7 +258,7 @@ def reset(self, assert_all_responses_were_requested: bool) -> None: not not_called ), f"The following responses are mocked but not requested:\n{matchers_description}" - def _reset_callbacks(self) -> List[_RequestMatcher]: + def _reset_callbacks(self) -> list[_RequestMatcher]: callbacks_not_executed = [ matcher for matcher, _ in self._callbacks if not matcher.nb_calls ] @@ -373,19 +267,21 @@ def _reset_callbacks(self) -> List[_RequestMatcher]: class _PytestSyncTransport(httpx.BaseTransport): - def __init__(self, mock: HTTPXMock): - self.mock = mock + def __init__(self, real_transport: httpx.BaseTransport, mock: HTTPXMock): + self._real_transport = real_transport + self._mock = mock - def handle_request(self, *args, **kwargs) -> httpx.Response: - return self.mock._handle_request(*args, **kwargs) + def handle_request(self, request: httpx.Request) -> httpx.Response: + return self._mock._handle_request(self._real_transport, request) class _PytestAsyncTransport(httpx.AsyncBaseTransport): - def __init__(self, mock: HTTPXMock): - self.mock = mock + def __init__(self, real_transport: httpx.AsyncBaseTransport, mock: HTTPXMock): + self._real_transport = real_transport + self._mock = mock - async def handle_async_request(self, *args, **kwargs) -> httpx.Response: - return await self.mock._handle_async_request(*args, **kwargs) + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + return await self._mock._handle_async_request(self._real_transport, request) def _unread(response: httpx.Response) -> httpx.Response: diff --git a/pytest_httpx/_pretty_print.py b/pytest_httpx/_pretty_print.py new file mode 100644 index 0000000..2d6f336 --- /dev/null +++ b/pytest_httpx/_pretty_print.py @@ -0,0 +1,67 @@ +from typing import Union + +import httpx + +from pytest_httpx._httpx_internals import _proxy_url +from pytest_httpx._request_matcher import _RequestMatcher + + +class RequestDescription: + def __init__( + self, + real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport], + request: httpx.Request, + matchers: list[_RequestMatcher], + ): + self.real_transport = real_transport + self.request = request + + headers_encoding = request.headers.encoding + self.expected_headers = set( + [ + # httpx uses lower cased header names as internal key + header.lower().encode(headers_encoding) + for matcher in matchers + if matcher.headers + for header in matcher.headers + ] + ) + self.expect_body = any( + [ + matcher.content is not None or matcher.json is not None + for matcher in matchers + ] + ) + self.expect_proxy = any([matcher.proxy_url is not None for matcher in matchers]) + + def __str__(self) -> str: + request_description = f"{self.request.method} request on {self.request.url}" + if extra_description := self.extra_request_description(): + request_description += f" with {extra_description}" + return request_description + + def extra_request_description(self) -> str: + extra_description = [] + + if self.expected_headers: + headers_encoding = self.request.headers.encoding + present_headers = {} + # Can be cleaned based on the outcome of https://github.com/encode/httpx/discussions/2841 + for name, lower_name, value in self.request.headers._list: + if lower_name in self.expected_headers: + name = name.decode(headers_encoding) + if name in present_headers: + present_headers[name] += f", {value.decode(headers_encoding)}" + else: + present_headers[name] = value.decode(headers_encoding) + + extra_description.append(f"{present_headers} headers") + + if self.expect_body: + extra_description.append(f"{self.request.read()} body") + + if self.expect_proxy: + proxy_url = _proxy_url(self.real_transport) + extra_description.append(f"{proxy_url if proxy_url else 'no'} proxy URL") + + return " and ".join(extra_description) diff --git a/pytest_httpx/_request_matcher.py b/pytest_httpx/_request_matcher.py new file mode 100644 index 0000000..e80df4b --- /dev/null +++ b/pytest_httpx/_request_matcher.py @@ -0,0 +1,139 @@ +import json +import re +from typing import Optional, Union, Pattern, Any + +import httpx + +from pytest_httpx._httpx_internals import _proxy_url + + +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, + 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, + ): + 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 + if match_content is not None and match_json is not None: + 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. Otherwise, use match_content." + ) + self.content = match_content + self.json = match_json + self.proxy_url = ( + httpx.URL(proxy_url) + if proxy_url and isinstance(proxy_url, str) + else proxy_url + ) + + def match( + self, + real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport], + 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) + ) + + 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 None and self.json is None: + return True + if self.content is not None: + return request.read() == self.content + try: + # httpx._content.encode_json hard codes utf-8 encoding. + return json.loads(request.read().decode("utf-8")) == self.json + except json.decoder.JSONDecodeError: + return False + + def _proxy_match( + self, real_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport] + ) -> 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 __str__(self) -> str: + matcher_description = f"Match {self.method or 'all'} requests" + 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.proxy_url: + extra_description.append(f"{self.proxy_url} proxy URL") + + return " and ".join(extra_description) diff --git a/pytest_httpx/version.py b/pytest_httpx/version.py index 3e581ce..2773c8d 100644 --- a/pytest_httpx/version.py +++ b/pytest_httpx/version.py @@ -3,4 +3,4 @@ # Major should be incremented in case there is a breaking change. (eg: 2.5.8 -> 3.0.0) # Minor should be incremented in case there is an enhancement. (eg: 2.5.8 -> 2.6.0) # Patch should be incremented in case there is a bug fix. (eg: 2.5.8 -> 2.5.9) -__version__ = "0.25.0" +__version__ = "0.26.0" diff --git a/tests/test_httpx_async.py b/tests/test_httpx_async.py index 1f1ab99..e9ad034 100644 --- a/tests/test_httpx_async.py +++ b/tests/test_httpx_async.py @@ -1229,6 +1229,119 @@ async def test_content_matching(httpx_mock: HTTPXMock) -> None: assert response.read() == b"" +@pytest.mark.asyncio +async def test_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://user:pwd@my_other_proxy/") + + async with httpx.AsyncClient( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + response = await client.get("https://test_url") + assert response.read() == b"" + + +@pytest.mark.asyncio +async def test_proxy_not_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://my_test_proxy") + + async with httpx.AsyncClient( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + with pytest.raises(httpx.TimeoutException) as exception_info: + await client.get("http://test_url") + assert ( + str(exception_info.value) + == """No response can be found for GET request on http://test_url with http://my_test_proxy/ proxy URL amongst: +Match all requests with http://my_test_proxy proxy URL""" + ) + + # Clean up responses to avoid assertion failure + httpx_mock.reset(assert_all_responses_were_requested=False) + + +@pytest.mark.asyncio +async def test_proxy_not_existing(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://my_test_proxy") + + async with httpx.AsyncClient() as client: + with pytest.raises(httpx.TimeoutException) as exception_info: + await client.get("http://test_url") + assert ( + str(exception_info.value) + == """No response can be found for GET request on http://test_url with no proxy URL amongst: +Match all requests with http://my_test_proxy proxy URL""" + ) + + # Clean up responses to avoid assertion failure + httpx_mock.reset(assert_all_responses_were_requested=False) + + +@pytest.mark.asyncio +async def test_requests_retrieval_content_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + async with httpx.AsyncClient() as client: + await client.post("https://test_url", content=b"This is the body") + await client.post("https://test_url2", content=b"This is the body") + await client.post("https://test_url2", content=b"This is the body2") + + assert len(httpx_mock.get_requests(match_content=b"This is the body")) == 2 + + +@pytest.mark.asyncio +async def test_requests_retrieval_json_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + async with httpx.AsyncClient() as client: + await client.post("https://test_url", json=["my_str"]) + await client.post("https://test_url2", json=["my_str"]) + await client.post("https://test_url2", json=["my_str2"]) + + assert len(httpx_mock.get_requests(match_json=["my_str"])) == 2 + + +@pytest.mark.asyncio +async def test_requests_retrieval_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + async with httpx.AsyncClient( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + await client.get("https://test_url") + await client.get("https://test_url2") + await client.get("http://test_url2") + + assert ( + len(httpx_mock.get_requests(proxy_url="http://user:pwd@my_other_proxy/")) == 2 + ) + + +@pytest.mark.asyncio +async def test_request_retrieval_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + async with httpx.AsyncClient( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + await client.get("https://test_url") + await client.get("https://test_url2") + await client.get("http://test_url2") + + assert httpx_mock.get_request(proxy_url="http://my_test_proxy/") + + @pytest.mark.asyncio async def test_content_not_matching(httpx_mock: HTTPXMock) -> None: httpx_mock.add_response(match_content=b"This is the body") diff --git a/tests/test_httpx_sync.py b/tests/test_httpx_sync.py index 5e63bbc..00431c5 100644 --- a/tests/test_httpx_sync.py +++ b/tests/test_httpx_sync.py @@ -977,6 +977,112 @@ def test_content_matching(httpx_mock: HTTPXMock) -> None: assert response.read() == b"" +def test_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://user:pwd@my_other_proxy/") + + with httpx.Client( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + response = client.get("https://test_url") + assert response.read() == b"" + + +def test_proxy_not_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://my_test_proxy") + + with httpx.Client( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + with pytest.raises(httpx.TimeoutException) as exception_info: + client.get("http://test_url") + assert ( + str(exception_info.value) + == """No response can be found for GET request on http://test_url with http://my_test_proxy/ proxy URL amongst: +Match all requests with http://my_test_proxy proxy URL""" + ) + + # Clean up responses to avoid assertion failure + httpx_mock.reset(assert_all_responses_were_requested=False) + + +def test_proxy_not_existing(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response(proxy_url="http://my_test_proxy") + + with httpx.Client() as client: + with pytest.raises(httpx.TimeoutException) as exception_info: + client.get("http://test_url") + assert ( + str(exception_info.value) + == """No response can be found for GET request on http://test_url with no proxy URL amongst: +Match all requests with http://my_test_proxy proxy URL""" + ) + + # Clean up responses to avoid assertion failure + httpx_mock.reset(assert_all_responses_were_requested=False) + + +def test_requests_retrieval_content_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + with httpx.Client() as client: + client.post("https://test_url", content=b"This is the body") + client.post("https://test_url2", content=b"This is the body") + client.post("https://test_url2", content=b"This is the body2") + + assert len(httpx_mock.get_requests(match_content=b"This is the body")) == 2 + + +def test_requests_retrieval_json_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + with httpx.Client() as client: + client.post("https://test_url", json=["my_str"]) + client.post("https://test_url2", json=["my_str"]) + client.post("https://test_url2", json=["my_str2"]) + + assert len(httpx_mock.get_requests(match_json=["my_str"])) == 2 + + +def test_requests_retrieval_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + with httpx.Client( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + client.get("https://test_url") + client.get("https://test_url2") + client.get("http://test_url2") + + assert ( + len(httpx_mock.get_requests(proxy_url="http://user:pwd@my_other_proxy/")) == 2 + ) + + +def test_request_retrieval_proxy_matching(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response() + + with httpx.Client( + proxies={ + "http://": "http://my_test_proxy", + "https://": "http://user:pwd@my_other_proxy", + } + ) as client: + client.get("https://test_url") + client.get("https://test_url2") + client.get("http://test_url2") + + assert httpx_mock.get_request(proxy_url="http://my_test_proxy/") + + def test_content_not_matching(httpx_mock: HTTPXMock) -> None: httpx_mock.add_response(match_content=b"This is the body")