From ecd8ed51dc842a8077835a5433cd8610c4975f21 Mon Sep 17 00:00:00 2001
From: i2y <6240399+i2y@users.noreply.github.com>
Date: Sat, 4 Oct 2025 14:04:13 +0900
Subject: [PATCH 1/2] Add doc on testing

Signed-off-by: i2y <6240399+i2y@users.noreply.github.com>
---
 docs/testing.md | 526 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 526 insertions(+)
 create mode 100644 docs/testing.md

diff --git a/docs/testing.md b/docs/testing.md
new file mode 100644
index 0000000..c7d093c
--- /dev/null
+++ b/docs/testing.md
@@ -0,0 +1,526 @@
+# Testing Guide
+
+This guide covers testing connect-python services and clients.
+
+> **Note:** The examples in this guide use a fictional `GreetService` for demonstration purposes. In your actual project, replace these with your own service definitions.
+
+## Setup
+
+Install the required testing dependencies:
+
+```bash
+pip install pytest pytest-asyncio httpx
+```
+
+Or if using uv:
+
+```bash
+uv add --dev pytest pytest-asyncio httpx
+```
+
+## Recommended approach: In-memory testing
+
+The recommended approach is **in-memory testing** using httpx's ASGI/WSGI transports. This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts.
+
+## Testing servers
+
+### In-memory testing
+
+Test services using httpx's ASGI/WSGI transport, which tests your full application stack while remaining fast and isolated:
+
+=== "ASGI"
+
+    ```python
+    import pytest
+    import httpx
+    from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient
+    from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+    class TestGreetService(GreetService):
+        async def greet(self, request, ctx):
+            return GreetResponse(greeting=f"Hello, {request.name}!")
+
+    @pytest.mark.asyncio
+    async def test_greet():
+        # Create the ASGI application
+        app = GreetServiceASGIApplication(TestGreetService())
+
+        # Test using httpx with ASGI transport
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url="http://test"
+        ) as session:
+            client = GreetServiceClient("http://test", session=session)
+            response = await client.greet(GreetRequest(name="Alice"))
+
+        assert response.greeting == "Hello, Alice!"
+    ```
+
+=== "WSGI"
+
+    ```python
+    import httpx
+    from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync
+    from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+    class TestGreetServiceSync(GreetServiceSync):
+        def greet(self, request, ctx):
+            return GreetResponse(greeting=f"Hello, {request.name}!")
+
+    def test_greet():
+        # Create the WSGI application
+        app = GreetServiceWSGIApplication(TestGreetServiceSync())
+
+        # Test using httpx with WSGI transport
+        with httpx.Client(
+            transport=httpx.WSGITransport(app=app),
+            base_url="http://test"
+        ) as session:
+            client = GreetServiceClientSync("http://test", session=session)
+            response = client.greet(GreetRequest(name="Alice"))
+
+        assert response.greeting == "Hello, Alice!"
+    ```
+
+This approach:
+
+- Tests your full application stack (routing, serialization, error handling)
+- Runs fast without network overhead
+- Provides isolation between tests
+- Works with all streaming types
+
+For integration tests with actual servers over TCP/HTTP, see standard pytest patterns for [server fixtures](https://docs.pytest.org/en/stable/how-to/fixtures.html).
+
+### Using fixtures for reusable test setup
+
+For cleaner tests, use pytest fixtures to set up clients and services:
+
+=== "ASGI"
+
+    ```python
+    import pytest
+    import pytest_asyncio
+    import httpx
+    from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient
+    from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+    class TestGreetService(GreetService):
+        async def greet(self, request, ctx):
+            return GreetResponse(greeting=f"Hello, {request.name}!")
+
+    @pytest_asyncio.fixture
+    async def greet_client():
+        app = GreetServiceASGIApplication(TestGreetService())
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url="http://test"
+        ) as session:
+            yield GreetServiceClient("http://test", session=session)
+
+    @pytest.mark.asyncio
+    async def test_greet(greet_client):
+        response = await greet_client.greet(GreetRequest(name="Alice"))
+        assert response.greeting == "Hello, Alice!"
+
+    @pytest.mark.asyncio
+    async def test_greet_multiple_names(greet_client):
+        response = await greet_client.greet(GreetRequest(name="Bob"))
+        assert response.greeting == "Hello, Bob!"
+    ```
+
+=== "WSGI"
+
+    ```python
+    import pytest
+    import httpx
+    from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync
+    from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+    class TestGreetServiceSync(GreetServiceSync):
+        def greet(self, request, ctx):
+            return GreetResponse(greeting=f"Hello, {request.name}!")
+
+    @pytest.fixture
+    def greet_client():
+        app = GreetServiceWSGIApplication(TestGreetServiceSync())
+        with httpx.Client(
+            transport=httpx.WSGITransport(app=app),
+            base_url="http://test"
+        ) as session:
+            yield GreetServiceClientSync("http://test", session=session)
+
+    def test_greet(greet_client):
+        response = greet_client.greet(GreetRequest(name="Alice"))
+        assert response.greeting == "Hello, Alice!"
+
+    def test_greet_multiple_names(greet_client):
+        response = greet_client.greet(GreetRequest(name="Bob"))
+        assert response.greeting == "Hello, Bob!"
+    ```
+
+This pattern:
+
+- Reduces code duplication across multiple tests
+- Makes tests more readable and focused on behavior
+- Follows pytest best practices
+- Matches the pattern used in connect-python's own test suite
+
+### Testing error handling
+
+Test that your service returns appropriate errors:
+
+```python
+from connectrpc.code import Code
+from connectrpc.errors import ConnectError
+
+class TestGreetService(GreetService):
+    async def greet(self, request, ctx):
+        if not request.name:
+            raise ConnectError(Code.INVALID_ARGUMENT, "name is required")
+        return GreetResponse(greeting=f"Hello, {request.name}!")
+
+@pytest.mark.asyncio
+async def test_greet_error():
+    app = GreetServiceASGIApplication(TestGreetService())
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        client = GreetServiceClient("http://test", session=session)
+
+        with pytest.raises(ConnectError) as exc_info:
+            await client.greet(GreetRequest(name=""))
+
+    assert exc_info.value.code == Code.INVALID_ARGUMENT
+    assert "name is required" in exc_info.value.message
+```
+
+### Testing streaming services
+
+For server streaming:
+
+```python
+@pytest.mark.asyncio
+async def test_server_streaming():
+    class StreamingGreetService(GreetService):
+        async def greet_stream(self, request, ctx):
+            for i in range(3):
+                yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!")
+
+    app = GreetServiceASGIApplication(StreamingGreetService())
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        client = GreetServiceClient("http://test", session=session)
+
+        responses = []
+        async for response in client.greet_stream(GreetRequest(name="Alice")):
+            responses.append(response)
+
+    assert len(responses) == 3
+    assert responses[0].greeting == "Hello Alice #1!"
+```
+
+For client streaming:
+
+```python
+@pytest.mark.asyncio
+async def test_client_streaming():
+    class ClientStreamingService(GreetService):
+        async def greet_many(self, request_stream, ctx):
+            names = []
+            async for req in request_stream:
+                names.append(req.name)
+            return GreetResponse(greeting=f"Hello, {', '.join(names)}!")
+
+    app = GreetServiceASGIApplication(ClientStreamingService())
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        client = GreetServiceClient("http://test", session=session)
+
+        async def request_stream():
+            yield GreetRequest(name="Alice")
+            yield GreetRequest(name="Bob")
+
+        response = await client.greet_many(request_stream())
+
+    assert "Alice" in response.greeting
+    assert "Bob" in response.greeting
+```
+
+### Testing with context (headers and trailers)
+
+Test code that uses request headers:
+
+```python
+class AuthGreetService(GreetService):
+    async def greet(self, request, ctx):
+        auth = ctx.request_headers().get("authorization")
+        if not auth or not auth.startswith("Bearer "):
+            raise ConnectError(Code.UNAUTHENTICATED, "Missing token")
+
+        ctx.response_headers()["greet-version"] = "v1"
+        return GreetResponse(greeting=f"Hello, {request.name}!")
+
+@pytest.mark.asyncio
+async def test_greet_with_headers():
+    app = GreetServiceASGIApplication(AuthGreetService())
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        client = GreetServiceClient("http://test", session=session)
+
+        response = await client.greet(
+            GreetRequest(name="Alice"),
+            headers={"authorization": "Bearer token123"}
+        )
+
+    assert response.greeting == "Hello, Alice!"
+```
+
+
+## Testing clients
+
+For testing client code that calls Connect services, use the same in-memory testing approach shown above. Create a test service implementation and use httpx transports to test your client logic without network overhead.
+
+## Testing interceptors
+
+### Testing with interceptors
+
+The recommended approach is to test interceptors as part of your full application stack:
+
+```python
+class LoggingInterceptor:
+    def __init__(self):
+        self.requests = []
+
+    async def on_start(self, ctx):
+        method_name = ctx.method().name
+        self.requests.append(method_name)
+        return method_name
+
+    async def on_end(self, token, ctx):
+        # token is the value returned from on_start
+        pass
+
+@pytest.mark.asyncio
+async def test_service_with_interceptor():
+    interceptor = LoggingInterceptor()
+    app = GreetServiceASGIApplication(
+        TestGreetService(),
+        interceptors=[interceptor]
+    )
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        client = GreetServiceClient("http://test", session=session)
+        await client.greet(GreetRequest(name="Alice"))
+
+    # Verify interceptor was called
+    assert "Greet" in interceptor.requests
+```
+
+## Test organization
+
+### Project structure
+
+Organize your tests in a `test/` directory at the root of your project:
+
+```
+my-project/
+├── greet/
+│   └── v1/
+│       ├── greet_connect.py
+│       └── greet_pb2.py
+├── test/
+│   ├── __init__.py
+│   ├── conftest.py          # Shared fixtures
+│   ├── test_greet.py        # Service tests
+│   └── test_integration.py  # Integration tests
+└── pyproject.toml
+```
+
+### Shared fixtures with conftest.py
+
+Use `conftest.py` to share fixtures across multiple test files:
+
+```python
+# test/conftest.py
+import pytest
+import pytest_asyncio
+import httpx
+from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient
+from greet.v1.greet_pb2 import GreetResponse
+
+class TestGreetService(GreetService):
+    async def greet(self, request, ctx):
+        return GreetResponse(greeting=f"Hello, {request.name}!")
+
+@pytest_asyncio.fixture
+async def greet_client():
+    """Shared client fixture available to all tests."""
+    app = GreetServiceASGIApplication(TestGreetService())
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        yield GreetServiceClient("http://test", session=session)
+```
+
+Then use it in any test file:
+
+```python
+# test/test_greet.py
+import pytest
+from greet.v1.greet_pb2 import GreetRequest
+
+@pytest.mark.asyncio
+async def test_greet(greet_client):
+    """Test basic greeting."""
+    response = await greet_client.greet(GreetRequest(name="Alice"))
+    assert response.greeting == "Hello, Alice!"
+```
+
+### Running tests
+
+Run all tests:
+
+```bash
+pytest
+```
+
+Run tests in a specific file:
+
+```bash
+pytest test/test_greet.py
+```
+
+Run a specific test:
+
+```bash
+pytest test/test_greet.py::test_greet
+```
+
+Run with verbose output:
+
+```bash
+pytest -v
+```
+
+## Practical examples
+
+### Testing with mock external dependencies
+
+Use fixtures to mock external services:
+
+```python
+import pytest
+import pytest_asyncio
+from unittest.mock import AsyncMock
+from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient
+from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+class DatabaseGreetService(GreetService):
+    def __init__(self, db):
+        self.db = db
+
+    async def greet(self, request, ctx):
+        # Fetch greeting from database
+        greeting_template = await self.db.get_greeting_template()
+        return GreetResponse(greeting=greeting_template.format(name=request.name))
+
+@pytest.fixture
+def mock_db():
+    """Mock database for testing."""
+    db = AsyncMock()
+    db.get_greeting_template.return_value = "Hello, {name}!"
+    return db
+
+@pytest_asyncio.fixture
+async def greet_client_with_db(mock_db):
+    app = GreetServiceASGIApplication(DatabaseGreetService(mock_db))
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        yield GreetServiceClient("http://test", session=session)
+
+@pytest.mark.asyncio
+async def test_greet_with_database(greet_client_with_db, mock_db):
+    response = await greet_client_with_db.greet(GreetRequest(name="Alice"))
+    assert response.greeting == "Hello, Alice!"
+    mock_db.get_greeting_template.assert_called_once()
+```
+
+### Testing authentication flows
+
+Test services that require authentication:
+
+```python
+import pytest
+import pytest_asyncio
+from connectrpc.code import Code
+from connectrpc.errors import ConnectError
+from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient
+from greet.v1.greet_pb2 import GreetRequest, GreetResponse
+
+class AuthGreetService(GreetService):
+    async def greet(self, request, ctx):
+        # Check for authorization header
+        auth = ctx.request_headers().get("authorization")
+        if not auth or not auth.startswith("Bearer "):
+            raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token")
+
+        # Validate token (simplified)
+        token = auth[7:]  # Remove "Bearer " prefix
+        if token != "valid-token":
+            raise ConnectError(Code.UNAUTHENTICATED, "Invalid token")
+
+        return GreetResponse(greeting=f"Hello, {request.name}!")
+
+@pytest_asyncio.fixture
+async def auth_greet_client():
+    app = GreetServiceASGIApplication(AuthGreetService())
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test"
+    ) as session:
+        yield GreetServiceClient("http://test", session=session)
+
+@pytest.mark.asyncio
+async def test_greet_with_valid_token(auth_greet_client):
+    response = await auth_greet_client.greet(
+        GreetRequest(name="Alice"),
+        headers={"authorization": "Bearer valid-token"}
+    )
+    assert response.greeting == "Hello, Alice!"
+
+@pytest.mark.asyncio
+async def test_greet_without_token(auth_greet_client):
+    with pytest.raises(ConnectError) as exc_info:
+        await auth_greet_client.greet(GreetRequest(name="Alice"))
+
+    assert exc_info.value.code == Code.UNAUTHENTICATED
+    assert "Missing or invalid token" in exc_info.value.message
+
+@pytest.mark.asyncio
+async def test_greet_with_invalid_token(auth_greet_client):
+    with pytest.raises(ConnectError) as exc_info:
+        await auth_greet_client.greet(
+            GreetRequest(name="Alice"),
+            headers={"authorization": "Bearer invalid-token"}
+        )
+
+    assert exc_info.value.code == Code.UNAUTHENTICATED
+    assert "Invalid token" in exc_info.value.message
+```

From 03f46560c1c49497bffdc719b155dc11f10efd51 Mon Sep 17 00:00:00 2001
From: "Anuraag (Rag) Agrawal" <anuraaga@gmail.com>
Date: Sat, 4 Oct 2025 13:07:34 +0900
Subject: [PATCH 2/2] Add doc about gRPC (lack of) compatibility and migration
 path (#20)

Signed-off-by: Anuraag Agrawal <anuraaga@gmail.com>
---
 docs/grpc-compatibility.md | 98 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 98 insertions(+)
 create mode 100644 docs/grpc-compatibility.md

diff --git a/docs/grpc-compatibility.md b/docs/grpc-compatibility.md
new file mode 100644
index 0000000..55b2e93
--- /dev/null
+++ b/docs/grpc-compatibility.md
@@ -0,0 +1,98 @@
+# gRPC compatibility
+
+Connect-Python currently does not support the gRPC protocol due to lack of support for HTTP/2 trailers
+in the Python ecosystem. If you have an existing codebase using grpc-python and want to introduce Connect
+in a transition without downtime, you will need a way for the gRPC servers to be accessible from both
+gRPC and Connect clients at the same time. Envoy is a widely used proxy server with support for translating
+the Connect protocol to gRPC via the [Connect-gRPC Bridge](https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/connect_grpc_bridge_filter).
+
+For example, if you have a gRPC server currently listening on port 8080, you update it to use port 8081
+and expose the service for both Connect and gRPC clients on port 8080 with this config.
+
+```yaml
+admin:
+  address:
+    socket_address: { address: 0.0.0.0, port_value: 9090 }
+
+static_resources:
+  listeners:
+    - name: listener_0
+      address:
+        socket_address: { address: 0.0.0.0, port_value: 8080 }
+      filter_chains:
+        - filters:
+            - name: envoy.filters.network.http_connection_manager
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+                stat_prefix: ingress_http
+                codec_type: AUTO
+                route_config:
+                  name: local_route
+                  virtual_hosts:
+                    - name: local_service
+                      domains: ["*"]
+                      routes:
+                        - match: { prefix: "/" }
+                          route: { cluster: service_0 }
+                http_filters:
+                  - name: envoy.filters.http.connect_grpc_bridge
+                    typed_config:
+                      "@type": type.googleapis.com/envoy.extensions.filters.http.connect_grpc_bridge.v3.FilterConfig
+                  - name: envoy.filters.http.grpc_web
+                    typed_config:
+                      "@type": type.googleapis.com/envoy.extensions.filters.http.grpc_web.v3.GrpcWeb
+                  - name: envoy.filters.http.router
+                    typed_config:
+                      "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+  clusters:
+    - name: service_0
+      connect_timeout: 0.25s
+      type: STRICT_DNS
+      lb_policy: ROUND_ROBIN
+      load_assignment:
+        cluster_name: service_0
+        endpoints:
+          - lb_endpoints:
+              - endpoint:
+                  address:
+                    socket_address:
+                      address: 127.0.0.1
+                      port_value: 8081
+      typed_extension_protocol_options:
+        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+          "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+          explicit_http_config:
+            http2_protocol_options:
+              max_concurrent_streams: 100
+```
+
+Refer to [Envoy docs](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/security/ssl) for more configuration such
+as TLS.
+
+## Migration
+
+Migrating from grpc-python to Connect using Envoy largely involves first adding Envoy in front of the server,
+then migrating clients to Connect, and finally migrating servers to Connect and removing Envoy.
+
+The general code structure of grpc-python and Connect are very similar - if your code is configured to use a
+type checker, any changes to parameter names and such should be quite easy to spot.
+
+1. Reconfigure your gRPC servers to include Envoy in front of the server port with a config similar to above.
+   For cloud deployments, this often means using functionality for sidecar containers.
+
+1. Begin generating code with `protoc-gen-connect-python`.
+
+1. Migrate clients to use Connect. Replace any special configuration of `ManagedChannel` with configured `httpx.Client` or
+   `httpx.AsyncClient` and switch to Connect's generated client types. If passing `metadata` at the call site, change
+   to `headers` - lists of string tuples can be passed directly to a `Headers` constructor, or can be changed to a raw
+   dictionary. Update any error handling to catch `ConnectError`.
+
+1. Complete deployment of all servers using Connect client. After this is done, your gRPC servers will only
+   be receiving traffic using the Connect protocol.
+
+1. Migrate service implementations to Connect generated stubs. It is recommended to extend the protocol classes
+   to have type checking find differences in method names. Change uses of `abort` to directly `raise ConnectError` -
+   for Connect services, it will be uncommon to pass the `RequestContext` into business logic code.
+
+1. Reconfigure server deployment to remove the Envoy proxy and deploy. You're done! You can stop generating code with
+   gRPC.