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

Question

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

假设我有以下服务器：

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和检查响应？

Answer 1

你可以在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!')

Answer 2

我采用了 JC 的想法并将其扩展为能够为每个测试用例创建一个假服务器（模拟）。另外，在端口 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!')

Answer 3

您可以使用 代码元素上的内联 API 文档字符串。有一个问题以良好的格式在 grpc.io 上托管： https://github.com/grpc/grpc/issues/13340