diff options
Diffstat (limited to 'src/python/grpcio_testing/grpc_testing')
13 files changed, 115 insertions, 103 deletions
diff --git a/src/python/grpcio_testing/grpc_testing/__init__.py b/src/python/grpcio_testing/grpc_testing/__init__.py index 994274500c..e87d0ffc96 100644 --- a/src/python/grpcio_testing/grpc_testing/__init__.py +++ b/src/python/grpcio_testing/grpc_testing/__init__.py @@ -495,8 +495,8 @@ class Server(six.with_metaclass(abc.ABCMeta)): """A server with which to test a system that services RPCs.""" @abc.abstractmethod - def invoke_unary_unary( - self, method_descriptor, invocation_metadata, request, timeout): + def invoke_unary_unary(self, method_descriptor, invocation_metadata, + request, timeout): """Invokes an RPC to be serviced by the system under test. Args: @@ -513,8 +513,8 @@ class Server(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() @abc.abstractmethod - def invoke_unary_stream( - self, method_descriptor, invocation_metadata, request, timeout): + def invoke_unary_stream(self, method_descriptor, invocation_metadata, + request, timeout): """Invokes an RPC to be serviced by the system under test. Args: @@ -531,8 +531,8 @@ class Server(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() @abc.abstractmethod - def invoke_stream_unary( - self, method_descriptor, invocation_metadata, timeout): + def invoke_stream_unary(self, method_descriptor, invocation_metadata, + timeout): """Invokes an RPC to be serviced by the system under test. Args: @@ -548,8 +548,8 @@ class Server(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() @abc.abstractmethod - def invoke_stream_stream( - self, method_descriptor, invocation_metadata, timeout): + def invoke_stream_stream(self, method_descriptor, invocation_metadata, + timeout): """Invokes an RPC to be serviced by the system under test. Args: diff --git a/src/python/grpcio_testing/grpc_testing/_channel/__init__.py b/src/python/grpcio_testing/grpc_testing/_channel/__init__.py index 8011975d0a..7a64cda889 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/__init__.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/__init__.py @@ -20,4 +20,6 @@ from grpc_testing._channel import _channel_state # pylint: disable=unused-argument def testing_channel(descriptors, time): return _channel.TestingChannel(time, _channel_state.State()) + + # pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel.py index fbd064db88..b015b8d738 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_channel.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel.py @@ -32,20 +32,28 @@ class TestingChannel(grpc_testing.Channel): def unsubscribe(self, callback): raise NotImplementedError() - def unary_unary( - self, method, request_serializer=None, response_deserializer=None): + def unary_unary(self, + method, + request_serializer=None, + response_deserializer=None): return _multi_callable.UnaryUnary(method, self._state) - def unary_stream( - self, method, request_serializer=None, response_deserializer=None): + def unary_stream(self, + method, + request_serializer=None, + response_deserializer=None): return _multi_callable.UnaryStream(method, self._state) - def stream_unary( - self, method, request_serializer=None, response_deserializer=None): + def stream_unary(self, + method, + request_serializer=None, + response_deserializer=None): return _multi_callable.StreamUnary(method, self._state) - def stream_stream( - self, method, request_serializer=None, response_deserializer=None): + def stream_stream(self, + method, + request_serializer=None, + response_deserializer=None): return _multi_callable.StreamStream(method, self._state) def take_unary_unary(self, method_descriptor): @@ -59,4 +67,6 @@ class TestingChannel(grpc_testing.Channel): def take_stream_stream(self, method_descriptor): return _channel_rpc.stream_stream(self._state, method_descriptor) + + # pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py index 762b6a035b..54499b3b55 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel_rpc.py @@ -27,8 +27,8 @@ class _UnaryUnary(grpc_testing.UnaryUnaryChannelRpc): self._rpc_state.cancelled() def terminate(self, response, trailing_metadata, code, details): - self._rpc_state.terminate_with_response( - response, trailing_metadata, code, details) + self._rpc_state.terminate_with_response(response, trailing_metadata, + code, details) class _UnaryStream(grpc_testing.UnaryStreamChannelRpc): @@ -67,8 +67,8 @@ class _StreamUnary(grpc_testing.StreamUnaryChannelRpc): self._rpc_state.cancelled() def terminate(self, response, trailing_metadata, code, details): - self._rpc_state.terminate_with_response( - response, trailing_metadata, code, details) + self._rpc_state.terminate_with_response(response, trailing_metadata, + code, details) class _StreamStream(grpc_testing.StreamStreamChannelRpc): diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py b/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py index 569c41d79d..779d59e59a 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_channel_state.py @@ -25,11 +25,10 @@ class State(_common.ChannelHandler): self._condition = threading.Condition() self._rpc_states = collections.defaultdict(list) - def invoke_rpc( - self, method_full_rpc_name, invocation_metadata, requests, - requests_closed, timeout): - rpc_state = _rpc_state.State( - invocation_metadata, requests, requests_closed) + def invoke_rpc(self, method_full_rpc_name, invocation_metadata, requests, + requests_closed, timeout): + rpc_state = _rpc_state.State(invocation_metadata, requests, + requests_closed) with self._condition: self._rpc_states[method_full_rpc_name].append(rpc_state) self._condition.notify_all() diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py b/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py index fe69257f5b..32b2f361d3 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_multi_callable.py @@ -16,6 +16,7 @@ import grpc from grpc_testing import _common from grpc_testing._channel import _invocation + # All per-call credentials parameters are unused by this test infrastructure. # pylint: disable=unused-argument class UnaryUnary(grpc.UnaryUnaryMultiCallable): @@ -26,20 +27,20 @@ class UnaryUnary(grpc.UnaryUnaryMultiCallable): def __call__(self, request, timeout=None, metadata=None, credentials=None): rpc_handler = self._channel_handler.invoke_rpc( - self._method_full_rpc_name, _common.fuss_with_metadata(metadata), - [request], True, timeout) + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [request], True, timeout) return _invocation.blocking_unary_response(rpc_handler) def with_call(self, request, timeout=None, metadata=None, credentials=None): rpc_handler = self._channel_handler.invoke_rpc( - self._method_full_rpc_name, _common.fuss_with_metadata(metadata), - [request], True, timeout) + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [request], True, timeout) return _invocation.blocking_unary_response_with_call(rpc_handler) def future(self, request, timeout=None, metadata=None, credentials=None): rpc_handler = self._channel_handler.invoke_rpc( - self._method_full_rpc_name, _common.fuss_with_metadata(metadata), - [request], True, timeout) + self._method_full_rpc_name, + _common.fuss_with_metadata(metadata), [request], True, timeout) return _invocation.future_call(rpc_handler) @@ -112,4 +113,6 @@ class StreamStream(grpc.StreamStreamMultiCallable): _common.fuss_with_metadata(metadata), [], False, timeout) _invocation.consume_requests(request_iterator, rpc_handler) return _invocation.ResponseIteratorCall(rpc_handler) + + # pylint: enable=unused-argument diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py b/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py index e1fa49a2a8..ee0233002d 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_rpc_state.py @@ -63,23 +63,22 @@ class State(_common.ChannelRpcHandler): if self._code is grpc.StatusCode.OK: if self._responses: response = self._responses.pop(0) - return _common.ChannelRpcRead( - response, None, None, None) + return _common.ChannelRpcRead(response, None, None, + None) else: return _common.ChannelRpcRead( - None, self._trailing_metadata, - grpc.StatusCode.OK, self._details) + None, self._trailing_metadata, grpc.StatusCode.OK, + self._details) elif self._code is None: if self._responses: response = self._responses.pop(0) - return _common.ChannelRpcRead( - response, None, None, None) + return _common.ChannelRpcRead(response, None, None, + None) else: self._condition.wait() else: - return _common.ChannelRpcRead( - None, self._trailing_metadata, self._code, - self._details) + return _common.ChannelRpcRead(None, self._trailing_metadata, + self._code, self._details) def termination(self): with self._condition: @@ -150,8 +149,8 @@ class State(_common.ChannelRpcHandler): self._responses.append(response) self._condition.notify_all() - def terminate_with_response( - self, response, trailing_metadata, code, details): + def terminate_with_response(self, response, trailing_metadata, code, + details): with self._condition: if self._initial_metadata is None: self._initial_metadata = _common.FUSSED_EMPTY_METADATA diff --git a/src/python/grpcio_testing/grpc_testing/_common.py b/src/python/grpcio_testing/grpc_testing/_common.py index 1517434ca7..05327b0ac2 100644 --- a/src/python/grpcio_testing/grpc_testing/_common.py +++ b/src/python/grpcio_testing/grpc_testing/_common.py @@ -21,11 +21,9 @@ import six def _fuss(tuplified_metadata): return tuplified_metadata + ( - ( - 'grpc.metadata_added_by_runtime', - 'gRPC is allowed to add metadata in transmission and does so.', - ), - ) + ('grpc.metadata_added_by_runtime', + 'gRPC is allowed to add metadata in transmission and does so.',),) + FUSSED_EMPTY_METADATA = _fuss(()) @@ -41,8 +39,8 @@ def rpc_names(service_descriptors): rpc_names_to_descriptors = {} for service_descriptor in service_descriptors: for method_descriptor in service_descriptor.methods_by_name.values(): - rpc_name = '/{}/{}'.format( - service_descriptor.full_name, method_descriptor.name) + rpc_name = '/{}/{}'.format(service_descriptor.full_name, + method_descriptor.name) rpc_names_to_descriptors[rpc_name] = method_descriptor return rpc_names_to_descriptors @@ -96,9 +94,8 @@ class ChannelRpcHandler(six.with_metaclass(abc.ABCMeta)): class ChannelHandler(six.with_metaclass(abc.ABCMeta)): @abc.abstractmethod - def invoke_rpc( - self, method_full_rpc_name, invocation_metadata, requests, - requests_closed, timeout): + def invoke_rpc(self, method_full_rpc_name, invocation_metadata, requests, + requests_closed, timeout): raise NotImplementedError() @@ -138,23 +135,21 @@ class ServerRpcHandler(six.with_metaclass(abc.ABCMeta)): class Serverish(six.with_metaclass(abc.ABCMeta)): @abc.abstractmethod - def invoke_unary_unary( - self, method_descriptor, handler, invocation_metadata, request, - deadline): + def invoke_unary_unary(self, method_descriptor, handler, + invocation_metadata, request, deadline): raise NotImplementedError() @abc.abstractmethod - def invoke_unary_stream( - self, method_descriptor, handler, invocation_metadata, request, - deadline): + def invoke_unary_stream(self, method_descriptor, handler, + invocation_metadata, request, deadline): raise NotImplementedError() @abc.abstractmethod - def invoke_stream_unary( - self, method_descriptor, handler, invocation_metadata, deadline): + def invoke_stream_unary(self, method_descriptor, handler, + invocation_metadata, deadline): raise NotImplementedError() @abc.abstractmethod - def invoke_stream_stream( - self, method_descriptor, handler, invocation_metadata, deadline): + def invoke_stream_stream(self, method_descriptor, handler, + invocation_metadata, deadline): raise NotImplementedError() diff --git a/src/python/grpcio_testing/grpc_testing/_server/__init__.py b/src/python/grpcio_testing/grpc_testing/_server/__init__.py index 759512949a..5f035a91ca 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/__init__.py +++ b/src/python/grpcio_testing/grpc_testing/_server/__init__.py @@ -16,5 +16,5 @@ from grpc_testing._server import _server def server_from_dictionary(descriptors_to_servicers, time): - return _server.server_from_descriptor_to_servicers( - descriptors_to_servicers, time) + return _server.server_from_descriptor_to_servicers(descriptors_to_servicers, + time) diff --git a/src/python/grpcio_testing/grpc_testing/_server/_handler.py b/src/python/grpcio_testing/grpc_testing/_server/_handler.py index b47e04c718..5e4730e087 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/_handler.py +++ b/src/python/grpcio_testing/grpc_testing/_server/_handler.py @@ -170,10 +170,8 @@ class _Handler(Handler): if self._unary_response is None: if self._responses: self._unary_response = self._responses.pop(0) - return ( - self._unary_response, self._trailing_metadata, - self._code, self._details,) - + return (self._unary_response, self._trailing_metadata, + self._code, self._details,) def stream_response_termination(self): with self._condition: diff --git a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py index f81876f4b2..2060e8daff 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py +++ b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py @@ -80,9 +80,8 @@ class Rpc(object): def application_cancel(self): with self._condition: - self._abort( - grpc.StatusCode.CANCELLED, - 'Cancelled by server-side application!') + self._abort(grpc.StatusCode.CANCELLED, + 'Cancelled by server-side application!') def application_exception_abort(self, exception): with self._condition: diff --git a/src/python/grpcio_testing/grpc_testing/_server/_server.py b/src/python/grpcio_testing/grpc_testing/_server/_server.py index 66bcfc13c0..3e358e50a9 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/_server.py +++ b/src/python/grpcio_testing/grpc_testing/_server/_server.py @@ -29,28 +29,34 @@ def _implementation(descriptors_to_servicers, method_descriptor): def _unary_unary_service(request): + def service(implementation, rpc, servicer_context): - _service.unary_unary( - implementation, rpc, request, servicer_context) + _service.unary_unary(implementation, rpc, request, servicer_context) + return service def _unary_stream_service(request): + def service(implementation, rpc, servicer_context): - _service.unary_stream( - implementation, rpc, request, servicer_context) + _service.unary_stream(implementation, rpc, request, servicer_context) + return service def _stream_unary_service(handler): + def service(implementation, rpc, servicer_context): _service.stream_unary(implementation, rpc, handler, servicer_context) + return service def _stream_stream_service(handler): + def service(implementation, rpc, servicer_context): _service.stream_stream(implementation, rpc, handler, servicer_context) + return service @@ -60,11 +66,10 @@ class _Serverish(_common.Serverish): self._descriptors_to_servicers = descriptors_to_servicers self._time = time - def _invoke( - self, service_behavior, method_descriptor, handler, - invocation_metadata, deadline): - implementation = _implementation( - self._descriptors_to_servicers, method_descriptor) + def _invoke(self, service_behavior, method_descriptor, handler, + invocation_metadata, deadline): + implementation = _implementation(self._descriptors_to_servicers, + method_descriptor) rpc = _rpc.Rpc(handler, invocation_metadata) if handler.add_termination_callback(rpc.extrinsic_abort): servicer_context = _servicer_context.ServicerContext( @@ -74,28 +79,26 @@ class _Serverish(_common.Serverish): args=(implementation, rpc, servicer_context,)) service_thread.start() - def invoke_unary_unary( - self, method_descriptor, handler, invocation_metadata, request, - deadline): + def invoke_unary_unary(self, method_descriptor, handler, + invocation_metadata, request, deadline): self._invoke( _unary_unary_service(request), method_descriptor, handler, invocation_metadata, deadline) - def invoke_unary_stream( - self, method_descriptor, handler, invocation_metadata, request, - deadline): + def invoke_unary_stream(self, method_descriptor, handler, + invocation_metadata, request, deadline): self._invoke( _unary_stream_service(request), method_descriptor, handler, invocation_metadata, deadline) - def invoke_stream_unary( - self, method_descriptor, handler, invocation_metadata, deadline): + def invoke_stream_unary(self, method_descriptor, handler, + invocation_metadata, deadline): self._invoke( _stream_unary_service(handler), method_descriptor, handler, invocation_metadata, deadline) - def invoke_stream_stream( - self, method_descriptor, handler, invocation_metadata, deadline): + def invoke_stream_stream(self, method_descriptor, handler, + invocation_metadata, deadline): self._invoke( _stream_stream_service(handler), method_descriptor, handler, invocation_metadata, deadline) @@ -106,7 +109,8 @@ def _deadline_and_handler(requests_closed, time, timeout): return None, _handler.handler_without_deadline(requests_closed) else: deadline = time.time() + timeout - handler = _handler.handler_with_deadline(requests_closed, time, deadline) + handler = _handler.handler_with_deadline(requests_closed, time, + deadline) return deadline, handler @@ -116,32 +120,32 @@ class _Server(grpc_testing.Server): self._serverish = serverish self._time = time - def invoke_unary_unary( - self, method_descriptor, invocation_metadata, request, timeout): + def invoke_unary_unary(self, method_descriptor, invocation_metadata, + request, timeout): deadline, handler = _deadline_and_handler(True, self._time, timeout) self._serverish.invoke_unary_unary( method_descriptor, handler, invocation_metadata, request, deadline) return _server_rpc.UnaryUnaryServerRpc(handler) - def invoke_unary_stream( - self, method_descriptor, invocation_metadata, request, timeout): + def invoke_unary_stream(self, method_descriptor, invocation_metadata, + request, timeout): deadline, handler = _deadline_and_handler(True, self._time, timeout) self._serverish.invoke_unary_stream( method_descriptor, handler, invocation_metadata, request, deadline) return _server_rpc.UnaryStreamServerRpc(handler) - def invoke_stream_unary( - self, method_descriptor, invocation_metadata, timeout): + def invoke_stream_unary(self, method_descriptor, invocation_metadata, + timeout): deadline, handler = _deadline_and_handler(False, self._time, timeout) - self._serverish.invoke_stream_unary( - method_descriptor, handler, invocation_metadata, deadline) + self._serverish.invoke_stream_unary(method_descriptor, handler, + invocation_metadata, deadline) return _server_rpc.StreamUnaryServerRpc(handler) - def invoke_stream_stream( - self, method_descriptor, invocation_metadata, timeout): + def invoke_stream_stream(self, method_descriptor, invocation_metadata, + timeout): deadline, handler = _deadline_and_handler(False, self._time, timeout) - self._serverish.invoke_stream_stream( - method_descriptor, handler, invocation_metadata, deadline) + self._serverish.invoke_stream_stream(method_descriptor, handler, + invocation_metadata, deadline) return _server_rpc.StreamStreamServerRpc(handler) diff --git a/src/python/grpcio_testing/grpc_testing/_server/_servicer_context.py b/src/python/grpcio_testing/grpc_testing/_server/_servicer_context.py index 496689ded0..90eeb130d3 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/_servicer_context.py +++ b/src/python/grpcio_testing/grpc_testing/_server/_servicer_context.py @@ -67,6 +67,9 @@ class ServicerContext(grpc.ServicerContext): self._rpc.set_trailing_metadata( _common.fuss_with_metadata(trailing_metadata)) + def abort(self, code, details): + raise NotImplementedError() + def set_code(self, code): self._rpc.set_code(code) |