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

【问题标题】：How to unit test a gRPC server's asynchronous methods?如何对 gRPC 服务器的异步方法进行单元测试？
【发布时间】：2021-03-31 22:04:10
【问题描述】：

按照this question 中的建议，我能够使用grpc_testing 库对我的gRPC 服务（使用grpc.aio API 构建）的同步方法进行单元测试。但是，当我在我的 gRPC 服务的异步方法上关注 this example 时，我得到：

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 代码库并进行更广泛的搜索，我找不到单元测试异步 gRPC 方法的示例。我能找到的最接近的东西是pytest-grpc 的unmerged branch，但the example service 没有任何异步方法。

谁能分享一个在 python 中对异步 gRPC 方法进行单元测试的示例？

【问题讨论】：

我用最简单的设置想出了答案How to do unit tests on python async functions?

标签： grpc-python

【解决方案1】：

我遵循@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

【讨论】：

【解决方案2】：

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

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

【讨论】：