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



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

假设我有以下服务器:

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并检查响应?

您可以在设置时启动真实服务器,并在拆卸时停止服务器。

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

我接受了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!')

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

你可以试试pytest-grpc。

如果你使用的是 Django,你可以看看 django-grpc-framework 测试。

最新更新