diff options
Diffstat (limited to 'src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py')
-rw-r--r-- | src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py | 187 |
1 files changed, 93 insertions, 94 deletions
diff --git a/src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py b/src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py index ac487bed4f..4e144a3635 100644 --- a/src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py +++ b/src/python/grpcio_tests/tests/unit/framework/interfaces/face/_invocation.py @@ -26,7 +26,6 @@ # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - """Coverage across the Face layer's generic-to-dynamic range for invocation.""" import abc @@ -65,149 +64,149 @@ _CARDINALITY_TO_MULTI_CALLABLE_ATTRIBUTE = { class Invoker(six.with_metaclass(abc.ABCMeta)): - """A type used to invoke test RPCs.""" + """A type used to invoke test RPCs.""" - @abc.abstractmethod - def blocking(self, group, name): - """Invokes an RPC with blocking control flow.""" - raise NotImplementedError() + @abc.abstractmethod + def blocking(self, group, name): + """Invokes an RPC with blocking control flow.""" + raise NotImplementedError() - @abc.abstractmethod - def future(self, group, name): - """Invokes an RPC with future control flow.""" - raise NotImplementedError() + @abc.abstractmethod + def future(self, group, name): + """Invokes an RPC with future control flow.""" + raise NotImplementedError() - @abc.abstractmethod - def event(self, group, name): - """Invokes an RPC with event control flow.""" - raise NotImplementedError() + @abc.abstractmethod + def event(self, group, name): + """Invokes an RPC with event control flow.""" + raise NotImplementedError() class InvokerConstructor(six.with_metaclass(abc.ABCMeta)): - """A type used to create Invokers.""" + """A type used to create Invokers.""" - @abc.abstractmethod - def name(self): - """Specifies the name of the Invoker constructed by this object.""" - raise NotImplementedError() + @abc.abstractmethod + def name(self): + """Specifies the name of the Invoker constructed by this object.""" + raise NotImplementedError() - @abc.abstractmethod - def construct_invoker(self, generic_stub, dynamic_stubs, methods): - """Constructs an Invoker for the given stubs and methods.""" - raise NotImplementedError() + @abc.abstractmethod + def construct_invoker(self, generic_stub, dynamic_stubs, methods): + """Constructs an Invoker for the given stubs and methods.""" + raise NotImplementedError() class _GenericInvoker(Invoker): - def __init__(self, generic_stub, methods): - self._stub = generic_stub - self._methods = methods + def __init__(self, generic_stub, methods): + self._stub = generic_stub + self._methods = methods - def _behavior(self, group, name, cardinality_to_generic_method): - method_cardinality = self._methods[group, name].cardinality() - behavior = getattr( - self._stub, cardinality_to_generic_method[method_cardinality]) - return lambda *args, **kwargs: behavior(group, name, *args, **kwargs) + def _behavior(self, group, name, cardinality_to_generic_method): + method_cardinality = self._methods[group, name].cardinality() + behavior = getattr(self._stub, + cardinality_to_generic_method[method_cardinality]) + return lambda *args, **kwargs: behavior(group, name, *args, **kwargs) - def blocking(self, group, name): - return self._behavior( - group, name, _CARDINALITY_TO_GENERIC_BLOCKING_BEHAVIOR) + def blocking(self, group, name): + return self._behavior(group, name, + _CARDINALITY_TO_GENERIC_BLOCKING_BEHAVIOR) - def future(self, group, name): - return self._behavior(group, name, _CARDINALITY_TO_GENERIC_FUTURE_BEHAVIOR) + def future(self, group, name): + return self._behavior(group, name, + _CARDINALITY_TO_GENERIC_FUTURE_BEHAVIOR) - def event(self, group, name): - return self._behavior(group, name, _CARDINALITY_TO_GENERIC_EVENT_BEHAVIOR) + def event(self, group, name): + return self._behavior(group, name, + _CARDINALITY_TO_GENERIC_EVENT_BEHAVIOR) class _GenericInvokerConstructor(InvokerConstructor): - def name(self): - return 'GenericInvoker' + def name(self): + return 'GenericInvoker' - def construct_invoker(self, generic_stub, dynamic_stub, methods): - return _GenericInvoker(generic_stub, methods) + def construct_invoker(self, generic_stub, dynamic_stub, methods): + return _GenericInvoker(generic_stub, methods) class _MultiCallableInvoker(Invoker): - def __init__(self, generic_stub, methods): - self._stub = generic_stub - self._methods = methods + def __init__(self, generic_stub, methods): + self._stub = generic_stub + self._methods = methods - def _multi_callable(self, group, name): - method_cardinality = self._methods[group, name].cardinality() - behavior = getattr( - self._stub, - _CARDINALITY_TO_MULTI_CALLABLE_ATTRIBUTE[method_cardinality]) - return behavior(group, name) + def _multi_callable(self, group, name): + method_cardinality = self._methods[group, name].cardinality() + behavior = getattr( + self._stub, + _CARDINALITY_TO_MULTI_CALLABLE_ATTRIBUTE[method_cardinality]) + return behavior(group, name) - def blocking(self, group, name): - return self._multi_callable(group, name) + def blocking(self, group, name): + return self._multi_callable(group, name) - def future(self, group, name): - method_cardinality = self._methods[group, name].cardinality() - behavior = getattr( - self._stub, - _CARDINALITY_TO_MULTI_CALLABLE_ATTRIBUTE[method_cardinality]) - if method_cardinality in ( - cardinality.Cardinality.UNARY_UNARY, - cardinality.Cardinality.STREAM_UNARY): - return behavior(group, name).future - else: - return behavior(group, name) + def future(self, group, name): + method_cardinality = self._methods[group, name].cardinality() + behavior = getattr( + self._stub, + _CARDINALITY_TO_MULTI_CALLABLE_ATTRIBUTE[method_cardinality]) + if method_cardinality in (cardinality.Cardinality.UNARY_UNARY, + cardinality.Cardinality.STREAM_UNARY): + return behavior(group, name).future + else: + return behavior(group, name) - def event(self, group, name): - return self._multi_callable(group, name).event + def event(self, group, name): + return self._multi_callable(group, name).event class _MultiCallableInvokerConstructor(InvokerConstructor): - def name(self): - return 'MultiCallableInvoker' + def name(self): + return 'MultiCallableInvoker' - def construct_invoker(self, generic_stub, dynamic_stub, methods): - return _MultiCallableInvoker(generic_stub, methods) + def construct_invoker(self, generic_stub, dynamic_stub, methods): + return _MultiCallableInvoker(generic_stub, methods) class _DynamicInvoker(Invoker): - def __init__(self, dynamic_stubs, methods): - self._stubs = dynamic_stubs - self._methods = methods + def __init__(self, dynamic_stubs, methods): + self._stubs = dynamic_stubs + self._methods = methods - def blocking(self, group, name): - return getattr(self._stubs[group], name) + def blocking(self, group, name): + return getattr(self._stubs[group], name) - def future(self, group, name): - if self._methods[group, name].cardinality() in ( - cardinality.Cardinality.UNARY_UNARY, - cardinality.Cardinality.STREAM_UNARY): - return getattr(self._stubs[group], name).future - else: - return getattr(self._stubs[group], name) + def future(self, group, name): + if self._methods[group, name].cardinality() in ( + cardinality.Cardinality.UNARY_UNARY, + cardinality.Cardinality.STREAM_UNARY): + return getattr(self._stubs[group], name).future + else: + return getattr(self._stubs[group], name) - def event(self, group, name): - return getattr(self._stubs[group], name).event + def event(self, group, name): + return getattr(self._stubs[group], name).event class _DynamicInvokerConstructor(InvokerConstructor): - def name(self): - return 'DynamicInvoker' + def name(self): + return 'DynamicInvoker' - def construct_invoker(self, generic_stub, dynamic_stubs, methods): - return _DynamicInvoker(dynamic_stubs, methods) + def construct_invoker(self, generic_stub, dynamic_stubs, methods): + return _DynamicInvoker(dynamic_stubs, methods) def invoker_constructors(): - """Creates a sequence of InvokerConstructors to use in tests of RPCs. + """Creates a sequence of InvokerConstructors to use in tests of RPCs. Returns: A sequence of InvokerConstructors. """ - return ( - _GenericInvokerConstructor(), - _MultiCallableInvokerConstructor(), - _DynamicInvokerConstructor(), - ) + return ( + _GenericInvokerConstructor(), + _MultiCallableInvokerConstructor(), + _DynamicInvokerConstructor(),) |