Implementation:Bentoml BentoML GRPC Testing Utils

Knowledge Sources	Bentoml_BentoML
Domains	gRPC, Testing, Utilities
Last Updated	2026-02-13 15:00 GMT

Overview

Provides testing utilities for BentoML's gRPC services, including async client calls, NDArray protobuf helpers, channel creation, and standalone server setup.

Description

This module contains helper functions designed for testing BentoML gRPC services:

• create_test_bento_servicer -- Dynamically imports and creates a BentoML gRPC servicer for a given service, supporting different protocol versions.
• randomize_pb_ndarray -- Generates a protobuf NDArray message with random float values of a given shape.
• make_pb_ndarray -- Converts a NumPy array to a protobuf NDArray message, mapping numpy dtypes to protobuf dtype and field enums.
• async_client_call -- Sends a unary-unary RPC call to a BentoML gRPC service via a channel, with built-in assertion support for response data, status codes, details, and trailing metadata. Handles both successful responses and AioRpcError exceptions.
• create_channel -- An async context manager that creates an insecure gRPC channel with optional client interceptors, waits for the channel to be ready, and ensures proper cleanup.
• make_standalone_server -- A cached context manager that creates a standalone aio.Server on a dynamically reserved port with optional interceptors, suitable for integration tests.

Usage

Use these utilities in test suites that need to spin up BentoML gRPC servers, create client channels, and make assertions on RPC responses.

Code Reference

Source Location

• Repository: Bentoml_BentoML
• File: src/bentoml/testing/grpc/__init__.py
• Lines: 1-248

Signature

def create_test_bento_servicer(
    service: Service,
    protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> t.Callable[[Service], t.Any]: ...

def randomize_pb_ndarray(
    shape: tuple[int, ...],
    protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> pb.NDArray: ...

def make_pb_ndarray(
    arr: NDArray[t.Any],
    protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> pb.NDArray: ...

async def async_client_call(
    method: str,
    channel: Channel,
    data: dict[str, Message | pb.Part | bytes | str | dict[str, t.Any]],
    sanity: bool = True,
    timeout: int | None = 90,
    assert_data: pb.Response | t.Callable[[pb.Response], bool] | None = None,
    assert_code: grpc.StatusCode | None = None,
    assert_details: str | None = None,
    assert_trailing_metadata: aio.Metadata | None = None,
    protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> pb.Response | None: ...

@asynccontextmanager
async def create_channel(
    host_url: str,
    interceptors: t.Sequence[aio.ClientInterceptor] | None = None,
) -> t.AsyncGenerator[Channel, None]: ...

@cached_contextmanager("{interceptors}")
def make_standalone_server(
    interceptors: t.Sequence[aio.ServerInterceptor] | None = None,
    host: str = "0.0.0.0",
) -> t.Generator[tuple[aio.Server, str], None, None]: ...

Import

from bentoml.testing.grpc import (
    async_client_call,
    randomize_pb_ndarray,
    make_pb_ndarray,
    create_channel,
    make_standalone_server,
    create_test_bento_servicer,
)

I/O Contract

Inputs

Name	Type	Required	Description
method	str	Yes (async_client_call)	API method name to call
channel	Channel	Yes (async_client_call)	gRPC async channel for the call
data	dict	Yes (async_client_call)	Request data payload
assert_data	Response, Callable, or None	No	Expected response data or validation function
assert_code	grpc.StatusCode or None	No	Expected gRPC status code
shape	tuple[int, ...]	Yes (randomize_pb_ndarray)	Shape of the random NDArray to generate
arr	NDArray	Yes (make_pb_ndarray)	NumPy array to convert to protobuf
host_url	str	Yes (create_channel)	URL to connect the channel to
interceptors	Sequence or None	No	Server or client interceptors

Outputs

Name	Type	Description
pb.Response or None	pb.Response	Response from the gRPC call, or None if error with no assert_code
pb.NDArray	pb.NDArray	Protobuf NDArray message (randomize_pb_ndarray, make_pb_ndarray)
Channel	Channel	Async gRPC channel (create_channel)
(aio.Server, str)	tuple	Server and host URL string (make_standalone_server)

Usage Examples

import asyncio
import grpc
from bentoml.testing.grpc import (
    create_channel,
    async_client_call,
    make_standalone_server,
    randomize_pb_ndarray,
)

async def test_grpc_service():
    with make_standalone_server() as (server, host_url):
        await server.start()
        try:
            async with create_channel(host_url) as channel:
                # Create a random NDArray
                nd = randomize_pb_ndarray((1, 4))

                # Call the service
                resp = await async_client_call(
                    method="predict",
                    channel=channel,
                    data={"ndarray": nd},
                    assert_code=grpc.StatusCode.OK,
                )
        finally:
            await server.stop(None)

asyncio.run(test_grpc_service())

Related Pages

• Implementation:Bentoml_BentoML_GRPC_Utils
• Implementation:Bentoml_BentoML_GRPC_Access_Interceptor