Skip to content

Testing grpc.aio Services

A gRPC service can be tested at three levels: calling the servicer's methods directly with a fake context, running a real in-process server and calling it through a real channel, or deploying it and testing over the network. The first two belong in the unit test suite, and they are fast. Measured with grpcio 1.84 and pytest-asyncio on 500 tests of the same unary method: calling the servicer directly took 0.47 s for the whole run, a real server and channel shared per module took 0.61 s, and a fresh server per test took 1.75 s — about 2.6 ms of setup per test. Even the slowest option is cheap enough to use freely, and only the real server exercises serialization, status codes, metadata and interceptors. This guide sets up each level with pytest and shows what each one can and cannot catch.

Prerequisites

1. Structure the servicer for testing

Keep the servicer thin and inject its dependencies, so tests can substitute fakes without patching:

import grpc

import greet_pb2
import greet_pb2_grpc


class Greeter(greet_pb2_grpc.GreeterServicer):
    def __init__(self, repo) -> None:
        self.repo = repo                                   # injected: a real DB repo or a fake

    async def Hello(self, request, context):
        if not request.name:
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "name required")
        title = await self.repo.title(request.name)
        return greet_pb2.HelloReply(msg=f"hi {title}")


class FakeRepo:
    async def title(self, name: str) -> str:
        return name.title()

context.abort raises inside the servicer to end the call with a status code; in a real server the framework turns it into the client's AioRpcError. That one behaviour is what makes direct tests and server tests differ, as the next two steps show.

Verify: the servicer can be constructed in a test with nothing but fakes.

2. Call the servicer directly for logic tests

For business logic, call the method as a coroutine with a mock context:

from unittest.mock import AsyncMock, MagicMock


async def test_hello_titles_the_name():
    reply = await Greeter(FakeRepo()).Hello(greet_pb2.HelloRequest(name="ada"), MagicMock())
    assert reply.msg == "hi Ada"


class Aborted(Exception):
    pass


async def test_empty_name_aborts():
    context = MagicMock()
    context.abort = AsyncMock(side_effect=Aborted)     # real abort raises; the mock must too
    with pytest.raises(Aborted):
        await Greeter(FakeRepo()).Hello(greet_pb2.HelloRequest(name=""), context)
    context.abort.assert_awaited_once_with(grpc.StatusCode.INVALID_ARGUMENT, "name required")

The side_effect matters: a plain AsyncMock returns instead of raising, so the servicer would carry on past the abort and the test would pass for the wrong reason. Measured, 500 direct tests ran in 0.47 s. Direct tests do not exercise protobuf serialization, metadata, deadlines, interceptors or how the framework maps exceptions to status codes — that needs a real server.

Verify: each validation branch has a direct test that asserts the exact status code and message passed to abort.

500 tests of one unary method, by test style 3 horizontal bars comparing servicer called directly with the others. 500 tests of one unary method, by test style servicer called directly 0.47 s real server, per module 0.61 s real server, per test 1.75 s grpcio 1.84, pytest-asyncio, includes pytest startup; per-test servers cost about 2.6 ms each. Real servers are cheap enough to use for anything that crosses the wire.

3. Run a real server on a free port

For tests that should see what clients see, start the server in-process on port 0 and connect a real channel to the port it chose:

import pytest_asyncio


@pytest_asyncio.fixture
async def greeter_stub():
    server = grpc.aio.server()
    greet_pb2_grpc.add_GreeterServicer_to_server(Greeter(FakeRepo()), server)
    port = server.add_insecure_port("127.0.0.1:0")       # 0 = let the OS choose
    await server.start()
    async with grpc.aio.insecure_channel(f"127.0.0.1:{port}") as channel:
        yield greet_pb2_grpc.GreeterStub(channel)
    await server.stop(None)


async def test_invalid_name_status(greeter_stub):
    with pytest.raises(grpc.aio.AioRpcError) as exc_info:
        await greeter_stub.Hello(greet_pb2.HelloRequest(name=""))
    assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT
    assert exc_info.value.details() == "name required"

add_insecure_port returns the bound port, so tests never collide on fixed ports when run in parallel. The test asserts on the status code and details as a client would receive them, after the full round trip through serialization and the gRPC core. Measured, a server per test added about 2.6 ms per test; for most suites that is acceptable and gives each test a clean server.

Verify: the suite runs in parallel (pytest -n auto) without port conflicts.

4. Share a server per module when tests are many

When a module has hundreds of tests against the same servicer, start one server per module. With pytest-asyncio, the fixture and the tests must share the module's event loop:

