There are some questions here regarding grpc testing in python, but none of them involve streaming responses. Unfortunately the grpc maintainers are not documenting the testing API besides a generic auto generated page.
Let's say there is a grpc interface like this:
service EchoService {
rpc doStuff(EchoRequest) returns (stream EchoResponse) {
}
}
message EchoRequest {
string name = 1;
}
message EchoResponse {
string name = 1;
}
With a server implementation like this:
class MyServer(myservice_pb2_grpc.EchoServiceServicer):
async def doStuff(
self, request: myservice_pb2.EchoRequest, context: grpc.aio.ServicerContext
):
yield myservice_pb2.EchoResponse(name=f"test response {request.name}")
How can you write a test for this server implementation?
I've tried several approaches not knowing which one could possibly work. The most promising one to me seems to be to use pytest-grpc
using code like this:
@pytest.fixture(scope="module")
def grpc_servicer():
return server.MyServer()
def test_do_stuff(grpc_stub):
request = myservice_pb2.EchoRequest(name="foo")
response = grpc_stub.doStuff(request)
breakpoint()
However here this error is returned, which seems to suggest that some grpc internal is not capable of handling async implementations:
Exception iterating responses: \'async_generator\' object is not an iterator"
<_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.UNKNOWN
details = "Exception iterating responses: 'async_generator' object is not an iterator"
debug_error_string = "UNKNOWN:Error received from peer {grpc_message:"Exception iterating responses: \'async_generator\' object is not an iterator", grpc_status:2, created_time:"2023-12-12T09:58:11.624820951+00:00"}"