如何用 Python 为你的 GRPC 服务器编写单元测试？答案

【问题标题】：How to write unit tests for your GRPC server in Python?如何用 Python 为你的 GRPC 服务器编写单元测试？
【发布时间】：2019-01-18 10:17:07
【问题描述】：

我想使用 Python unittest 为我的 GRPC 服务器实现编写测试。我找到了 grpcio-testing 包，但我找不到任何文档如何使用它。

假设我有以下server：

import helloworld_pb2
import helloworld_pb2_grpc


class Greeter(helloworld_pb2_grpc.GreeterServicer):    
    def SayHello(self, request, context):
        return helloworld_pb2.HelloReply(message='Hello, %s!' % request.name)

如何创建单元测试来调用SayHello 并检查响应？

【问题讨论】：

将代码用作文档 - python 库主要是您可以阅读的 python 代码。这个问题在这里是题外话，因为你必须提供一些代码尝试并提出一个特定的问题。
代码没有文档。见here。我为服务器添加了代码。
@Mitar 我认为 nosklo 建议您深入研究 grpcio-testing 的代码以了解如何使用它。
是的，我已经这样做了，但我无法弄清楚。所以我来到这个很棒的网站寻求帮助。可能这个站点上的任何开源编程问题都可以用“阅读代码并弄清楚”来回答。我认为这个网站不存在这样的答案。

标签： python grpc

【解决方案1】：

您可以使用的代码元素上有inline API docstrings。在 grpc.io 上以一种不错的格式托管它存在一个问题：https://github.com/grpc/grpc/issues/13340

【讨论】：

是的，但我真的很感激一些例子或其他东西。
请在 GitHub 上提出问题。

【解决方案2】：

你可以在setUp时启动一个真实的服务器，在tearDown时停止服务器。

import unittest
from concurrent import futures


class RPCGreeterServerTest(unittest.TestCase):
    server_class = Greeter
    port = 50051

    def setUp(self):
        self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

        helloworld_pb2_grpc.add_GreeterServicer_to_server(self.server_class(), self.server)
        self.server.add_insecure_port(f'[::]:{self.port}')
        self.server.start()

    def tearDown(self):
        self.server.stop(None)

    def test_server(self):
        with grpc.insecure_channel(f'localhost:{self.port}') as channel:
            stub = helloworld_pb2_grpc.GreeterStub(channel)
            response = stub.SayHello(helloworld_pb2.HelloRequest(name='Jack'))
        self.assertEqual(response.message, 'Hello, Jack!')

【讨论】：

【解决方案3】：

我采纳了 J.C 的想法并将其扩展为能够为每个测试用例创建一个假服务器（模拟）。另外，绑定端口 0 以避免端口冲突：

@contextmanager
def helloworld(cls):
    """Instantiate a helloworld server and return a stub for use in tests"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    helloworld_pb2_grpc.add_GreeterServicer_to_server(cls(), server)
    port = server.add_insecure_port('[::]:0')
    server.start()

    try:
        with grpc.insecure_channel('localhost:%d' % port) as channel:
            yield helloworld_pb2_grpc.GreeterStub(channel)
    finally:
        server.stop(None)


class HelloWorldTest(unittest.TestCase):
    def test_hello_name(self):
        # may do something extra for this mock if it's stateful
        class FakeHelloworld(helloworld_pb2_grpc.GreeterServicer):
            def SayHello(self, request, context):
                return helloworld_pb2.SayHelloResponse()

        with helloworld(Fakehelloworld) as stub:
            response = stub.SayHello(helloworld_pb2.HelloRequest(name='Jack'))
            self.assertEqual(response.message, 'Hello, Jack!')

【讨论】：

【解决方案4】：

你可以试试pytest-grpc。

如果您使用的是 Django，可以查看django-grpc-framework testing。

【讨论】：