1a6e254f72
* Added httpx.BaseTransport and httpx.AsyncBaseTransport * Test coverage and default transports to calling .close on __exit__ * BaseTransport documentation * Use 'handle_request' for the transport API. * Docs tweaks * Docs tweaks * Minor docstring tweak * Transport API docs * Drop 'Optional' on Transport API * Docs tweaks * Tweak CHANGELOG * Drop erronous example.py * Push httpcore exception wrapping out of client into transport (#1524) * Push httpcore exception wrapping out of client into transport * Include close/aclose extensions in docstring * Comment about the request property on RequestError exceptions * Extensions reason_phrase and http_version as bytes (#1526) * Extensions reason_phrase and http_version as bytes * Update BaseTransport docstring * Neaten up our try...except structure for ensuring responses (#1525) * Fix CHANGELOG typo Co-authored-by: Florimond Manca <florimond.manca@gmail.com> * Fix CHANGELOG typo Co-authored-by: Florimond Manca <florimond.manca@gmail.com> * stream: Iterator[bytes] -> stream: Iterable[bytes] * Use proper bytestream interfaces when calling into httpcore * Grungy typing workaround due to httpcore using Iterator instead of Iterable in bytestream types * Update docs/advanced.md Co-authored-by: Florimond Manca <florimond.manca@gmail.com> * Consistent typing imports across tranports * Update docs/advanced.md Co-authored-by: Florimond Manca <florimond.manca@gmail.com> Co-authored-by: Florimond Manca <florimond.manca@gmail.com>
96 lines
2.5 KiB
Python
96 lines
2.5 KiB
Python
from unittest import mock
|
|
|
|
import httpcore
|
|
import pytest
|
|
|
|
import httpx
|
|
from httpx._transports.default import HTTPCORE_EXC_MAP
|
|
|
|
|
|
def test_httpcore_all_exceptions_mapped() -> None:
|
|
"""
|
|
All exception classes exposed by HTTPCore are properly mapped to an HTTPX-specific
|
|
exception class.
|
|
"""
|
|
not_mapped = [
|
|
value
|
|
for name, value in vars(httpcore).items()
|
|
if isinstance(value, type)
|
|
and issubclass(value, Exception)
|
|
and value not in HTTPCORE_EXC_MAP
|
|
]
|
|
|
|
if not_mapped: # pragma: nocover
|
|
pytest.fail(f"Unmapped httpcore exceptions: {not_mapped}")
|
|
|
|
|
|
def test_httpcore_exception_mapping(server) -> None:
|
|
"""
|
|
HTTPCore exception mapping works as expected.
|
|
"""
|
|
|
|
def connect_failed(*args, **kwargs):
|
|
raise httpcore.ConnectError()
|
|
|
|
class TimeoutStream:
|
|
def __iter__(self):
|
|
raise httpcore.ReadTimeout()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
class CloseFailedStream:
|
|
def __iter__(self):
|
|
yield b""
|
|
|
|
def close(self):
|
|
raise httpcore.CloseError()
|
|
|
|
with mock.patch("httpcore.SyncConnectionPool.request", side_effect=connect_failed):
|
|
with pytest.raises(httpx.ConnectError):
|
|
httpx.get(server.url)
|
|
|
|
with mock.patch(
|
|
"httpcore.SyncConnectionPool.request",
|
|
return_value=(200, [], TimeoutStream(), {}),
|
|
):
|
|
with pytest.raises(httpx.ReadTimeout):
|
|
httpx.get(server.url)
|
|
|
|
with mock.patch(
|
|
"httpcore.SyncConnectionPool.request",
|
|
return_value=(200, [], CloseFailedStream(), {}),
|
|
):
|
|
with pytest.raises(httpx.CloseError):
|
|
httpx.get(server.url)
|
|
|
|
|
|
def test_httpx_exceptions_exposed() -> None:
|
|
"""
|
|
All exception classes defined in `httpx._exceptions`
|
|
are exposed as public API.
|
|
"""
|
|
|
|
not_exposed = [
|
|
value
|
|
for name, value in vars(httpx._exceptions).items()
|
|
if isinstance(value, type)
|
|
and issubclass(value, Exception)
|
|
and not hasattr(httpx, name)
|
|
]
|
|
|
|
if not_exposed: # pragma: nocover
|
|
pytest.fail(f"Unexposed HTTPX exceptions: {not_exposed}")
|
|
|
|
|
|
def test_request_attribute() -> None:
|
|
# Exception without request attribute
|
|
exc = httpx.ReadTimeout("Read operation timed out")
|
|
with pytest.raises(RuntimeError):
|
|
exc.request
|
|
|
|
# Exception with request attribute
|
|
request = httpx.Request("GET", "https://www.example.com")
|
|
exc = httpx.ReadTimeout("Read operation timed out", request=request)
|
|
assert exc.request == request
|