I have created python grpc client to communicate with java grpc server and have implemented one grpc api from client side.
self.channel = grpc.insecure_channel(
self.host, options=[("grpc.lb_policy_name", "round_robin")]
)
self.blocking_stub = contacts_pb2_grpc.ExampleServiceStub(self.channel
and invoked the grpc api by following call
def create_call(self, request):
return self.blocking_stub.createCall(
request, metadata=add_meta_data(self), timeout=self.deadline_seconds
)
my question here is I want to mock the create_call and need to get the response back so I have created python service implementation.
I want to make sure the service code is called whenever I made create_call in unitest .In java we have something like this.
String serverName = InProcessServerBuilder.generateName();
grpcCleanup.register(
InProcessServerBuilder.forName(serverName)
.directExecutor()
.addService(ServiceImplTestHelper.serviceImpl)
.build()
.start());
channel =
grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build());
Can someone suggest how we can do this in python ?