Skip to content
Connect
Created by

Testing

This guide covers testing Connect-Python services and clients.

Setup

Section titled “Setup”

For pytest examples in this guide, you’ll need pytest and pytest-asyncio. unittest requires no additional dependencies.

Section titled “Recommended approach: In-memory testing”

The recommended approach is in-memory testing using pyqwest’s ASGI/WSGI transports (provided by pyqwest, not Connect-Python). This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts.

Here’s a minimal example without any test framework:

from pyqwest import Client
from pyqwest.testing import ASGITransport
from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient
from greet.v1.greet_pb2 import GreetRequest
from server import Greeter  # Your service implementation


# Create ASGI app with your service
app = GreetServiceASGIApplication(Greeter())


# Connect client to service using in-memory transport. Unless you need to test
# ASGI lifespan, there is no need to use `async with`.
client = GreetServiceClient(
    "http://test",
    http_client=Client(ASGITransport(app))
)
response = await client.greet(GreetRequest(name="Alice"))


print(response.greeting)  # "Hello, Alice!"

This pattern works with any test framework (pytest, unittest) or none at all. The examples below show how to integrate with both pytest and unittest.

Testing servers

Section titled “Testing servers”

Using pytest

Section titled “Using pytest”

Testing the service we created in the Getting Started guide looks like this:

import pytest
from pyqwest import Client
from pyqwest.testing import ASGITransport
from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient
from greet.v1.greet_pb2 import GreetRequest
from server import Greeter  # Import your actual service implementation


@pytest.mark.asyncio
async def test_greet():
    # Create the ASGI application with your service
    app = GreetServiceASGIApplication(Greeter())


    client = GreetServiceClient(
        "http://test",
        http_client=Client(ASGITransport(app))
    )
    response = await client.greet(GreetRequest(name="Alice"))


    assert response.greeting == "Hello, Alice!"

Using unittest

Section titled “Using unittest”

The same in-memory testing approach works with unittest:

import asyncio
import unittest
from pyqwest import Client
from pyqwest.testing import ASGITransport
from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient
from greet.v1.greet_pb2 import GreetRequest
from server import Greeter


class TestGreet(unittest.TestCase):
    def test_greet(self):
        async def run_test():
            app = GreetServiceASGIApplication(Greeter())
            client = GreetServiceClient(
                "http://test",
                http_client=Client(ASGITransport(app))
            )
            response = await client.greet(GreetRequest(name="Alice"))
            self.assertEqual(response.greeting, "Hello, Alice!")


        asyncio.run(run_test())

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.

Using fixtures for reusable test setup

Section titled “Using fixtures for reusable test setup”

For cleaner tests, use pytest fixtures to set up clients and services:

from pyqwest import Client
from pyqwest.testing import ASGITransport
import pytest
import pytest_asyncio
from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient
from greet.v1.greet_pb2 import GreetRequest
from server import Greeter


@pytest_asyncio.fixture
async def greet_client():
    app = GreetServiceASGIApplication(Greeter())
    return GreetServiceClient(
        "http://test",
        http_client=Client(ASGITransport(app))
    )


@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_empty_name(greet_client):
    response = await greet_client.greet(GreetRequest(name=""))
    assert response.greeting == "Hello, !"

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

With your test client setup, you can use any Connect code for interacting with the service under test including streaming, reading headers and trailers, or checking errors. For example, to test error handling:

with pytest.raises(ConnectError) as exc_info:
    await client.greet(GreetRequest(name=""))


assert exc_info.value.code == Code.INVALID_ARGUMENT

See the Errors guide for more details on error handling.

Testing clients

Section titled “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 pyqwest transports to test your client logic without network overhead.

Example: Testing client error handling

Section titled “Example: Testing client error handling”
import pytest
from pyqwest import Client
from pyqwest.testing import ASGITransport
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


async def fetch_user_greeting(user_id: str, client: GreetServiceClient):
    """Client code that handles errors."""
    try:
        response = await client.greet(GreetRequest(name=user_id))
        return response.greeting
    except ConnectError as e:
        if e.code == Code.NOT_FOUND:
            return "User not found"
        elif e.code == Code.UNAUTHENTICATED:
            return "Please login"
        raise


@pytest.mark.asyncio
async def test_client_error_handling():
    class TestGreetService(GreetService):
        async def greet(self, request, ctx):
            if request.name == "unknown":
                raise ConnectError(Code.NOT_FOUND, "User not found")
            return GreetResponse(greeting=f"Hello, {request.name}!")


    app = GreetServiceASGIApplication(TestGreetService())
    client = GreetServiceClient(
        "http://test",
        http_client=Client(ASGITransport(app))
    )


    # Test successful case
    result = await fetch_user_greeting("Alice", client)
    assert result == "Hello, Alice!"


    # Test error handling
    result = await fetch_user_greeting("unknown", client)
    assert result == "User not found"

Testing interceptors

Section titled “Testing interceptors”

Test interceptors as part of your full application stack. For example, testing the ServerAuthInterceptor from the Interceptors guide:

import pytest
from pyqwest import Client
from pyqwest.testing import ASGITransport
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient
from greet.v1.greet_pb2 import GreetRequest
from interceptors import ServerAuthInterceptor
from server import Greeter


@pytest.mark.asyncio
async def test_server_auth_interceptor():
    interceptor = ServerAuthInterceptor(["valid-token"])
    app = GreetServiceASGIApplication(
        Greeter(),
        interceptors=[interceptor]
    )


    client = GreetServiceClient(
        "http://test",
        http_client=Client(ASGITransport(app))
    )


    # Valid token succeeds
    response = await client.greet(
        GreetRequest(name="Alice"),
        headers={"authorization": "Bearer valid-token"}
    )
    assert response.greeting == "Hello, Alice!"


    # Invalid token format fails with UNAUTHENTICATED
    with pytest.raises(ConnectError) as exc_info:
        await client.greet(
            GreetRequest(name="Bob"),
            headers={"authorization": "invalid"}
        )
    assert exc_info.value.code == Code.UNAUTHENTICATED


    # Wrong token fails with PERMISSION_DENIED
    with pytest.raises(ConnectError) as exc_info:
        await client.greet(
            GreetRequest(name="Bob"),
            headers={"authorization": "Bearer wrong-token"}
        )
    assert exc_info.value.code == Code.PERMISSION_DENIED

See the Interceptors guide for more details on implementing interceptors.