pytest-grpc-aio

Write tests for gRPC services with pytest using grpc.aio (asyncio-based gRPC).

Installation

pip install pytest-grpc-aio

Features

• Async/await support - Built on grpc.aio for native asyncio integration
• Flexible servicer configuration - Provide servicers via fixtures or at test time
• Real and fake servers - Run tests against actual gRPC servers or direct Python calls
• Context tracking - Access server/channel details via grpc_context fixture
• Type-safe - Full type hints with Protocol-based interfaces

Quick Start

1. Define Your Service

Given a proto file:

syntax = "proto3";

package example.v1;

service EchoService {
    rpc SayHello(HelloRequest) returns (HelloResponse) {}
}

message HelloRequest {
    string name = 1;
}

message HelloResponse {
    string message = 1;
}

Implement your servicer:

from example_pb2 import HelloRequest, HelloResponse
from example_pb2_grpc import EchoServiceServicer


class EchoServicer(EchoServiceServicer):
    async def SayHello(self, request: HelloRequest, context) -> HelloResponse:
        return HelloResponse(message=f"Hello, {request.name}!")

2. Configure Test Fixtures

import pytest
from example_pb2_grpc import add_EchoServiceServicer_to_server, EchoServiceStub
from servicer import EchoServicer


@pytest.fixture
def grpc_add_to_server():
    """Required: Tell pytest how to register your servicer."""
    return add_EchoServiceServicer_to_server


@pytest.fixture
def grpc_stub_cls():
    """Required: Tell pytest which stub class to use."""
    return EchoServiceStub


# Option 1: Provide servicer via fixture (module/session scope)
@pytest.fixture(scope="module")
def grpc_servicer():
    """Optional: Provide a default servicer for all tests."""
    return EchoServicer()

3. Write Tests

Using the Fixture-Provided Servicer

import pytest
from example_pb2 import HelloRequest


@pytest.mark.asyncio
async def test_say_hello(grpc_aio_stub):
    """Test using the servicer from grpc_servicer fixture."""
    async with grpc_aio_stub() as stub:
        request = HelloRequest(name="World")
        response = await stub.SayHello(request)
        assert response.message == "Hello, World!"

Providing Servicer Per-Test

@pytest.mark.asyncio
async def test_with_custom_servicer(grpc_aio_stub):
    """Override the servicer for a specific test."""
    custom_servicer = EchoServicer()

    async with grpc_aio_stub(servicer=custom_servicer) as stub:
        request = HelloRequest(name="Custom")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Custom!"

Testing Error Handling

import grpc


class ErrorServicer(EchoServiceServicer):
    async def SayHello(self, request: HelloRequest, context):
        await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid name")


@pytest.mark.asyncio
async def test_error_handling(grpc_aio_stub):
    """Test error scenarios."""
    error_servicer = ErrorServicer()

    async with grpc_aio_stub(servicer=error_servicer) as stub:
        request = HelloRequest(name="")

        with pytest.raises(grpc.RpcError) as exc_info:
            await stub.SayHello(request)

        assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT

Advanced Usage

Working with Credentials

import grpc
from pathlib import Path


@pytest.fixture
def my_channel_credentials():
    """Provide SSL credentials for secure channels."""
    cert_path = Path("/path/to/cert.pem")
    return grpc.ssl_channel_credentials(
        root_certificates=cert_path.read_bytes()
    )


@pytest.mark.asyncio
async def test_secure_connection(grpc_aio_stub, my_channel_credentials):
    """Test with SSL/TLS credentials."""
    async with grpc_aio_stub(credentials=my_channel_credentials) as stub:
        request = HelloRequest(name="Secure")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Secure!"

Using Channel Options

@pytest.mark.asyncio
async def test_with_options(grpc_aio_stub):
    """Test with custom channel options."""
    options = [
        ("grpc.max_receive_message_length", 1024 * 1024 * 10),
        ("grpc.max_send_message_length", 1024 * 1024 * 10),
    ]

    async with grpc_aio_stub(options=options) as stub:
        request = HelloRequest(name="Options")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Options!"

Accessing gRPC Context

The grpc_context fixture provides access to:

• grpc_context.addr: server address
• grpc_context.server: the gRPC server instance
• grpc_context.channel: the gRPC channel instance
• grpc_context.servicer: the servicer being tested
• grpc_context.add_to_server: the add_to_server function
• grpc_context.interceptors: any configured interceptors

@pytest.mark.asyncio
async def test_inspect_context(grpc_aio_stub, grpc_context):
    """Access server and channel details via grpc_context."""
    async with grpc_aio_stub() as stub:
        print(f"Testing against server at {grpc_context.addr}")

        request = HelloRequest(name="Context")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Context!"

