如何对 gRPC 服务器的异步方法进行单元测试？

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 中关于我的 gRPC 服务异步方法的建议，我得到：

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

查看 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 没有任何异步方法。

任何人都可以在 python 中分享单元测试异步 gRPC 方法的示例吗？

Answer 1

gRPC 测试是一个不错的项目。但我们需要工程资源使其支持 asyncio，最重要的是，采用现有的 API 来实现 asyncio 的理念。

为了测试 gRPC asyncio，我建议只使用 pytest，它有 pytest-asyncio 来顺利测试 asyncio 功能。这是一个例子：code.

Answer 2

我遵循了@Lidi 的建议（谢谢）并使用 pytest-asyncio 实现了测试。对于它的价值，这是一个测试异步流流方法的基本示例：

import mock
import grpc
import pytest

from tokenize_pb2 import MyMessage
from my_implementation import MyService

async def my_message_generator(messages):
    for message in messages:
        yield message

@pytest.mark.asyncio
async def test_my_async_stream_stream_method():
    service = MyService()
    my_messages = my_message_generator([MyMessage(), MyMessage()])
    mock_context = mock.create_autospec(spec=grpc.aio.ServicerContext)
    response = service.MyStreamStreamMethod(my_messages, mock_context)
    results = [x async for x in response]
    assert results == expected_results

如何对 gRPC 服务器的异步方法进行单元测试？

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

grpc-python