测试asyncio grpc服务器(主要是关于asyncio)

Question

免责声明:我刚开始使用asyncio，所以这可能是一个容易的修复。

我正在尝试为异步grpc服务器的端点编写测试。服务器必须使用在无限循环中运行的函数定期检查一些东西，并且在无限循环休眠时仍然响应-这就是为什么我使用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())

但这也不起作用。

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

我一整天都在忙这个，任何帮助都将是非常感激的。

(使用Python 3.9)

Answer 1

这不是我真正想要的解决方案，但我只是决定直接进行端到端测试，而不是试图弄清楚这一点。因此，我使用Python Docker SDK通过pytest fixture启动服务器，并向其发送客户端命令。如果需要的话，我也可以使用调试器来启动它。

不像我想的那么方便，但是我已经在这个问题上花了太多时间了，这样就可以马上测试了。