@pytest_asyncio.fixture(scope="module", loop_scope="module")
async def shared_stub():
    server = grpc.aio.server()
    greet_pb2_grpc.add_GreeterServicer_to_server(Greeter(FakeRepo()), server)
    port = server.add_insecure_port("127.0.0.1:0")
    await server.start()
    async with grpc.aio.insecure_channel(f"127.0.0.1:{port}") as channel:
        yield greet_pb2_grpc.GreeterStub(channel)
    await server.stop(None)


pytestmark = pytest.mark.asyncio(loop_scope="module")      # tests run on the module's loop

Measured, 500 tests took 0.61 s against 1.75 s with a server per test. Both the fixture's loop_scope and the tests' marker are needed: a grpc.aio server and channel are bound to the event loop that created them, and a test running on a different loop fails with errors about futures attached to a different loop. Shared servers need stateless fakes, or tests that reset state, so test order does not matter — the general pattern is in Testing Async Code.

Verify: running the module's tests in random order (pytest -p random_order) still passes.

What each test level exercises A grid of 3 rows by 3 columns. What each test level exercises level exercises misses servicer called directly logic, abort arguments serialization, metadata, interceptors in-process server wire format, status codes, deadlines TLS, LB, real dependencies deployed service the whole path speed; use sparingly Most tests belong in the first two rows.

5. Test deadlines, metadata and streams through the server

The behaviours that only exist on the wire deserve explicit tests against the in-process server:

async def test_deadline_is_enforced(greeter_stub_with_slow_repo):
    with pytest.raises(grpc.aio.AioRpcError) as exc_info:
        await greeter_stub_with_slow_repo.Hello(greet_pb2.HelloRequest(name="ada"), timeout=0.05)
    assert exc_info.value.code() == grpc.StatusCode.DEADLINE_EXCEEDED


async def test_auth_metadata_required(secured_stub):
    with pytest.raises(grpc.aio.AioRpcError) as exc_info:
        await secured_stub.Hello(greet_pb2.HelloRequest(name="ada"))           # no token
    assert exc_info.value.code() == grpc.StatusCode.UNAUTHENTICATED
    reply = await secured_stub.Hello(greet_pb2.HelloRequest(name="ada"),
                                     metadata=[("authorization", "Bearer test-token")])
    assert reply.msg == "hi Ada"

Build fixture variants by parameterizing the servicer's dependencies (a slow fake repo) and the server's interceptors (the auth interceptor from securing grpc.aio with TLS and auth metadata). Streaming methods are tested the same way: iterate the response stream in the test and assert on the messages and on what happens when the client cancels mid-stream, following streaming RPCs with grpc.aio.

Verify: removing the auth interceptor or the deadline handling makes a test fail.

Which level should this test use? A decision on What is being tested with 3 outcomes. Which level should this test use? What is being tested? business logic, validation call servicer directly fastest status codes, metadata, deadlines, streams in-process server port 0 TLS, LB, real infra deployed smoke tests few Test each behaviour at the cheapest level that can observe it.

Verification

A grpc.aio test suite is sound when:

  • Logic is tested directly with injected fakes and a raising abort mock.
  • Wire behaviour is tested through an in-process server on port 0.
  • Fixtures and tests share an event loop when servers are shared.
  • Deadlines, metadata, interceptors and streams each have a test against the server.

Diagnostic Hook: run the suite with --durations=20. Fixture setup dominating the slowest tests means servers are created per test where a module fixture would do; tests that fail only when run together mean a shared server holds state between tests.

Pitfalls & edge cases

  • A non-raising abort mock. The servicer continues past the abort and the test passes wrongly.
  • Fixed ports. Parallel runs collide; bind to port 0.
  • Mismatched event loops. Shared servers need loop_scope on fixture and tests.
  • Only direct tests. Serialization, status mapping and interceptors go untested.

Frequently Asked Questions

How do I unit test a grpc.aio servicer?

Call its methods directly as coroutines with a request message and a MagicMock context, making context.abort an AsyncMock that raises. In testing, 500 such tests ran in 0.47 s.

How do I run a gRPC server inside a pytest test?

Create grpc.aio.server(), add the servicer, bind with add_insecure_port("127.0.0.1:0") to get a free port, start it in an async fixture, and yield a stub on a channel to that port.

Is starting a gRPC server per test too slow?

Usually not: it added about 2.6 ms per test in testing. Share one per module, with matching loop_scope, when a module has hundreds of tests.

How do I assert a gRPC status code in a test?

Catch grpc.aio.AioRpcError from the stub call with pytest.raises and compare exc.code() to the expected grpc.StatusCode and exc.details() to the message.