Run interceptor and compression tests once against sync and async clients

test/_util.py

@@ -1,14 +1,35 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import asyncio
+from collections.abc import AsyncIterator, Iterator
+from typing import TYPE_CHECKING, Any, Literal
+
+from pyqwest import Client, SyncClient
+from pyqwest.testing import ASGITransport, WSGITransport
 
 from connectrpc._compression import IdentityCompression
 from connectrpc.compression.brotli import BrotliCompression
 from connectrpc.compression.gzip import GzipCompression
 from connectrpc.compression.zstd import ZstdCompression
 
+from .connectrpc.example.haberdasher_connect import (
+    Haberdasher,
+    HaberdasherASGIApplication,
+    HaberdasherClient,
+    HaberdasherClientSync,
+    HaberdasherSync,
+    HaberdasherWSGIApplication,
+)
+
 if TYPE_CHECKING:
+    from collections.abc import Callable, Mapping
+
+    from pyqwest import SyncTransport, Transport
+
     from connectrpc.compression import Compression
+    from connectrpc.request import RequestContext
+
+    from .connectrpc.example.haberdasher_pb import Hat, Size
 
 
 def resolve_compression(encoding: str) -> Compression:
@@ -24,3 +45,75 @@ def resolve_compression(encoding: str) -> Compression:
         case _:
             msg = f"unknown encoding '{encoding}'"
             raise ValueError(msg)
+
+
+def haberdasher_client(transport: Transport, **kwargs) -> HaberdasherClient:
+    return HaberdasherClient(
+        "http://localhost", http_client=Client(transport), **kwargs
+    )
+
+
+def haberdasher_client_sync(
+    transport: SyncTransport, **kwargs
+) -> HaberdasherClientSync:
+    return HaberdasherClientSync(
+        "http://localhost", http_client=SyncClient(transport), **kwargs
+    )
+
+
+async def call(
+    client: HaberdasherClient | HaberdasherClientSync,
+    method: str,
+    request: Size | list[Size],
+    **kwargs,
+) -> Hat | list[Hat]:
+    """Calls method on client, passing and returning streams as lists.
+
+    A sync client runs in a worker thread, as it would in a sync program.
+    """
+    if isinstance(client, HaberdasherClientSync):
+
+        def run() -> Hat | list[Hat]:
+            req = iter(request) if isinstance(request, list) else request
+            result = getattr(client, method)(req, **kwargs)
+            return list(result) if isinstance(result, Iterator) else result
+
+        return await asyncio.to_thread(run)
+
+    async def stream(requests: list[Size]) -> AsyncIterator[Size]:
+        for r in requests:
+            yield r
+
+    result = getattr(client, method)(
+        stream(request) if isinstance(request, list) else request, **kwargs
+    )
+    if isinstance(result, AsyncIterator):
+        return [r async for r in result]
+    return await result
+
+
+def unary_client(
+    mode: Literal["async", "sync"],
+    make_hat: Callable[[Size, RequestContext], Hat],
+    app_options: Mapping[str, Any] | None = None,
+    **kwargs,
+) -> HaberdasherClient | HaberdasherClientSync:
+    """Returns a client of an ASGI or WSGI server whose MakeHat calls make_hat.
+
+    app_options are passed to the application and kwargs to the client.
+    """
+    if mode == "async":
+
+        class UnaryHaberdasher(Haberdasher):
+            async def make_hat(self, request, ctx):
+                return make_hat(request, ctx)
+
+        app = HaberdasherASGIApplication(UnaryHaberdasher(), **(app_options or {}))
+        return haberdasher_client(ASGITransport(app), **kwargs)
+
+    class UnaryHaberdasherSync(HaberdasherSync):
+        def make_hat(self, request, ctx):
+            return make_hat(request, ctx)
+
+    app = HaberdasherWSGIApplication(UnaryHaberdasherSync(), **(app_options or {}))
+    return haberdasher_client_sync(WSGITransport(app), **kwargs)

test/test_client.py

@@ -4,10 +4,8 @@
 
 import pytest
 from pyqwest import (
-    Client,
     Request,
     Response,
-    SyncClient,
     SyncRequest,
     SyncResponse,
     SyncTransport,
@@ -18,11 +16,10 @@
 from connectrpc.client import ResponseMetadata
 from connectrpc.protocol import ProtocolType
 
+from ._util import call, haberdasher_client, haberdasher_client_sync, unary_client
 from .connectrpc.example.haberdasher_connect import (
     Haberdasher,
     HaberdasherASGIApplication,
-    HaberdasherClient,
-    HaberdasherClientSync,
     HaberdasherSync,
     HaberdasherWSGIApplication,
 )
@@ -61,77 +58,28 @@
 ]
 
 
