vmagamedov/grpclib

How to make a unittest with data stream?

Closed this issue · 3 comments

ozangungor12 commented

I am trying to implement unittests for a service that expects a stream of data.
How can I create a Stream object to pass to my channel?

service MyService {
  rpc MyMethod(stream int) returns(int) {}
}

When I use the example from the documentation,

greeter = Greeter()

async with ChannelFor([greeter]) as channel:
    stub = GreeterStub(channel)

as

    servicer = MyServicer()

    async with ChannelFor([servicer]) as channel:
        stub = MyService(channel)

I get the following error.

'Channel' object has no attribute 'stream_unary'

Thanks for the help.

vmagamedov commented

You probably importing MyService client stub from grpcio stubs (*_pb2_grpc.py) and not from grpclib stubs (*_grpc.py).

ozangungor12 commented

Thanks @vmagamedov, that actually solved the problem. But, I am not sure I understood the point here, what's the difference between those two stubs, coming from *_pb2_grpc.py and *_grpc.py?

vmagamedov commented

grpclib and grpcio are completely different implementations, grpcio started as synchronous-only implementation (now with async support) and grpclib started as asynchronous-only implementation, also grpclib stubs are completely annotated with types. That's why we can't use the same stubs for both libraries, they also rely on some internals, but maybe this will change in the future, I already have a branch with grpcio-compatible Channel implementation.

  • 👍1