测试 asyncio grpc 服务器（主要是关于 asyncio）答案

【问题标题】：Testing asyncio grpc server (mostly about asyncio)测试 asyncio grpc 服务器（主要是关于 asyncio）
【发布时间】：2021-10-08 15:55:59
【问题描述】：

免责声明：我是使用 asyncio 的新手，所以这可能是一个简单的解决方法。

我正在尝试为异步 grpc-server 的端点编写测试。服务器必须使用在无限循环中运行的函数定期检查某些内容，并且在无限循环处于休眠状态时仍然响应 - 这就是我使用 grpc-asyncio 和 pytest-asyncio 的原因。

示例测试（由 pytest-asyncio 创建的 event_loop）：

@pytest.mark.asyncio
async def test_endpoint(
        event_loop,
        test_client: test_pb2_grpc.TesterStub,
):
    await serve()  # THIS BLOCKS THE REST OF THE TEST
    response = await test_client.TemporaryEndpointForTesting(request=test_pb2.TmpRequest())
    assert response

客户端夹具：

@pytest.fixture
def test_client() -> test_pb2_grpc.TesterStub:
    port = 50551
    channel = aio.insecure_channel(f'localhost:{port}')
    stub = test_pb2_grpc.TesterStub(channel)
    return stub

服务器端点：

class Servicer(test_pb2_grpc.TesterServicer):

    # ENDPOINT
    async def TemporaryEndpointForTesting(self, request, context):
        print("request received!")
        return test_pb2.TmpResponse()

    async def infinite_loop(self):
        await asyncio.sleep(1.0)
        print("<looping>")
        return asyncio.ensure_future(self.infinite_loop())

服务器启动：

async def serve():
    port = 50551
    server: aio.Server = aio.server()
    servicer = Servicer()
    test_pb2_grpc.add_TesterServicer_to_server(servicer, server)
    server.add_insecure_port(f'[::]:{port}')

    task_1 = asyncio.create_task(servicer.infinite_loop())
    task_2 = asyncio.create_task(server.start())
    task_3 = asyncio.create_task(server.wait_for_termination())

    await task_1
    await task_2
    await task_3

目标是设置服务器，然后向它发送请求以查看它是否按预期响应。当我使用await serve() 单独启动服务器然后运行我的测试时，它似乎可以完美运行。但是当我尝试从测试用例启动它时，它卡住了......我有点明白，因为它正在等待（无限）服务器任务完成，但我不知道如何解决这个问题。我认为对服务器任务使用不同的event_loop 可以解决问题...

new_event_loop = asyncio.new_event_loop()
task_1 = new_event_loop.create_task(servicer.infinite_loop())
task_2 = new_event_loop.create_task(server.start())
task_3 = new_event_loop.create_task(server.wait_for_termination())

但这也没有用。

最好的情况是在夹具中启动服务器，这样我就可以将它传递给所有测试功能。我猜这也可以使用线程来完成，但考虑到服务器已经在使用asyncio，这似乎有点多余。

我整天都在这，任何帮助将不胜感激。

（使用 Python 3.9）

【问题讨论】：

您需要调整 serve 以使用毒丸方法，您可以从代码中指定将其关闭
但我不想在测试运行之前将其关闭。服务器需要启动才能响应
是的，你的测试完成后会给出毒丸

标签： python pytest grpc python-3.9 grpc-python

【解决方案1】：

这并不是我真正想要的解决方案，但我只是决定直接进行端到端测试，而不是试图解决这个问题。所以我使用Python Docker SDK 通过pytest fixture 启动服务器，然后向它发送客户端命令。或者，如果需要，我使用调试器启动它。

不像我希望的那样方便，但我已经在这个问题上花费了太多时间，这样它就可以立即进行 e2e 测试。

【讨论】：