-@pytest.mark.parametrize(
-    ("headers", "trailers", "response_headers", "response_trailers"), _headers_cases
-)
-def test_headers_sync(headers, trailers, response_headers, response_trailers) -> None:
-    class HeadersHaberdasherSync(HaberdasherSync):
-        def __init__(
-            self, headers: list[tuple[str, str]], trailers: list[tuple[str, str]]
-        ) -> None:
-            self.headers = headers
-            self.trailers = trailers
-
-        def make_hat(self, _request, ctx):
-            for key, value in self.headers:
-                ctx.response_headers.add(key, value)
-            for key, value in self.trailers:
-                ctx.response_trailers.add(key, value)
-            return Hat()
-
-    transport = WSGITransport(
-        HaberdasherWSGIApplication(HeadersHaberdasherSync(headers, trailers))
-    )
-
-    client = HaberdasherClientSync(
-        "http://localhost", http_client=SyncClient(transport=transport)
-    )
-
-    with ResponseMetadata() as resp:
-        assert resp.http_status is None
-        assert list(resp.headers.allitems()) == []
-        assert list(resp.trailers.allitems()) == []
-        client.make_hat(Size(inches=10))
-
-    assert resp.http_status == 200
-    assert list(resp.headers.allitems()) == response_headers
-    assert list(resp.trailers.allitems()) == response_trailers
-
-
 @pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["async", "sync"])
 @pytest.mark.parametrize(
     ("headers", "trailers", "response_headers", "response_trailers"), _headers_cases
 )
-async def test_headers_async(
-    headers, trailers, response_headers, response_trailers
+async def test_headers(
+    mode, headers, trailers, response_headers, response_trailers
 ) -> None:
-    class HeadersHaberdasher(Haberdasher):
-        def __init__(
-            self, headers: list[tuple[str, str]], trailers: list[tuple[str, str]]
-        ) -> None:
-            self.headers = headers
-            self.trailers = trailers
-
-        async def make_hat(self, _request, ctx):
-            for key, value in self.headers:
-                ctx.response_headers.add(key, value)
-            for key, value in self.trailers:
-                ctx.response_trailers.add(key, value)
-            return Hat()
-
-    transport = ASGITransport(
-        HaberdasherASGIApplication(HeadersHaberdasher(headers, trailers))
-    )
+    def make_hat(_request, ctx):
+        for key, value in headers:
+            ctx.response_headers.add(key, value)
+        for key, value in trailers:
+            ctx.response_trailers.add(key, value)
+        return Hat()
 
-    client = HaberdasherClient(
-        "http://localhost", http_client=Client(transport=transport)
-    )
+    client = unary_client(mode, make_hat)
 
     with ResponseMetadata() as resp:
         assert resp.http_status is None
         assert list(resp.headers.allitems()) == []
         assert list(resp.trailers.allitems()) == []
-        await client.make_hat(Size(inches=10))
+        await call(client, "make_hat", Size(inches=10))
 
     assert resp.http_status == 200
     assert list(resp.headers.allitems()) == response_headers
@@ -176,9 +124,7 @@ async def sizes() -> AsyncIterator[Size]:
     transport = ContentRecorder(
         ASGITransport(HaberdasherASGIApplication(SizeHaberdasher()))
     )
-    async with HaberdasherClient(
-        "http://localhost", protocol=protocol, http_client=Client(transport=transport)
-    ) as client:
+    async with haberdasher_client(transport, protocol=protocol) as client:
         await client.make_hat(Size(inches=10))
         assert [hat async for hat in client.make_similar_hats(Size(inches=10))]
         await client.make_flexible_hat(sizes())
@@ -221,9 +167,7 @@ def execute_sync(self, request: SyncRequest) -> SyncResponse:
     transport = ContentRecorder(
         WSGITransport(HaberdasherWSGIApplication(SizeHaberdasherSync()))
     )
-    with HaberdasherClientSync(
-        "http://localhost", protocol=protocol, http_client=SyncClient(transport)
-    ) as client:
+    with haberdasher_client_sync(transport, protocol=protocol) as client:
         client.make_hat(Size(inches=10))
         assert list(client.make_similar_hats(Size(inches=10)))
         client.make_flexible_hat(iter([Size(inches=10)]))

test/test_codec.py

@@ -3,10 +3,8 @@
 import pytest
 from protobuf import Message
 from pyqwest import (
-    Client,
     Request,
     Response,
-    SyncClient,
     SyncRequest,
     SyncResponse,
     SyncTransport,
@@ -16,11 +14,10 @@
 
 from connectrpc.codec import Codec
 
+from ._util import haberdasher_client, haberdasher_client_sync
 from .connectrpc.example.haberdasher_connect import (
     Haberdasher,
     HaberdasherASGIApplication,
-    HaberdasherClient,
-    HaberdasherClientSync,
     HaberdasherSync,
     HaberdasherWSGIApplication,
 )
@@ -96,12 +93,7 @@ async def execute(self, request: Request) -> Response:
             HaberdasherASGIApplication(SimpleHaberdasher(), codecs=[CustomCodec()])
         )
     )
-    client = HaberdasherClient(
-        "http://localhost",
-        http_client=Client(transport),
-        codec=CustomCodec(),
-        send_compression=None,
-    )
+    client = haberdasher_client(transport, codec=CustomCodec(), send_compression=None)
 
     res = await client.make_hat(Size(inches=10))
     assert res.size == 10
@@ -137,11 +129,8 @@ def execute_sync(self, request: SyncRequest) -> SyncResponse:
             HaberdasherWSGIApplication(SimpleHaberdasherSync(), codecs=[CustomCodec()])
         )
     )
-    client = HaberdasherClientSync(
-        "http://localhost",
-        http_client=SyncClient(transport=transport),
-        codec=CustomCodec(),
-        send_compression=None,
+    client = haberdasher_client_sync(
+        transport, codec=CustomCodec(), send_compression=None
     )
 
     res = client.make_hat(Size(inches=10))