grpc_testing_utils.py

import grpc
import grpc_testing
from google.protobuf import symbol_database


class _UnaryUnary:

    def __init__(self, server, method_descriptor):
        self._server = server
        self._method_descriptor = method_descriptor

    def __call__(self, request, timeout=None, metadata=None):
        rpc = self._server.invoke_unary_unary(
            method_descriptor=self._method_descriptor,
            invocation_metadata=metadata,
            request=request,
            timeout=timeout,
        )
        response, response_metadata, code, details = rpc.termination()
        if code is not grpc.StatusCode.OK:
            raise grpc.RpcError(code, details)
        else:
            return response


class ChannelFor:

    def __init__(self, handler):
        assert not isinstance(handler, type), 'Instance of a handler required'
        self._handler = handler

    def unary_unary(self, method: str, *args, **kwargs):
        sym_db = symbol_database.Default()

        service_name, _, method_name = method[1:].partition('/')
        service_descriptor = sym_db.pool.FindServiceByName(service_name)
        method_descriptor = service_descriptor.methods_by_name[method_name]

        server = grpc_testing.server_from_dictionary(
            {service_descriptor: self._handler},
            grpc_testing.strict_real_time(),
        )
        return _UnaryUnary(server, method_descriptor)


from helloworld import helloworld_pb2
from helloworld import helloworld_pb2_grpc
from _reference.server import Greeter


def test():
    channel = ChannelFor(Greeter())
    stub = helloworld_pb2_grpc.GreeterStub(channel)
    print(stub.SayHello(helloworld_pb2.HelloRequest(name='World!')))