Search code examples
grpc-python

How to unit test a gRPC server's asynchronous methods?


Following the suggestions in this question I was able to unit test the synchronous methods of my gRPC service (which is built with the grpc.aio API) using the grpc_testing library. However, when I follow this example on an asynchronous method of my gRPC service I get:

ERROR    grpc_testing._server._rpc:_rpc.py:92 Exception calling application!
Traceback (most recent call last):
  File "/home/jp/venvs/grpc/lib/python3.8/site-packages/grpc_testing/_server/_service.py", line 63, in _stream_response
    response = copy.deepcopy(next(response_iterator))
TypeError: 'async_generator' object is not an iterator

Looking through the grpc_testing codebase and searching more broadly, I cannot find examples of unit testing async gRPC methods. The closest thing I could find is an unmerged branch of pytest-grpc, but the example service does not have any async methods.

Can anyone share an example of unit testing an asynchronous gRPC method in python?


Solution

  • gRPC Testing is a nice project. But we need engineering resources to make it support asyncio, and mostly importantly, adopt the existing APIs to asyncio's philosophy.

    For testing gRPC asyncio, I would recommend just use pytest which has pytest-asyncio to smoothly test out asyncio features. Here is an example: code.