Drop private imports from test_exceptions.py (#2571)

diff --git a/tests/conftest.py b/tests/conftest.py
index 4479b2b084cb06388d12415ada4c0c25dad43853..d33ce9fb2803fcc89c7195e77bed42e14533d468 100644 (file)
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -97,7 +97,6 @@ async def hello_world_json(scope: _Scope, receive: "_Receive", send: "_Send") ->
 
 
 async def slow_response(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
-    await sleep(1.0)
     await send(
         {
             "type": "http.response.start",
@@ -105,6 +104,7 @@ async def slow_response(scope: _Scope, receive: "_Receive", send: "_Send") -> No
             "headers": [[b"content-type", b"text/plain"]],
         }
     )
+    await sleep(1.0)  # Allow triggering a read timeout.
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 

diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py
index b765d46a2d8916763394b93fc0213d4c32f48c88..e17bc1ae95c706a523ee4a4d8aedf7c3d9464594 100644 (file)
--- a/tests/test_exceptions.py
+++ b/tests/test_exceptions.py
@@ -1,11 +1,9 @@
 import typing
-from unittest import mock
 
 import httpcore
 import pytest
 
 import httpx
-from httpx._transports.default import HTTPCORE_EXC_MAP
 
 if typing.TYPE_CHECKING:  # pragma: no cover
     from conftest import TestServer
@@ -16,66 +14,39 @@ 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()
+    expected_mapped_httpcore_exceptions = {
+        value.__name__
+        for _, value in vars(httpcore).items()
         if isinstance(value, type)
         and issubclass(value, Exception)
-        and value not in HTTPCORE_EXC_MAP
         and value is not httpcore.ConnectionNotAvailable
-    ]
+    }
 
-    if not_mapped:  # pragma: no cover
-        pytest.fail(f"Unmapped httpcore exceptions: {not_mapped}")
+    httpx_exceptions = {
+        value.__name__
+        for _, value in vars(httpx).items()
+        if isinstance(value, type) and issubclass(value, Exception)
+    }
 
+    unmapped_exceptions = expected_mapped_httpcore_exceptions - httpx_exceptions
 
-def test_httpcore_exception_mapping(server: "TestServer") -> 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
-
-    with mock.patch(
-        "httpcore.ConnectionPool.handle_request", side_effect=connect_failed
-    ):
-        with pytest.raises(httpx.ConnectError):
-            httpx.get(server.url)
+    if unmapped_exceptions:  # pragma: no cover
+        pytest.fail(f"Unmapped httpcore exceptions: {unmapped_exceptions}")
 
-    with mock.patch(
-        "httpcore.ConnectionPool.handle_request",
-        return_value=httpcore.Response(
-            200, headers=[], content=TimeoutStream(), extensions={}
-        ),
-    ):
-        with pytest.raises(httpx.ReadTimeout):
-            httpx.get(server.url)
 
-
-def test_httpx_exceptions_exposed() -> None:
+def test_httpcore_exception_mapping(server: "TestServer") -> None:
     """
-    All exception classes defined in `httpx._exceptions`
-    are exposed as public API.
+    HTTPCore exception mapping works as expected.
     """
-
-    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: no cover
-        pytest.fail(f"Unexposed HTTPX exceptions: {not_exposed}")
+    impossible_port = 123456
+    with pytest.raises(httpx.ConnectError):
+        httpx.get(server.url.copy_with(port=impossible_port))
+
+    with pytest.raises(httpx.ReadTimeout):
+        httpx.get(
+            server.url.copy_with(path="/slow_response"),
+            timeout=httpx.Timeout(5, read=0.01),
+        )
 
 
 def test_request_attribute() -> None: