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

Question

我想使用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并检查响应？

Answer 1

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

Answer 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!')

Answer 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!')