grpcio-testing

import unittest import grpc from grpc_testing import server_from_dictionary, strict_real_time # Assume you have a compiled protobuf service 'helloworld_pb2_grpc.py' # and message types 'helloworld_pb2.py' # For this example, we'll mock them: class MockHelloRequest: def __init__(self, name): self.name = name class MockHelloReply: def __init__(self, message): self.message = message class MockGreeterServicer: def SayHello(self, request, context): if not request.name: context.set_code(grpc.StatusCode.INVALID_ARGUMENT) context.set_details('Name cannot be empty!') return MockHelloReply(message='') return MockHelloReply(message=f'Hello, {request.name}!') # Mock descriptor for server_from_dictionary # In a real scenario, this would come from your generated _pb2_grpc.py def get_method_descriptor(self, method_name): if method_name == '/Greeter/SayHello': # This is a simplified representation; actual descriptor is complex return MockMethodDescriptor('/Greeter/SayHello', MockHelloRequest, MockHelloReply) return None class MockMethodDescriptor: def __init__(self, name, request_type, response_type): self.full_method = name self.input_type = request_type self.output_type = response_type self.has_request_stream = False self.has_response_stream = False class TestGreeterService(unittest.TestCase): def setUp(self): # In a real application, replace with actual generated descriptors # from helloworld_pb2_grpc.DESCRIPTOR or similar. mock_servicer = MockGreeterServicer() service_descriptors = { '/Greeter/SayHello': grpc.unary_unary_rpc_method_handler( mock_servicer.SayHello, request_deserializer=lambda x: MockHelloRequest(''), # Placeholder response_serializer=lambda x: x.message.encode() # Placeholder ) } self.test_server = server_from_dictionary(service_descriptors, strict_real_time()) def test_say_hello_success(self): method_descriptor = self.test_server.get_method_descriptor('/Greeter/SayHello') request = MockHelloRequest(name='World') response, _, code, _ = self.test_server.invoke_unary_unary( method_descriptor, (), # initial metadata request.name.encode(), # Serialized request ).termination() self.assertEqual(code, grpc.StatusCode.OK) # Deserialize response to compare reply = MockHelloReply('') reply.message = response.decode() self.assertEqual(reply.message, 'Hello, World!') def test_say_hello_empty_name(self): method_descriptor = self.test_server.get_method_descriptor('/Greeter/SayHello') request = MockHelloRequest(name='') _, _, code, details = self.test_server.invoke_unary_unary( method_descriptor, (), request.name.encode(), ).termination() self.assertEqual(code, grpc.StatusCode.INVALID_ARGUMENT) self.assertEqual(details, 'Name cannot be empty!') if __name__ == '__main__': unittest.main()