test: fix the warnings in the tests by not using deprecated APIs from httpx

Author: WSH032

tests/app/tool.py

@@ -213,7 +213,7 @@ def contx_socket_url(self) -> httpx.URL:
         )
 
 
-class TestServer:
+class AutoServer:
     """An AsyncContext to launch and shutdown Hypercorn or Uvicorn server automatically."""
 
     def __init__(

tests/conftest.py

@@ -31,7 +31,7 @@
 
 from .app.echo_http_app import get_app as get_http_test_app
 from .app.echo_ws_app import get_app as get_ws_test_app
-from .app.tool import AppDataclass4Test, TestServer
+from .app.tool import AppDataclass4Test, AutoServer
 
 # ASGI types.
 # Copied from: https://github.com/florimondmanca/asgi-lifespan/blob/fbb0f440337314be97acaae1a3c0c7a2ec8298dd/src/asgi_lifespan/_types.py
@@ -62,14 +62,14 @@ class LifeAppDataclass4Test(AppDataclass4Test):
 """The lifespan of app will be managed automatically by pytest."""
 
 
-class TestServerFixture(Protocol):  # noqa: D101
+class AutoServerFixture(Protocol):  # noqa: D101
     def __call__(  # noqa: D102
         self,
         app: FastAPI,
         host: str,
         port: int,
         server_type: Optional[Literal["uvicorn", "hypercorn"]] = None,
-    ) -> Coroutine[None, None, TestServer]: ...
+    ) -> Coroutine[None, None, AutoServer]: ...
 
 
 # https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on
