Add local_address support for curio

Author: florimondmanca

httpcore/_backends/curio.py

@@ -155,9 +155,13 @@ async def open_tcp_stream(
             OSError: ConnectError,
         }
         host = hostname.decode("ascii")
-        kwargs = (
-            {} if ssl_context is None else {"ssl": ssl_context, "server_hostname": host}
-        )
+
+        kwargs: dict = {}
+        if ssl_context is not None:
+            kwargs["ssl"] = ssl_context
+            kwargs["server_hostname"] = host
+        if local_address is not None:
+            kwargs["source_addr"] = (local_address, 0)
 
         with map_exceptions(exc_map):
             sock: curio.io.Socket = await curio.timeout_after(

tests/async_tests/test_interfaces.py

@@ -6,6 +6,7 @@
 import httpcore
 from httpcore._types import URL
 from tests.conftest import Server
+from tests.utils import current_async_library
 
 
 async def read_body(stream: httpcore.AsyncByteStream) -> bytes:
@@ -210,9 +211,11 @@ async def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
         assert reason == b"OK"
 
 
-@pytest.mark.asyncio
-# This doesn't run with trio, since trio doesn't support local_address.
+@pytest.mark.usefixtures("async_environment")
 async def test_http_request_local_address() -> None:
+    if current_async_library() == "trio":
+        pytest.skip("trio does not support local_address.")
+
     async with httpcore.AsyncConnectionPool(local_address="0.0.0.0") as http:
         method = b"GET"
         url = (b"http", b"example.org", 80, b"/")

tests/sync_tests/test_interfaces.py

@@ -6,6 +6,7 @@
 import httpcore
 from httpcore._types import URL
 from tests.conftest import Server
+from tests.utils import current_async_library
 
 
 def read_body(stream: httpcore.SyncByteStream) -> bytes:
@@ -211,8 +212,10 @@ def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
 
 
 
-# This doesn't run with trio, since trio doesn't support local_address.
 def test_http_request_local_address() -> None:
+    if current_async_library() == "trio":
+        pytest.skip("trio does not support local_address.")
+
     with httpcore.SyncConnectionPool(local_address="0.0.0.0") as http:
         method = b"GET"
         url = (b"http", b"example.org", 80, b"/")

tests/utils.py

@@ -0,0 +1,10 @@
+import sniffio
+
+
+def current_async_library() -> str:
+    try:
+        library = sniffio.current_async_library()
+    except Exception:
+        return "sync"
+    else:
+        return library