Using Interceptors

import grpc.aio


class LoggingInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        print(f"Handling RPC: {handler_call_details.method}")
        return await continuation(handler_call_details)


@pytest.fixture
def grpc_interceptors():
    """Add server interceptors."""
    return [LoggingInterceptor()]

Lower-Level Access

If you need more control, you can use the channel or server fixtures directly:

@pytest.mark.asyncio
async def test_with_channel(grpc_aio_channel, grpc_stub_cls):
    """Use the channel fixture directly."""
    servicer = EchoServicer()

    async with grpc_aio_channel(servicer) as channel:
        stub = grpc_stub_cls(channel)
        request = HelloRequest(name="Channel")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Channel!"


@pytest.mark.asyncio
async def test_with_server(grpc_aio_server, grpc_addr, grpc_stub_cls):
    """Use the server fixture directly."""
    servicer = EchoServicer()

    async with grpc_aio_server(servicer):
        async with grpc.aio.insecure_channel(grpc_addr) as channel:
            stub = grpc_stub_cls(channel)
            request = HelloRequest(name="Server")
            response = await stub.SayHello(request)
            assert response.message == "Hello, Server!"

Command-Line Options

The plugin provides several command-line options:

Fake Server Mode

Run tests by calling service handlers directly (no real gRPC server):

pytest --grpc-fake-server

Benefits:

• Faster test execution
• Direct exception propagation (easier debugging)
• No network overhead

Trade-offs:

• Doesn't test actual gRPC serialization/networking
• May miss integration issues

Example output with fake server:

def test_error_handling(grpc_aio_stub):
    async with grpc_aio_stub(servicer=ErrorServicer()) as stub:
        request = HelloRequest(name="")
>       response = await stub.SayHello(request)

test_example.py:45:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    async def SayHello(self, request: HelloRequest, context):
>       await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid name")
E       FakeRpcError: Invalid name

servicer.py:12: FakeRpcError

Configuring Worker Threads

Control the ThreadPoolExecutor used by gRPC servers:

# Set maximum workers via command line
pytest --grpc-max-workers=10

# Or in a test module
grpc_max_workers = 10

The effective value is the maximum of the CLI option and module variable.

Use cases:

• Test thread-safety of servicers
• Simulate concurrent client requests
• Stress test resource locking

Available Fixtures

Required User Fixtures

Fixture	Scope	Description
grpc_add_to_server	Any	Function that registers your servicer with a gRPC server
grpc_stub_cls	Any	Your stub class (e.g., from *_pb2_grpc.py)

Optional User Fixtures

Fixture	Scope	Description
grpc_servicer	Any	Default servicer instance to use in tests
grpc_interceptors	Any	List of gRPC interceptors

Provided Fixtures

Fixture	Description
grpc_aio_stub	Most useful - Returns a factory for creating async stubs with context managers
grpc_aio_channel	Returns a factory for creating async channels with context managers
grpc_aio_server	Returns a factory for creating async servers with context managers
grpc_context	Provides access to server/channel details during tests
grpc_addr	The address where the test server is listening

Synchronous gRPC (grpc.Server)

For legacy/sync gRPC code, use the non-aio fixtures:

def test_sync(grpc_stub):
    """Use synchronous gRPC (no async/await)."""
    with grpc_stub() as stub:
        request = HelloRequest(name="Sync")
        response = stub.SayHello(request)
        assert response.message == "Hello, Sync!"

Available sync fixtures: grpc_stub, grpc_channel, grpc_server

Type Safety

All fixtures and factories are fully typed using typing.Protocol:

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from pytest_grpc_aio import GrpcAioStubFactory, GrpcContext

    # Your IDE will provide autocomplete and type checking
    stub_factory: GrpcAioStubFactory[EchoServiceStub, EchoServicer]
    context: GrpcContext

Migration from pytest-grpc

If you're migrating from the original pytest-grpc:

1. Use async fixtures: Replace grpc_stub with grpc_aio_stub
2. Use context managers: Stubs are now created via async with grpc_aio_stub() as stub:
3. Pass servicers explicitly: Can override grpc_servicer by passing servicer= argument
4. Add @pytest.mark.asyncio: Required for async test functions

Examples

See the example/ directory for complete working examples including:

• Basic echo service tests
• Error handling
• Secure connections
• Custom interceptors
• Thread-safety testing

Contributing

Contributions welcome! Please open an issue or PR on GitHub.

License

MIT