@@ -203,22 +203,22 @@ def reverse_ws_app_fct(
 
 
 @pytest.fixture()
-async def test_server_fixture() -> AsyncIterator[TestServerFixture]:
-    """Fixture for TestServer.
+async def auto_server_fixture() -> AsyncIterator[AutoServerFixture]:
+    """Fixture for AutoServer.
 
     Will launch and shutdown automatically.
     """
     async with AsyncExitStack() as exit_stack:
 
-        async def test_server_fct(
+        async def auto_server_fct(
             app: FastAPI,
             host: str,
             port: int,
             server_type: Optional[Literal["uvicorn", "hypercorn"]] = None,
-        ) -> TestServer:
-            test_server = await exit_stack.enter_async_context(
-                TestServer(app=app, host=host, port=port, server_type=server_type)
+        ) -> AutoServer:
+            auto_server = await exit_stack.enter_async_context(
+                AutoServer(app=app, host=host, port=port, server_type=server_type)
             )
-            return test_server
+            return auto_server
 
-        yield test_server_fct
+        yield auto_server_fct

tests/test_core_lib.py

@@ -70,7 +70,10 @@ async def _() -> JSONResponse:
     #     }
     # }
 
-    client = httpx.AsyncClient(app=app, base_url="http://www.example.com")
+    client = httpx.AsyncClient(
+        transport=httpx.ASGITransport(app),  # pyright: ignore[reportArgumentType]
+        base_url="http://www.example.com",
+    )
     resp = await client.get("http://www.example.com/exception")
     assert resp.status_code == 0
     assert resp.json()["detail"] == test_err_msg

tests/test_http.py

@@ -32,7 +32,10 @@ async def tool_4_test_fixture(  # pyright: ignore[reportIncompatibleMethodOverri
     ) -> Tool4TestFixture:
         """目标服务器请参考`tests.app.echo_http_app.get_app`."""
         client_for_conn_to_target_server = httpx.AsyncClient(
-            app=echo_http_test_model.app, base_url=DEFAULT_TARGET_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                echo_http_test_model.app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_TARGET_SERVER_BASE_URL,
         )
 
         reverse_http_app = await reverse_http_app_fct(
@@ -41,7 +44,10 @@ async def tool_4_test_fixture(  # pyright: ignore[reportIncompatibleMethodOverri
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=reverse_http_app, base_url=DEFAULT_PROXY_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                reverse_http_app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_PROXY_SERVER_BASE_URL,
         )
 
         get_request = echo_http_test_model.get_request
@@ -198,7 +204,10 @@ async def test_bad_url_request(
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=reverse_http_app, base_url=DEFAULT_PROXY_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                reverse_http_app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_PROXY_SERVER_BASE_URL,
         )
 
         r = await client_for_conn_to_proxy_server.get(DEFAULT_PROXY_SERVER_BASE_URL)
@@ -233,9 +242,9 @@ async def test_cookie_leakage(
         assert not client_for_conn_to_proxy_server.cookies
 
         # check if cookie is not leaked
+        client_for_conn_to_proxy_server.cookies.set("a", "b")
         r = await client_for_conn_to_proxy_server.get(
-            proxy_server_base_url + "get/cookies",
-            cookies={"a": "b"},
+            proxy_server_base_url + "get/cookies"
         )
         assert "foo" not in r.json()  # not leaked
         assert r.json()["a"] == "b"  # send cookies normally
@@ -252,15 +261,21 @@ async def tool_4_test_fixture(  # pyright: ignore[reportIncompatibleMethodOverri
     ) -> Tool4TestFixture:
         """目标服务器请参考`tests.app.echo_http_app.get_app`."""
         client_for_conn_to_target_server = httpx.AsyncClient(
-            app=echo_http_test_model.app, base_url=DEFAULT_TARGET_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                echo_http_test_model.app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_TARGET_SERVER_BASE_URL,
         )
 
         forward_http_app = await forward_http_app_fct(
             client=client_for_conn_to_target_server, proxy_filter=default_proxy_filter
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=forward_http_app, base_url=DEFAULT_PROXY_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                forward_http_app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_PROXY_SERVER_BASE_URL,
         )
 
         get_request = echo_http_test_model.get_request
@@ -310,7 +325,10 @@ async def test_bad_url_request(
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=forward_http_app, base_url=DEFAULT_PROXY_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                forward_http_app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_PROXY_SERVER_BASE_URL,
         )
 
         # 错误的无法发出请求的URL
@@ -356,7 +374,10 @@ async def connect_error_mock_handler(
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=forward_http_app, base_url=DEFAULT_PROXY_SERVER_BASE_URL
+            transport=httpx.ASGITransport(
+                forward_http_app  # pyright: ignore[reportArgumentType]
+            ),
+            base_url=DEFAULT_PROXY_SERVER_BASE_URL,
         )
 
         r = await client_for_conn_to_proxy_server.get(
@@ -385,7 +406,9 @@ async def test_denial_http2(
         )
 
         client_for_conn_to_proxy_server = httpx.AsyncClient(
-            app=forward_http_app,
+            transport=httpx.ASGITransport(
+                forward_http_app
+            ),  # pyright: ignore[reportArgumentType]
             base_url=proxy_server_base_url,
             http2=True,
             http1=False,

tests/test_ws.py

@@ -3,7 +3,7 @@
 
 from contextlib import AsyncExitStack
 from multiprocessing import Process, Queue
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 
 import anyio
 import httpx
@@ -15,8 +15,8 @@
 from typing_extensions import override
 
 from .app.echo_ws_app import get_app as get_ws_test_app
-from .app.tool import TestServer
-from .conftest import TestServerFixture
+from .app.tool import AutoServer
+from .conftest import AutoServerFixture
 from .tool import (
     AbstractTestProxy,
     Tool4TestFixture,
@@ -25,7 +25,8 @@
 DEFAULT_HOST = "127.0.0.1"
 DEFAULT_PORT = 0  # random port
 
-# https://www.python-httpx.org/advanced/#http-proxying
+# https://www.python-httpx.org/advanced/proxies/
+# NOTE: Foce to connect directly, avoid using system proxies
 NO_PROXIES: Dict[Any, Any] = {"all://": None}
 
 
@@ -36,7 +37,7 @@ def _subprocess_run_echo_ws_server(queue: "Queue[str]"):
         queue: The queue for subprocess to put the url of echo ws app.
             After the server is started, the url will be put into the queue.
     """
-    target_ws_server = TestServer(
+    target_ws_server = AutoServer(
         app=get_ws_test_app().app,
         host=DEFAULT_HOST,
         port=DEFAULT_PORT,
@@ -55,29 +56,22 @@ async def run():
 
 def _subprocess_run_httpx_ws(
     queue: "Queue[str]",
-    kwargs_async_client: Optional[Dict[str, Any]] = None,
-    kwargs_aconnect_ws: Optional[Dict[str, Any]] = None,
+    aconnect_ws_url: str,
 ):
     """Run aconnect_ws in subprocess.
 
     Args:
         queue: The queue for subprocess to put something for flag of ws connection established.
-        kwargs_async_client: The kwargs for `httpx.AsyncClient`
-        kwargs_aconnect_ws: The kwargs for `httpx_ws.aconnect_ws`
+        aconnect_ws_url: The websocket url for aconnect_ws.
     """
-    kwargs_async_client = kwargs_async_client or {}
-    kwargs_aconnect_ws = kwargs_aconnect_ws or {}
-
-    kwargs_async_client.pop("proxies", None)
-    kwargs_aconnect_ws.pop("client", None)
 
     async def run():
         _exit_stack = AsyncExitStack()
-        _temp_client = httpx.AsyncClient(proxies=NO_PROXIES, **kwargs_async_client)
+        _temp_client = httpx.AsyncClient(mounts=NO_PROXIES)
         _ = await _exit_stack.enter_async_context(
             aconnect_ws(
                 client=_temp_client,
-                **kwargs_aconnect_ws,
+                url=aconnect_ws_url,
             )
         )
         queue.put("done")
@@ -95,32 +89,32 @@ class TestReverseWsProxy(AbstractTestProxy):
     @pytest.fixture()
     async def tool_4_test_fixture(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
-        test_server_fixture: TestServerFixture,
+        auto_server_fixture: AutoServerFixture,
     ) -> Tool4TestFixture:
         """目标服务器请参考`tests.app.echo_ws_app.get_app`."""
         echo_ws_test_model = get_ws_test_app()
         echo_ws_app = echo_ws_test_model.app
         echo_ws_get_request = echo_ws_test_model.get_request
 
-        target_ws_server = await test_server_fixture(
+        target_ws_server = await auto_server_fixture(
             app=echo_ws_app, port=DEFAULT_PORT, host=DEFAULT_HOST
         )
 
         target_server_base_url = str(target_ws_server.contx_socket_url)
 
-        client_for_conn_to_target_server = httpx.AsyncClient(proxies=NO_PROXIES)
+        client_for_conn_to_target_server = httpx.AsyncClient(mounts=NO_PROXIES)
 
         reverse_ws_app = get_reverse_ws_app(
             client=client_for_conn_to_target_server, base_url=target_server_base_url
         )
 
-        proxy_ws_server = await test_server_fixture(
+        proxy_ws_server = await auto_server_fixture(
             app=reverse_ws_app, port=DEFAULT_PORT, host=DEFAULT_HOST
         )
 
         proxy_server_base_url = str(proxy_ws_server.contx_socket_url)
 
-        client_for_conn_to_proxy_server = httpx.AsyncClient(proxies=NO_PROXIES)
+        client_for_conn_to_proxy_server = httpx.AsyncClient(mounts=NO_PROXIES)
 
         return Tool4TestFixture(
             client_for_conn_to_target_server=client_for_conn_to_target_server,
@@ -189,18 +183,11 @@ async def test_ws_proxy(self, tool_4_test_fixture: Tool4TestFixture) -> None:
         # 是因为这里已经有现成的target server，放在这里测试可以节省启动服务器时间
 
         aconnect_ws_subprocess_queue: "Queue[str]" = Queue()
-
-        kwargs_async_client = {"proxies": NO_PROXIES}
-        kwargs_aconnect_ws = {"url": proxy_server_base_url + "do_nothing"}
-        kwargs = {
-            "kwargs_async_client": kwargs_async_client,
-            "kwargs_aconnect_ws": kwargs_aconnect_ws,
-        }
+        aconnect_ws_url = proxy_server_base_url + "do_nothing"
 
         aconnect_ws_subprocess = Process(
             target=_subprocess_run_httpx_ws,
-            args=(aconnect_ws_subprocess_queue,),
-            kwargs=kwargs,
+            args=(aconnect_ws_subprocess_queue, aconnect_ws_url),
         )
         aconnect_ws_subprocess.start()
 
@@ -246,13 +233,13 @@ async def test_target_server_shutdown_abnormally(self) -> None:
             await anyio.sleep(0.1)
         target_server_base_url = subprocess_queue.get()
 
-        client_for_conn_to_target_server = httpx.AsyncClient(proxies=NO_PROXIES)
+        client_for_conn_to_target_server = httpx.AsyncClient(mounts=NO_PROXIES)
 
         reverse_ws_app = get_reverse_ws_app(
             client=client_for_conn_to_target_server, base_url=target_server_base_url
         )
 
-        async with TestServer(
+        async with AutoServer(
             app=reverse_ws_app,
             port=DEFAULT_PORT,
             host=DEFAULT_HOST,
@@ -261,10 +248,10 @@ async def test_target_server_shutdown_abnormally(self) -> None:
 
             async with aconnect_ws(
                 proxy_server_base_url + "do_nothing",
-                httpx.AsyncClient(proxies=NO_PROXIES),
+                httpx.AsyncClient(mounts=NO_PROXIES),
             ) as ws0, aconnect_ws(
                 proxy_server_base_url + "do_nothing",
-                httpx.AsyncClient(proxies=NO_PROXIES),
+                httpx.AsyncClient(mounts=NO_PROXIES),
             ) as ws1:
                 # force shutdown target server
                 target_ws_server_subprocess.terminate()