diff options
Diffstat (limited to 'src/python')
42 files changed, 841 insertions, 188 deletions
diff --git a/src/python/grpcio/commands.py b/src/python/grpcio/commands.py index ea3b6f3391..701c6af017 100644 --- a/src/python/grpcio/commands.py +++ b/src/python/grpcio/commands.py @@ -62,6 +62,7 @@ napoleon_numpy_docstring = True napoleon_include_special_with_doc = True html_theme = 'sphinx_rtd_theme' +copyright = "2016, The gRPC Authors" """ API_GLOSSARY = """ diff --git a/src/python/grpcio/grpc/__init__.py b/src/python/grpcio/grpc/__init__.py index 526bd9e14f..e3c10156d0 100644 --- a/src/python/grpcio/grpc/__init__.py +++ b/src/python/grpcio/grpc/__init__.py @@ -31,6 +31,7 @@ import abc import enum +import sys import six @@ -767,8 +768,8 @@ class ServicerContext(six.with_metaclass(abc.ABCMeta, RpcContext)): gRPC runtime to determine the status code of the RPC. Args: - code: The integer status code of the RPC to be transmitted to the - invocation side of the RPC. + code: A StatusCode value to be transmitted to the invocation side of the + RPC as the status code of the RPC. """ raise NotImplementedError() @@ -780,8 +781,8 @@ class ServicerContext(six.with_metaclass(abc.ABCMeta, RpcContext)): details to transmit. Args: - details: The details string of the RPC to be transmitted to - the invocation side of the RPC. + details: A string to be transmitted to the invocation side of the RPC as + the status details of the RPC. """ raise NotImplementedError() @@ -849,6 +850,26 @@ class GenericRpcHandler(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() +class ServiceRpcHandler(six.with_metaclass(abc.ABCMeta, GenericRpcHandler)): + """An implementation of RPC methods belonging to a service. + + A service handles RPC methods with structured names of the form + '/Service.Name/Service.MethodX', where 'Service.Name' is the value + returned by service_name(), and 'Service.MethodX' is the service method + name. A service can have multiple service methods names, but only a single + service name. + """ + + @abc.abstractmethod + def service_name(self): + """Returns this services name. + + Returns: + The service name. + """ + raise NotImplementedError() + + ############################# Server Interface ############################### @@ -927,10 +948,16 @@ class Server(six.with_metaclass(abc.ABCMeta)): passed in a previous call will not have the effect of stopping the server later. + This method does not block for any significant length of time. If None is + passed as the grace value, existing RPCs are immediately aborted and this + method blocks until this Server is completely stopped. + Args: - grace: A duration of time in seconds to allow existing RPCs to complete - before being aborted by this Server's stopping. If None, this method - will block until the server is completely stopped. + grace: A duration of time in seconds or None. If a duration of time in + seconds, the time to allow existing RPCs to complete before being + aborted by this Server's stopping. If None, all RPCs will be aborted + immediately and this method will block until this Server is completely + stopped. Returns: A threading.Event that will be set when this Server has completely @@ -1274,6 +1301,7 @@ __all__ = ( 'RpcMethodHandler', 'HandlerCallDetails', 'GenericRpcHandler', + 'ServiceRpcHandler', 'Server', 'unary_unary_rpc_method_handler', 'unary_stream_rpc_method_handler', @@ -1291,3 +1319,24 @@ __all__ = ( 'secure_channel', 'server', ) + + +############################### Extension Shims ################################ + + +# Here to maintain backwards compatibility; avoid using these in new code! +try: + import grpc_tools + sys.modules.update({'grpc.tools': grpc_tools}) +except ImportError: + pass +try: + import grpc_health + sys.modules.update({'grpc.health': grpc_health}) +except ImportError: + pass +try: + import grpc_reflection + sys.modules.update({'grpc.reflection': grpc_reflection}) +except ImportError: + pass diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py index 53a26727ab..e8c6a99cb1 100644 --- a/src/python/grpcio/grpc/_channel.py +++ b/src/python/grpcio/grpc/_channel.py @@ -32,12 +32,13 @@ import sys import threading import time +import logging import grpc from grpc import _common from grpc import _grpcio_metadata -from grpc.framework.foundation import callable_util from grpc._cython import cygrpc +from grpc.framework.foundation import callable_util _USER_AGENT = 'Python-gRPC-{}'.format(_grpcio_metadata.__version__) @@ -99,6 +100,22 @@ def _wait_once_until(condition, until): else: condition.wait(timeout=remaining) +_INTERNAL_CALL_ERROR_MESSAGE_FORMAT = ( + 'Internal gRPC call error %d. ' + + 'Please report to https://github.com/grpc/grpc/issues') + +def _check_call_error(call_error, metadata): + if call_error == cygrpc.CallError.invalid_metadata: + raise ValueError('metadata was invalid: %s' % metadata) + elif call_error != cygrpc.CallError.ok: + raise ValueError(_INTERNAL_CALL_ERROR_MESSAGE_FORMAT % call_error) + +def _call_error_set_RPCstate(state, call_error, metadata): + if call_error == cygrpc.CallError.invalid_metadata: + _abort(state, grpc.StatusCode.INTERNAL, 'metadata was invalid: %s' % metadata) + else: + _abort(state, grpc.StatusCode.INTERNAL, + _INTERNAL_CALL_ERROR_MESSAGE_FORMAT % call_error) class _RPCState(object): @@ -181,7 +198,16 @@ def _consume_request_iterator( event_handler = _event_handler(state, call, None) def consume_request_iterator(): - for request in request_iterator: + while True: + try: + request = next(request_iterator) + except StopIteration: + break + except Exception as e: + logging.exception("Exception iterating requests!") + call.cancel() + _abort(state, grpc.StatusCode.UNKNOWN, "Exception iterating requests!") + return serialized_request = _common.serialize(request, request_serializer) with state.condition: if state.code is None and not state.cancelled: @@ -358,7 +384,7 @@ class _Rendezvous(grpc.RpcError, grpc.Future, grpc.Call): if self._state.callbacks is None: return False else: - self._state.callbacks.append(lambda: callback()) + self._state.callbacks.append(callback) return True def initial_metadata(self): @@ -435,10 +461,10 @@ def _end_unary_response_blocking(state, with_call, deadline): class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): def __init__( - self, channel, create_managed_call, method, request_serializer, + self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel - self._create_managed_call = create_managed_call + self._managed_call = managed_call self._method = method self._request_serializer = request_serializer self._response_deserializer = response_deserializer @@ -472,7 +498,8 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): None, 0, completion_queue, self._method, None, deadline_timespec) if credentials is not None: call.set_credentials(credentials._credentials) - call.start_client_batch(cygrpc.Operations(operations), None) + call_error = call.start_client_batch(cygrpc.Operations(operations), None) + _check_call_error(call_error, metadata) _handle_event(completion_queue.poll(), state, self._response_deserializer) return state, deadline @@ -490,23 +517,28 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): if rendezvous: return rendezvous else: - call = self._create_managed_call( + call, drive_call = self._managed_call( None, 0, self._method, None, deadline_timespec) if credentials is not None: call.set_credentials(credentials._credentials) event_handler = _event_handler(state, call, self._response_deserializer) with state.condition: - call.start_client_batch(cygrpc.Operations(operations), event_handler) + call_error = call.start_client_batch(cygrpc.Operations(operations), + event_handler) + if call_error != cygrpc.CallError.ok: + _call_error_set_RPCstate(state, call_error, metadata) + return _Rendezvous(state, None, None, deadline) + drive_call() return _Rendezvous(state, call, self._response_deserializer, deadline) class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): def __init__( - self, channel, create_managed_call, method, request_serializer, + self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel - self._create_managed_call = create_managed_call + self._managed_call = managed_call self._method = method self._request_serializer = request_serializer self._response_deserializer = response_deserializer @@ -518,7 +550,7 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): raise rendezvous else: state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) - call = self._create_managed_call( + call, drive_call = self._managed_call( None, 0, self._method, None, deadline_timespec) if credentials is not None: call.set_credentials(credentials._credentials) @@ -535,17 +567,22 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): cygrpc.operation_send_close_from_client(_EMPTY_FLAGS), cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), ) - call.start_client_batch(cygrpc.Operations(operations), event_handler) + call_error = call.start_client_batch(cygrpc.Operations(operations), + event_handler) + if call_error != cygrpc.CallError.ok: + _call_error_set_RPCstate(state, call_error, metadata) + return _Rendezvous(state, None, None, deadline) + drive_call() return _Rendezvous(state, call, self._response_deserializer, deadline) class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): def __init__( - self, channel, create_managed_call, method, request_serializer, + self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel - self._create_managed_call = create_managed_call + self._managed_call = managed_call self._method = method self._request_serializer = request_serializer self._response_deserializer = response_deserializer @@ -569,7 +606,8 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): cygrpc.operation_receive_message(_EMPTY_FLAGS), cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), ) - call.start_client_batch(cygrpc.Operations(operations), None) + call_error = call.start_client_batch(cygrpc.Operations(operations), None) + _check_call_error(call_error, metadata) _consume_request_iterator( request_iterator, state, call, self._request_serializer) while True: @@ -597,7 +635,7 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): self, request_iterator, timeout=None, metadata=None, credentials=None): deadline, deadline_timespec = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) - call = self._create_managed_call( + call, drive_call = self._managed_call( None, 0, self._method, None, deadline_timespec) if credentials is not None: call.set_credentials(credentials._credentials) @@ -613,7 +651,12 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): cygrpc.operation_receive_message(_EMPTY_FLAGS), cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), ) - call.start_client_batch(cygrpc.Operations(operations), event_handler) + call_error = call.start_client_batch(cygrpc.Operations(operations), + event_handler) + if call_error != cygrpc.CallError.ok: + _call_error_set_RPCstate(state, call_error, metadata) + return _Rendezvous(state, None, None, deadline) + drive_call() _consume_request_iterator( request_iterator, state, call, self._request_serializer) return _Rendezvous(state, call, self._response_deserializer, deadline) @@ -622,10 +665,10 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): def __init__( - self, channel, create_managed_call, method, request_serializer, + self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel - self._create_managed_call = create_managed_call + self._managed_call = managed_call self._method = method self._request_serializer = request_serializer self._response_deserializer = response_deserializer @@ -634,7 +677,7 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): self, request_iterator, timeout=None, metadata=None, credentials=None): deadline, deadline_timespec = _deadline(timeout) state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None) - call = self._create_managed_call( + call, drive_call = self._managed_call( None, 0, self._method, None, deadline_timespec) if credentials is not None: call.set_credentials(credentials._credentials) @@ -649,7 +692,12 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): _common.cygrpc_metadata(metadata), _EMPTY_FLAGS), cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), ) - call.start_client_batch(cygrpc.Operations(operations), event_handler) + call_error = call.start_client_batch(cygrpc.Operations(operations), + event_handler) + if call_error != cygrpc.CallError.ok: + _call_error_set_RPCstate(state, call_error, metadata) + return _Rendezvous(state, None, None, deadline) + drive_call() _consume_request_iterator( request_iterator, state, call, self._request_serializer) return _Rendezvous(state, call, self._response_deserializer, deadline) @@ -687,16 +735,13 @@ def _run_channel_spin_thread(state): channel_spin_thread.start() -def _create_channel_managed_call(state): - def create_channel_managed_call(parent, flags, method, host, deadline): - """Creates a managed cygrpc.Call. +def _channel_managed_call_management(state): + def create(parent, flags, method, host, deadline): + """Creates a managed cygrpc.Call and a function to call to drive it. - Callers of this function must conduct at least one operation on the returned - call. The tags associated with operations conducted on the returned call - must be no-argument callables that return None to indicate that this channel - should continue polling for events associated with the call and return the - call itself to indicate that no more events associated with the call will be - generated. + If operations are successfully added to the returned cygrpc.Call, the + returned function must be called. If operations are not successfully added + to the returned cygrpc.Call, the returned function must not be called. Args: parent: A cygrpc.Call to be used as the parent of the created call. @@ -706,18 +751,22 @@ def _create_channel_managed_call(state): deadline: A cygrpc.Timespec to be the deadline of the created call. Returns: - A cygrpc.Call with which to conduct an RPC. + A cygrpc.Call with which to conduct an RPC and a function to call if + operations are successfully started on the call. """ - with state.lock: - call = state.channel.create_call( - parent, flags, state.completion_queue, method, host, deadline) - if state.managed_calls is None: - state.managed_calls = set((call,)) - _run_channel_spin_thread(state) - else: - state.managed_calls.add(call) - return call - return create_channel_managed_call + call = state.channel.create_call( + parent, flags, state.completion_queue, method, host, deadline) + + def drive(): + with state.lock: + if state.managed_calls is None: + state.managed_calls = set((call,)) + _run_channel_spin_thread(state) + else: + state.managed_calls.add(call) + + return call, drive + return create class _ChannelConnectivityState(object): @@ -847,6 +896,7 @@ def _options(options): class Channel(grpc.Channel): + """A cygrpc.Channel-backed implementation of grpc.Channel.""" def __init__(self, target, options, credentials): """Constructor. @@ -871,25 +921,25 @@ class Channel(grpc.Channel): def unary_unary( self, method, request_serializer=None, response_deserializer=None): return _UnaryUnaryMultiCallable( - self._channel, _create_channel_managed_call(self._call_state), + self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), request_serializer, response_deserializer) def unary_stream( self, method, request_serializer=None, response_deserializer=None): return _UnaryStreamMultiCallable( - self._channel, _create_channel_managed_call(self._call_state), + self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), request_serializer, response_deserializer) def stream_unary( self, method, request_serializer=None, response_deserializer=None): return _StreamUnaryMultiCallable( - self._channel, _create_channel_managed_call(self._call_state), + self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), request_serializer, response_deserializer) def stream_stream( self, method, request_serializer=None, response_deserializer=None): return _StreamStreamMultiCallable( - self._channel, _create_channel_managed_call(self._call_state), + self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), request_serializer, response_deserializer) def __del__(self): diff --git a/src/python/grpcio/grpc/_utilities.py b/src/python/grpcio/grpc/_utilities.py index 4850967fbc..a375896e6e 100644 --- a/src/python/grpcio/grpc/_utilities.py +++ b/src/python/grpcio/grpc/_utilities.py @@ -53,13 +53,17 @@ class RpcMethodHandler( pass -class DictionaryGenericHandler(grpc.GenericRpcHandler): +class DictionaryGenericHandler(grpc.ServiceRpcHandler): def __init__(self, service, method_handlers): + self._name = service self._method_handlers = { _common.fully_qualified_method(service, method): method_handler for method, method_handler in six.iteritems(method_handlers)} + def service_name(self): + return self._name + def service(self, handler_call_details): return self._method_handlers.get(handler_call_details.method) diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 5affff10f5..e27e9e181d 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -82,6 +82,8 @@ CORE_SOURCE_FILES = [ 'src/core/lib/channel/connected_channel.c', 'src/core/lib/channel/deadline_filter.c', 'src/core/lib/channel/handshaker.c', + 'src/core/lib/channel/handshaker_factory.c', + 'src/core/lib/channel/handshaker_registry.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', 'src/core/lib/channel/message_size_filter.c', @@ -181,8 +183,8 @@ CORE_SOURCE_FILES = [ 'src/core/lib/transport/mdstr_hash_table.c', 'src/core/lib/transport/metadata.c', 'src/core/lib/transport/metadata_batch.c', - 'src/core/lib/transport/method_config.c', 'src/core/lib/transport/pid_controller.c', + 'src/core/lib/transport/service_config.c', 'src/core/lib/transport/static_metadata.c', 'src/core/lib/transport/timeout_encoding.c', 'src/core/lib/transport/transport.c', @@ -226,9 +228,9 @@ CORE_SOURCE_FILES = [ 'src/core/lib/security/credentials/plugin/plugin_credentials.c', 'src/core/lib/security/credentials/ssl/ssl_credentials.c', 'src/core/lib/security/transport/client_auth_filter.c', - 'src/core/lib/security/transport/handshake.c', 'src/core/lib/security/transport/secure_endpoint.c', 'src/core/lib/security/transport/security_connector.c', + 'src/core/lib/security/transport/security_handshaker.c', 'src/core/lib/security/transport/server_auth_filter.c', 'src/core/lib/security/transport/tsi_error.c', 'src/core/lib/security/util/b64.c', @@ -237,6 +239,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/tsi/fake_transport_security.c', 'src/core/lib/tsi/ssl_transport_security.c', 'src/core/lib/tsi/transport_security.c', + 'src/core/ext/transport/chttp2/server/chttp2_server.c', 'src/core/ext/transport/chttp2/client/secure/secure_channel_create.c', 'src/core/ext/client_channel/channel_connectivity.c', 'src/core/ext/client_channel/client_channel.c', @@ -256,6 +259,7 @@ CORE_SOURCE_FILES = [ 'src/core/ext/client_channel/subchannel.c', 'src/core/ext/client_channel/subchannel_index.c', 'src/core/ext/client_channel/uri_parser.c', + 'src/core/ext/transport/chttp2/client/chttp2_connector.c', 'src/core/ext/transport/chttp2/server/insecure/server_chttp2.c', 'src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c', 'src/core/ext/transport/chttp2/client/insecure/channel_create.c', diff --git a/src/python/grpcio/support.py b/src/python/grpcio/support.py index f363f5fdc5..b226e690fd 100644 --- a/src/python/grpcio/support.py +++ b/src/python/grpcio/support.py @@ -100,9 +100,15 @@ def diagnose_compile_error(build_ext, error): .format(source) ) +def diagnose_attribute_error(build_ext, error): + if any('_needs_stub' in arg for arg in error.args): + raise commands.CommandError( + "We expect a missing `_needs_stub` attribute from older versions of " + "setuptools. Consider upgrading setuptools.") _ERROR_DIAGNOSES = { - errors.CompileError: diagnose_compile_error + errors.CompileError: diagnose_compile_error, + AttributeError: diagnose_attribute_error } def diagnose_build_ext_error(build_ext, error, formatted): diff --git a/src/python/grpcio_health_checking/MANIFEST.in b/src/python/grpcio_health_checking/MANIFEST.in index 7407f646d1..5255e4c403 100644 --- a/src/python/grpcio_health_checking/MANIFEST.in +++ b/src/python/grpcio_health_checking/MANIFEST.in @@ -1,4 +1,4 @@ include grpc_version.py include health_commands.py -graft grpc +graft grpc_health global-exclude *.pyc diff --git a/src/python/grpcio_health_checking/grpc/__init__.py b/src/python/grpcio_health_checking/grpc/__init__.py deleted file mode 100644 index fcc7048815..0000000000 --- a/src/python/grpcio_health_checking/grpc/__init__.py +++ /dev/null @@ -1,30 +0,0 @@ -# Copyright 2015, Google Inc. -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are -# met: -# -# * Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# * Redistributions in binary form must reproduce the above -# copyright notice, this list of conditions and the following disclaimer -# in the documentation and/or other materials provided with the -# distribution. -# * Neither the name of Google Inc. nor the names of its -# contributors may be used to endorse or promote products derived from -# this software without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -# 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. - -__import__('pkg_resources').declare_namespace(__name__) diff --git a/src/python/grpcio_health_checking/grpc/health/__init__.py b/src/python/grpcio_health_checking/grpc_health/__init__.py index 7086519106..7086519106 100644 --- a/src/python/grpcio_health_checking/grpc/health/__init__.py +++ b/src/python/grpcio_health_checking/grpc_health/__init__.py diff --git a/src/python/grpcio_health_checking/grpc/health/v1/__init__.py b/src/python/grpcio_health_checking/grpc_health/v1/__init__.py index 7086519106..7086519106 100644 --- a/src/python/grpcio_health_checking/grpc/health/v1/__init__.py +++ b/src/python/grpcio_health_checking/grpc_health/v1/__init__.py diff --git a/src/python/grpcio_health_checking/grpc/health/v1/health.py b/src/python/grpcio_health_checking/grpc_health/v1/health.py index 8108ac1096..0df679b0e2 100644 --- a/src/python/grpcio_health_checking/grpc/health/v1/health.py +++ b/src/python/grpcio_health_checking/grpc_health/v1/health.py @@ -33,7 +33,7 @@ import threading import grpc -from grpc.health.v1 import health_pb2 +from grpc_health.v1 import health_pb2 class HealthServicer(health_pb2.HealthServicer): diff --git a/src/python/grpcio_health_checking/health_commands.py b/src/python/grpcio_health_checking/health_commands.py index 66df25da63..0c420a655f 100644 --- a/src/python/grpcio_health_checking/health_commands.py +++ b/src/python/grpcio_health_checking/health_commands.py @@ -54,7 +54,7 @@ class CopyProtoModules(setuptools.Command): if os.path.isfile(HEALTH_PROTO): shutil.copyfile( HEALTH_PROTO, - os.path.join(ROOT_DIR, 'grpc/health/v1/health.proto')) + os.path.join(ROOT_DIR, 'grpc_health/v1/health.proto')) class BuildPackageProtos(setuptools.Command): @@ -74,5 +74,5 @@ class BuildPackageProtos(setuptools.Command): # directory is provided as an 'include' directory. We assume it's the '' key # to `self.distribution.package_dir` (and get a key error if it's not # there). - from grpc.tools import command + from grpc_tools import command command.build_package_protos(self.distribution.package_dir['']) diff --git a/src/python/grpcio_health_checking/setup.py b/src/python/grpcio_health_checking/setup.py index 8c92ee16a9..e88f389ba8 100644 --- a/src/python/grpcio_health_checking/setup.py +++ b/src/python/grpcio_health_checking/setup.py @@ -66,7 +66,6 @@ setuptools.setup( license='3-clause BSD', package_dir=PACKAGE_DIRECTORIES, packages=setuptools.find_packages('.'), - namespace_packages=['grpc'], install_requires=INSTALL_REQUIRES, setup_requires=SETUP_REQUIRES, cmdclass=COMMAND_CLASS diff --git a/src/python/grpcio_reflection/grpc/__init__.py b/src/python/grpcio_reflection/grpc/__init__.py deleted file mode 100644 index 70ac5edd48..0000000000 --- a/src/python/grpcio_reflection/grpc/__init__.py +++ /dev/null @@ -1,30 +0,0 @@ -# Copyright 2016, Google Inc. -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are -# met: -# -# * Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# * Redistributions in binary form must reproduce the above -# copyright notice, this list of conditions and the following disclaimer -# in the documentation and/or other materials provided with the -# distribution. -# * Neither the name of Google Inc. nor the names of its -# contributors may be used to endorse or promote products derived from -# this software without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -# 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. - -__import__('pkg_resources').declare_namespace(__name__) diff --git a/src/python/grpcio_reflection/grpc/reflection/__init__.py b/src/python/grpcio_reflection/grpc_reflection/__init__.py index d5ad73a74a..d5ad73a74a 100644 --- a/src/python/grpcio_reflection/grpc/reflection/__init__.py +++ b/src/python/grpcio_reflection/grpc_reflection/__init__.py diff --git a/src/python/grpcio_reflection/grpc/reflection/v1alpha/__init__.py b/src/python/grpcio_reflection/grpc_reflection/v1alpha/__init__.py index d5ad73a74a..d5ad73a74a 100644 --- a/src/python/grpcio_reflection/grpc/reflection/v1alpha/__init__.py +++ b/src/python/grpcio_reflection/grpc_reflection/v1alpha/__init__.py diff --git a/src/python/grpcio_reflection/grpc/reflection/v1alpha/reflection.py b/src/python/grpcio_reflection/grpc_reflection/v1alpha/reflection.py index 3c399b0d79..bfcbce8e04 100644 --- a/src/python/grpcio_reflection/grpc/reflection/v1alpha/reflection.py +++ b/src/python/grpcio_reflection/grpc_reflection/v1alpha/reflection.py @@ -35,7 +35,7 @@ import grpc from google.protobuf import descriptor_pb2 from google.protobuf import descriptor_pool -from grpc.reflection.v1alpha import reflection_pb2 +from grpc_reflection.v1alpha import reflection_pb2 _POOL = descriptor_pool.Default() diff --git a/src/python/grpcio_reflection/reflection_commands.py b/src/python/grpcio_reflection/reflection_commands.py index d189aee577..dee5491e0a 100644 --- a/src/python/grpcio_reflection/reflection_commands.py +++ b/src/python/grpcio_reflection/reflection_commands.py @@ -54,7 +54,7 @@ class CopyProtoModules(setuptools.Command): if os.path.isfile(HEALTH_PROTO): shutil.copyfile( HEALTH_PROTO, - os.path.join(ROOT_DIR, 'grpc/reflection/v1alpha/reflection.proto')) + os.path.join(ROOT_DIR, 'grpc_reflection/v1alpha/reflection.proto')) class BuildPackageProtos(setuptools.Command): @@ -74,5 +74,5 @@ class BuildPackageProtos(setuptools.Command): # directory is provided as an 'include' directory. We assume it's the '' key # to `self.distribution.package_dir` (and get a key error if it's not # there). - from grpc.tools import command + from grpc_tools import command command.build_package_protos(self.distribution.package_dir['']) diff --git a/src/python/grpcio_reflection/setup.py b/src/python/grpcio_reflection/setup.py index df95af4de1..cfc41f4fe7 100644 --- a/src/python/grpcio_reflection/setup.py +++ b/src/python/grpcio_reflection/setup.py @@ -66,7 +66,6 @@ setuptools.setup( license='3-clause BSD', package_dir=PACKAGE_DIRECTORIES, packages=setuptools.find_packages('.'), - namespace_packages=['grpc'], install_requires=INSTALL_REQUIRES, setup_requires=SETUP_REQUIRES, cmdclass=COMMAND_CLASS diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py index 5ee551cfe1..e822971fe0 100644 --- a/src/python/grpcio_tests/commands.py +++ b/src/python/grpcio_tests/commands.py @@ -100,7 +100,7 @@ class BuildProtoModules(setuptools.Command): pass def run(self): - import grpc.tools.protoc as protoc + import grpc_tools.protoc as protoc include_regex = re.compile(self.include) exclude_regex = re.compile(self.exclude) if self.exclude else None @@ -116,7 +116,7 @@ class BuildProtoModules(setuptools.Command): # but we currently have name conflicts in src/proto for path in paths: command = [ - 'grpc.tools.protoc', + 'grpc_tools.protoc', '-I {}'.format(PROTO_STEM), '--python_out={}'.format(PROTO_STEM), '--grpc_python_out={}'.format(PROTO_STEM), diff --git a/src/python/grpcio_tests/setup.py b/src/python/grpcio_tests/setup.py index 01d5fa875b..375fbd6c77 100644 --- a/src/python/grpcio_tests/setup.py +++ b/src/python/grpcio_tests/setup.py @@ -35,7 +35,7 @@ import sys import setuptools -import grpc.tools.command +import grpc_tools.command PY3 = sys.version_info.major == 3 @@ -68,7 +68,7 @@ COMMAND_CLASS = { # Run `preprocess` *before* doing any packaging! 'preprocess': commands.GatherProto, - 'build_package_protos': grpc.tools.command.BuildPackageProtos, + 'build_package_protos': grpc_tools.command.BuildPackageProtos, 'build_py': commands.BuildPy, 'run_interop': commands.RunInterop, 'test_lite': commands.TestLite diff --git a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py index 80300d13df..5dde72b169 100644 --- a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py +++ b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py @@ -27,14 +27,14 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -"""Tests of grpc.health.v1.health.""" +"""Tests of grpc_health.v1.health.""" import unittest import grpc from grpc.framework.foundation import logging_pool -from grpc.health.v1 import health -from grpc.health.v1 import health_pb2 +from grpc_health.v1 import health +from grpc_health.v1 import health_pb2 from tests.unit.framework.common import test_constants diff --git a/src/python/grpcio_tests/tests/http2/_negative_http2_client.py b/src/python/grpcio_tests/tests/http2/_negative_http2_client.py new file mode 100644 index 0000000000..f8604683b3 --- /dev/null +++ b/src/python/grpcio_tests/tests/http2/_negative_http2_client.py @@ -0,0 +1,153 @@ +# Copyright 2016, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# 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. + +"""The Python client used to test negative http2 conditions.""" + +import argparse + +import grpc +from src.proto.grpc.testing import test_pb2 +from src.proto.grpc.testing import messages_pb2 + +def _validate_payload_type_and_length(response, expected_type, expected_length): + if response.payload.type is not expected_type: + raise ValueError( + 'expected payload type %s, got %s' % + (expected_type, type(response.payload.type))) + elif len(response.payload.body) != expected_length: + raise ValueError( + 'expected payload body size %d, got %d' % + (expected_length, len(response.payload.body))) + +def _expect_status_code(call, expected_code): + if call.code() != expected_code: + raise ValueError( + 'expected code %s, got %s' % (expected_code, call.code())) + +def _expect_status_details(call, expected_details): + if call.details() != expected_details: + raise ValueError( + 'expected message %s, got %s' % (expected_details, call.details())) + +def _validate_status_code_and_details(call, expected_code, expected_details): + _expect_status_code(call, expected_code) + _expect_status_details(call, expected_details) + +# common requests +_REQUEST_SIZE = 314159 +_RESPONSE_SIZE = 271828 + +_SIMPLE_REQUEST = messages_pb2.SimpleRequest( + response_type=messages_pb2.COMPRESSABLE, + response_size=_RESPONSE_SIZE, + payload=messages_pb2.Payload(body=b'\x00' * _REQUEST_SIZE)) + +def _goaway(stub): + first_response = stub.UnaryCall(_SIMPLE_REQUEST) + _validate_payload_type_and_length(first_response, + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + second_response = stub.UnaryCall(_SIMPLE_REQUEST) + _validate_payload_type_and_length(second_response, + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + +def _rst_after_header(stub): + resp_future = stub.UnaryCall.future(_SIMPLE_REQUEST) + _validate_status_code_and_details(resp_future, grpc.StatusCode.UNAVAILABLE, "") + +def _rst_during_data(stub): + resp_future = stub.UnaryCall.future(_SIMPLE_REQUEST) + _validate_status_code_and_details(resp_future, grpc.StatusCode.UNKNOWN, "") + +def _rst_after_data(stub): + resp_future = stub.UnaryCall.future(_SIMPLE_REQUEST) + _validate_payload_type_and_length(next(resp_future), + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + _validate_status_code_and_details(resp_future, grpc.StatusCode.UNKNOWN, "") + +def _ping(stub): + response = stub.UnaryCall(_SIMPLE_REQUEST) + _validate_payload_type_and_length(response, + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + +def _max_streams(stub): + # send one req to ensure server sets MAX_STREAMS + response = stub.UnaryCall(_SIMPLE_REQUEST) + _validate_payload_type_and_length(response, + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + + # give the streams a workout + futures = [] + for _ in range(15): + futures.append(stub.UnaryCall.future(_SIMPLE_REQUEST)) + for future in futures: + _validate_payload_type_and_length(future.result(), + messages_pb2.COMPRESSABLE, _RESPONSE_SIZE) + +def _run_test_case(test_case, stub): + if test_case == 'goaway': + _goaway(stub) + elif test_case == 'rst_after_header': + _rst_after_header(stub) + elif test_case == 'rst_during_data': + _rst_during_data(stub) + elif test_case == 'rst_after_data': + _rst_after_data(stub) + elif test_case =='ping': + _ping(stub) + elif test_case == 'max_streams': + _max_streams(stub) + else: + raise ValueError("Invalid test case: %s" % test_case) + +def _args(): + parser = argparse.ArgumentParser() + parser.add_argument( + '--server_host', help='the host to which to connect', type=str, + default="127.0.0.1") + parser.add_argument( + '--server_port', help='the port to which to connect', type=int, + default="8080") + parser.add_argument( + '--test_case', help='the test case to execute', type=str, + default="goaway") + return parser.parse_args() + +def _stub(server_host, server_port): + target = '{}:{}'.format(server_host, server_port) + channel = grpc.insecure_channel(target) + return test_pb2.TestServiceStub(channel) + +def main(): + args = _args() + stub = _stub(args.server_host, args.server_port) + _run_test_case(args.test_case, stub) + + +if __name__ == '__main__': + main() diff --git a/src/python/grpcio_tests/tests/interop/_insecure_interop_test.py b/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py index 936c895bd2..4fb22b4d9d 100644 --- a/src/python/grpcio_tests/tests/interop/_insecure_interop_test.py +++ b/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py @@ -35,13 +35,13 @@ import unittest import grpc from src.proto.grpc.testing import test_pb2 -from tests.interop import _interop_test_case +from tests.interop import _intraop_test_case from tests.interop import methods from tests.interop import server -class InsecureInteropTest( - _interop_test_case.InteropTestCase, +class InsecureIntraopTest( + _intraop_test_case.IntraopTestCase, unittest.TestCase): def setUp(self): diff --git a/src/python/grpcio_tests/tests/interop/_interop_test_case.py b/src/python/grpcio_tests/tests/interop/_intraop_test_case.py index ccea17a66d..fe1c173992 100644 --- a/src/python/grpcio_tests/tests/interop/_interop_test_case.py +++ b/src/python/grpcio_tests/tests/interop/_intraop_test_case.py @@ -32,7 +32,7 @@ from tests.interop import methods -class InteropTestCase(object): +class IntraopTestCase(object): """Unit test methods. This class must be mixed in with unittest.TestCase and a class that defines diff --git a/src/python/grpcio_tests/tests/interop/_secure_interop_test.py b/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py index eaca553e1b..3665c69726 100644 --- a/src/python/grpcio_tests/tests/interop/_secure_interop_test.py +++ b/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py @@ -35,15 +35,15 @@ import unittest import grpc from src.proto.grpc.testing import test_pb2 -from tests.interop import _interop_test_case +from tests.interop import _intraop_test_case from tests.interop import methods from tests.interop import resources _SERVER_HOST_OVERRIDE = 'foo.test.google.fr' -class SecureInteropTest( - _interop_test_case.InteropTestCase, +class SecureIntraopTest( + _intraop_test_case.IntraopTestCase, unittest.TestCase): def setUp(self): diff --git a/src/python/grpcio_tests/tests/interop/client.py b/src/python/grpcio_tests/tests/interop/client.py index 4fbf58f7d9..afaa466254 100644 --- a/src/python/grpcio_tests/tests/interop/client.py +++ b/src/python/grpcio_tests/tests/interop/client.py @@ -43,11 +43,13 @@ from tests.interop import resources def _args(): parser = argparse.ArgumentParser() parser.add_argument( - '--server_host', help='the host to which to connect', type=str) + '--server_host', help='the host to which to connect', type=str, + default="127.0.0.1") parser.add_argument( '--server_port', help='the port to which to connect', type=int) parser.add_argument( - '--test_case', help='the test case to execute', type=str) + '--test_case', help='the test case to execute', type=str, + default="large_unary") parser.add_argument( '--use_tls', help='require a secure connection', default=False, type=resources.parse_bool) @@ -55,7 +57,7 @@ def _args(): '--use_test_ca', help='replace platform root CAs with ca.pem', default=False, type=resources.parse_bool) parser.add_argument( - '--server_host_override', + '--server_host_override', default="foo.test.google.fr", help='the server host to which to claim to connect', type=str) parser.add_argument('--oauth_scope', help='scope for OAuth tokens', type=str) parser.add_argument( diff --git a/src/python/grpcio_tests/tests/interop/methods.py b/src/python/grpcio_tests/tests/interop/methods.py index 52e56f3502..9038ae5751 100644 --- a/src/python/grpcio_tests/tests/interop/methods.py +++ b/src/python/grpcio_tests/tests/interop/methods.py @@ -33,7 +33,6 @@ import enum import json import os import threading -import time from oauth2client import client as oauth2client_client @@ -196,16 +195,6 @@ def _server_streaming(stub): response, messages_pb2.COMPRESSABLE, sizes[index]) -def _cancel_after_begin(stub): - sizes = (27182, 8, 1828, 45904,) - payloads = (messages_pb2.Payload(body=b'\x00' * size) for size in sizes) - requests = (messages_pb2.StreamingInputCallRequest(payload=payload) - for payload in payloads) - response_future = stub.StreamingInputCall.future(requests) - response_future.cancel() - if not response_future.cancelled(): - raise ValueError('expected call to be cancelled') - class _Pipe(object): @@ -265,6 +254,16 @@ def _ping_pong(stub): response, messages_pb2.COMPRESSABLE, response_size) +def _cancel_after_begin(stub): + with _Pipe() as pipe: + response_future = stub.StreamingInputCall.future(pipe) + response_future.cancel() + if not response_future.cancelled(): + raise ValueError('expected cancelled method to return True') + if response_future.code() is not grpc.StatusCode.CANCELLED: + raise ValueError('expected status code CANCELLED') + + def _cancel_after_first_response(stub): request_response_sizes = (31415, 9, 2653, 58979,) request_payload_sizes = (27182, 8, 1828, 45904,) @@ -302,7 +301,6 @@ def _timeout_on_sleeping_server(stub): response_type=messages_pb2.COMPRESSABLE, payload=messages_pb2.Payload(body=b'\x00' * request_payload_size)) pipe.add(request) - time.sleep(0.1) try: next(response_iterator) except grpc.RpcError as rpc_error: diff --git a/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py b/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py index 089366a8c7..f8ae05bb7a 100644 --- a/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py +++ b/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py @@ -44,7 +44,7 @@ import threading import unittest import grpc -from grpc.tools import protoc +from grpc_tools import protoc from tests.unit.framework.common import test_constants _MESSAGES_IMPORT = b'import "messages.proto";' @@ -167,7 +167,7 @@ class SameSeparateTest(unittest.TestCase, SeparateTestMixin): '', '--proto_path={}'.format(self.proto_directory), '--python_out={}'.format(self.python_out_directory), - '--grpc_python_out={}'.format(self.grpc_python_out_directory), + '--grpc_python_out=grpc_2_0:{}'.format(self.grpc_python_out_directory), same_proto_file, ]) if protoc_result != 0: @@ -241,7 +241,7 @@ class SplitCommonTest(unittest.TestCase, CommonTestMixin): '', '--proto_path={}'.format(self.proto_directory), '--python_out={}'.format(self.python_out_directory), - '--grpc_python_out={}'.format(self.python_out_directory), + '--grpc_python_out={}'.format(self.grpc_python_out_directory), services_proto_file, messages_proto_file, ]) @@ -285,7 +285,7 @@ class SplitSeparateTest(unittest.TestCase, SeparateTestMixin): '', '--proto_path={}'.format(self.proto_directory), '--python_out={}'.format(self.python_out_directory), - '--grpc_python_out={}'.format(self.grpc_python_out_directory), + '--grpc_python_out=grpc_2_0:{}'.format(self.grpc_python_out_directory), services_proto_file, messages_proto_file, ]) diff --git a/src/python/grpcio_tests/tests/qps/benchmark_client.py b/src/python/grpcio_tests/tests/qps/benchmark_client.py index 83b46c914e..650e4756e7 100644 --- a/src/python/grpcio_tests/tests/qps/benchmark_client.py +++ b/src/python/grpcio_tests/tests/qps/benchmark_client.py @@ -68,12 +68,8 @@ class BenchmarkClient: else: channel = grpc.insecure_channel(server) - connected_event = threading.Event() - def wait_for_ready(connectivity): - if connectivity == grpc.ChannelConnectivity.READY: - connected_event.set() - channel.subscribe(wait_for_ready, try_to_connect=True) - connected_event.wait() + # waits for the channel to be ready before we start sending messages + grpc.channel_ready_future(channel).result() if config.payload_config.WhichOneof('payload') == 'simple_params': self._generic = False diff --git a/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py b/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py index 87264cf9ba..43d6c971b5 100644 --- a/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py +++ b/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py @@ -27,14 +27,14 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -"""Tests of grpc.reflection.v1alpha.reflection.""" +"""Tests of grpc_reflection.v1alpha.reflection.""" import unittest import grpc from grpc.framework.foundation import logging_pool -from grpc.reflection.v1alpha import reflection -from grpc.reflection.v1alpha import reflection_pb2 +from grpc_reflection.v1alpha import reflection +from grpc_reflection.v1alpha import reflection_pb2 from google.protobuf import descriptor_pool from google.protobuf import descriptor_pb2 @@ -75,7 +75,7 @@ class ReflectionServicerTest(unittest.TestCase): file_by_filename='i-donut-exist' ), ) - responses = tuple(self._stub.ServerReflectionInfo(requests)) + responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) expected_responses = ( reflection_pb2.ServerReflectionResponse( valid_host='', @@ -93,7 +93,7 @@ class ReflectionServicerTest(unittest.TestCase): ) ), ) - self.assertEqual(expected_responses, responses) + self.assertSequenceEqual(expected_responses, responses) def testFileBySymbol(self): requests = ( @@ -104,7 +104,7 @@ class ReflectionServicerTest(unittest.TestCase): file_containing_symbol='i.donut.exist.co.uk.org.net.me.name.foo' ), ) - responses = tuple(self._stub.ServerReflectionInfo(requests)) + responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) expected_responses = ( reflection_pb2.ServerReflectionResponse( valid_host='', @@ -122,7 +122,7 @@ class ReflectionServicerTest(unittest.TestCase): ) ), ) - self.assertEqual(expected_responses, responses) + self.assertSequenceEqual(expected_responses, responses) @unittest.skip('TODO(atash): implement file-containing-extension reflection ' '(see https://github.com/google/protobuf/issues/2248)') @@ -141,7 +141,7 @@ class ReflectionServicerTest(unittest.TestCase): ), ), ) - responses = tuple(self._stub.ServerReflectionInfo(requests)) + responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) expected_responses = ( reflection_pb2.ServerReflectionResponse( valid_host='', @@ -159,7 +159,7 @@ class ReflectionServicerTest(unittest.TestCase): ) ), ) - self.assertEqual(expected_responses, responses) + self.assertSequenceEqual(expected_responses, responses) def testListServices(self): requests = ( @@ -167,7 +167,7 @@ class ReflectionServicerTest(unittest.TestCase): list_services='', ), ) - responses = tuple(self._stub.ServerReflectionInfo(requests)) + responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) expected_responses = ( reflection_pb2.ServerReflectionResponse( valid_host='', @@ -179,7 +179,7 @@ class ReflectionServicerTest(unittest.TestCase): ) ), ) - self.assertEqual(expected_responses, responses) + self.assertSequenceEqual(expected_responses, responses) if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/stress/client.py b/src/python/grpcio_tests/tests/stress/client.py index 975f33b4c1..b8116729b5 100644 --- a/src/python/grpcio_tests/tests/stress/client.py +++ b/src/python/grpcio_tests/tests/stress/client.py @@ -39,6 +39,7 @@ from src.proto.grpc.testing import metrics_pb2 from src.proto.grpc.testing import test_pb2 from tests.interop import methods +from tests.interop import resources from tests.qps import histogram from tests.stress import metrics_server from tests.stress import test_runner @@ -71,6 +72,16 @@ def _args(): '--metrics_port', help='the port to listen for metrics requests on', default=8081, type=int) + parser.add_argument( + '--use_test_ca', + help='Whether to use our fake CA. Requires --use_tls=true', + default=False, type=bool) + parser.add_argument( + '--use_tls', + help='Whether to use TLS', default=False, type=bool) + parser.add_argument( + '--server_host_override', default="foo.test.google.fr", + help='the server host to which to claim to connect', type=str) return parser.parse_args() @@ -90,6 +101,22 @@ def _parse_weighted_test_cases(test_case_args): weighted_test_cases[test_case] = int(weight) return weighted_test_cases +def _get_channel(target, args): + if args.use_tls: + if args.use_test_ca: + root_certificates = resources.test_root_certificates() + else: + root_certificates = None # will load default roots. + channel_credentials = grpc.ssl_channel_credentials( + root_certificates=root_certificates) + options = (('grpc.ssl_target_name_override', args.server_host_override,),) + channel = grpc.secure_channel(target, channel_credentials, options=options) + else: + channel = grpc.insecure_channel(target) + + # waits for the channel to be ready before we start sending messages + grpc.channel_ready_future(channel).result() + return channel def run_test(args): test_cases = _parse_weighted_test_cases(args.test_cases) @@ -108,7 +135,7 @@ def run_test(args): for test_server_target in test_server_targets: for _ in xrange(args.num_channels_per_server): - channel = grpc.insecure_channel(test_server_target) + channel = _get_channel(test_server_target, args) for _ in xrange(args.num_stubs_per_channel): stub = test_pb2.TestServiceStub(channel) runner = test_runner.TestRunner(stub, test_cases, hist, diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json index dd4a0257f5..70d965d3ca 100644 --- a/src/python/grpcio_tests/tests/tests.json +++ b/src/python/grpcio_tests/tests/tests.json @@ -1,7 +1,7 @@ [ "health_check._health_servicer_test.HealthServicerTest", - "interop._insecure_interop_test.InsecureInteropTest", - "interop._secure_interop_test.SecureInteropTest", + "interop._insecure_intraop_test.InsecureIntraopTest", + "interop._secure_intraop_test.SecureIntraopTest", "protoc_plugin._python_plugin_test.PythonPluginTest", "protoc_plugin._split_definitions_test.SameCommonTest", "protoc_plugin._split_definitions_test.SameSeparateTest", @@ -27,6 +27,8 @@ "unit._cython.cygrpc_test.TypeSmokeTest", "unit._empty_message_test.EmptyMessageTest", "unit._exit_test.ExitTest", + "unit._invalid_metadata_test.InvalidMetadataTest", + "unit._invocation_defects_test.InvocationDefectsTest", "unit._metadata_code_details_test.MetadataCodeDetailsTest", "unit._metadata_test.MetadataTest", "unit._rpc_test.RPCTest", diff --git a/src/python/grpcio_tests/tests/unit/_api_test.py b/src/python/grpcio_tests/tests/unit/_api_test.py index 2fe89499f5..51dc425420 100644 --- a/src/python/grpcio_tests/tests/unit/_api_test.py +++ b/src/python/grpcio_tests/tests/unit/_api_test.py @@ -65,6 +65,7 @@ class AllTest(unittest.TestCase): 'RpcMethodHandler', 'HandlerCallDetails', 'GenericRpcHandler', + 'ServiceRpcHandler', 'Server', 'unary_unary_rpc_method_handler', 'unary_stream_rpc_method_handler', diff --git a/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py b/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py index e0a7d15aa7..46a964db8c 100644 --- a/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py +++ b/src/python/grpcio_tests/tests/unit/_channel_ready_future_test.py @@ -64,7 +64,7 @@ class ChannelReadyFutureTest(unittest.TestCase): ready_future = grpc.channel_ready_future(channel) ready_future.add_done_callback(callback.accept_value) with self.assertRaises(grpc.FutureTimeoutError): - ready_future.result(test_constants.SHORT_TIMEOUT) + ready_future.result(timeout=test_constants.SHORT_TIMEOUT) self.assertFalse(ready_future.cancelled()) self.assertFalse(ready_future.done()) self.assertTrue(ready_future.running()) @@ -85,7 +85,7 @@ class ChannelReadyFutureTest(unittest.TestCase): ready_future = grpc.channel_ready_future(channel) ready_future.add_done_callback(callback.accept_value) - self.assertIsNone(ready_future.result(test_constants.SHORT_TIMEOUT)) + self.assertIsNone(ready_future.result(timeout=test_constants.LONG_TIMEOUT)) value_passed_to_callback = callback.block_until_called() self.assertIs(ready_future, value_passed_to_callback) self.assertFalse(ready_future.cancelled()) diff --git a/src/python/grpcio_tests/tests/unit/_compression_test.py b/src/python/grpcio_tests/tests/unit/_compression_test.py index 83b9109466..4d3f02e917 100644 --- a/src/python/grpcio_tests/tests/unit/_compression_test.py +++ b/src/python/grpcio_tests/tests/unit/_compression_test.py @@ -125,7 +125,7 @@ class CompressionTest(unittest.TestCase): compressed_channel = grpc.insecure_channel('localhost:%d' % self._port, options=[('grpc.default_compression_algorithm', 1)]) multi_callable = compressed_channel.stream_stream(_STREAM_STREAM) - call = multi_callable([request] * test_constants.STREAM_LENGTH) + call = multi_callable(iter([request] * test_constants.STREAM_LENGTH)) for response in call: self.assertEqual(request, response) diff --git a/src/python/grpcio_tests/tests/unit/_empty_message_test.py b/src/python/grpcio_tests/tests/unit/_empty_message_test.py index 131f6e9452..69f4689279 100644 --- a/src/python/grpcio_tests/tests/unit/_empty_message_test.py +++ b/src/python/grpcio_tests/tests/unit/_empty_message_test.py @@ -123,12 +123,12 @@ class EmptyMessageTest(unittest.TestCase): def testStreamUnary(self): response = self._channel.stream_unary(_STREAM_UNARY)( - [_REQUEST] * test_constants.STREAM_LENGTH) + iter([_REQUEST] * test_constants.STREAM_LENGTH)) self.assertEqual(_RESPONSE, response) def testStreamStream(self): response_iterator = self._channel.stream_stream(_STREAM_STREAM)( - [_REQUEST] * test_constants.STREAM_LENGTH) + iter([_REQUEST] * test_constants.STREAM_LENGTH)) self.assertSequenceEqual( [_RESPONSE] * test_constants.STREAM_LENGTH, list(response_iterator)) diff --git a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py index b33802bf57..777527137f 100644 --- a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py +++ b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py @@ -240,7 +240,7 @@ if __name__ == '__main__': multi_callable = channel.stream_unary(method) future = multi_callable.future(infinite_request_iterator()) result, call = multi_callable.with_call( - [REQUEST] * test_constants.STREAM_LENGTH) + iter([REQUEST] * test_constants.STREAM_LENGTH)) elif (args.scenario == IN_FLIGHT_STREAM_STREAM_CALL or args.scenario == IN_FLIGHT_PARTIAL_STREAM_STREAM_CALL): multi_callable = channel.stream_stream(method) diff --git a/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py b/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py new file mode 100644 index 0000000000..2dc225de29 --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py @@ -0,0 +1,175 @@ +# Copyright 2016, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# 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. + +"""Test of RPCs made against gRPC Python's application-layer API.""" + +import unittest + +import grpc + +from tests.unit.framework.common import test_constants + +_SERIALIZE_REQUEST = lambda bytestring: bytestring * 2 +_DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:] +_SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3 +_DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3] + +_UNARY_UNARY = '/test/UnaryUnary' +_UNARY_STREAM = '/test/UnaryStream' +_STREAM_UNARY = '/test/StreamUnary' +_STREAM_STREAM = '/test/StreamStream' + + +def _unary_unary_multi_callable(channel): + return channel.unary_unary(_UNARY_UNARY) + + +def _unary_stream_multi_callable(channel): + return channel.unary_stream( + _UNARY_STREAM, + request_serializer=_SERIALIZE_REQUEST, + response_deserializer=_DESERIALIZE_RESPONSE) + + +def _stream_unary_multi_callable(channel): + return channel.stream_unary( + _STREAM_UNARY, + request_serializer=_SERIALIZE_REQUEST, + response_deserializer=_DESERIALIZE_RESPONSE) + + +def _stream_stream_multi_callable(channel): + return channel.stream_stream(_STREAM_STREAM) + + +class InvalidMetadataTest(unittest.TestCase): + + def setUp(self): + self._channel = grpc.insecure_channel('localhost:8080') + self._unary_unary = _unary_unary_multi_callable(self._channel) + self._unary_stream = _unary_stream_multi_callable(self._channel) + self._stream_unary = _stream_unary_multi_callable(self._channel) + self._stream_stream = _stream_stream_multi_callable(self._channel) + + def testUnaryRequestBlockingUnaryResponse(self): + request = b'\x07\x08' + metadata = (('InVaLiD', 'UnaryRequestBlockingUnaryResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + with self.assertRaises(ValueError) as exception_context: + self._unary_unary(request, metadata=metadata) + self.assertIn(expected_error_details, str(exception_context.exception)) + + def testUnaryRequestBlockingUnaryResponseWithCall(self): + request = b'\x07\x08' + metadata = (('InVaLiD', 'UnaryRequestBlockingUnaryResponseWithCall'),) + expected_error_details = "metadata was invalid: %s" % metadata + with self.assertRaises(ValueError) as exception_context: + self._unary_unary.with_call(request, metadata=metadata) + self.assertIn(expected_error_details, str(exception_context.exception)) + + def testUnaryRequestFutureUnaryResponse(self): + request = b'\x07\x08' + metadata = (('InVaLiD', 'UnaryRequestFutureUnaryResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + response_future = self._unary_unary.future(request, metadata=metadata) + with self.assertRaises(grpc.RpcError) as exception_context: + response_future.result() + self.assertEqual( + exception_context.exception.details(), expected_error_details) + self.assertEqual( + exception_context.exception.code(), grpc.StatusCode.INTERNAL) + self.assertEqual(response_future.details(), expected_error_details) + self.assertEqual(response_future.code(), grpc.StatusCode.INTERNAL) + + def testUnaryRequestStreamResponse(self): + request = b'\x37\x58' + metadata = (('InVaLiD', 'UnaryRequestStreamResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + response_iterator = self._unary_stream(request, metadata=metadata) + with self.assertRaises(grpc.RpcError) as exception_context: + next(response_iterator) + self.assertEqual( + exception_context.exception.details(), expected_error_details) + self.assertEqual( + exception_context.exception.code(), grpc.StatusCode.INTERNAL) + self.assertEqual(response_iterator.details(), expected_error_details) + self.assertEqual(response_iterator.code(), grpc.StatusCode.INTERNAL) + + def testStreamRequestBlockingUnaryResponse(self): + request_iterator = (b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) + metadata = (('InVaLiD', 'StreamRequestBlockingUnaryResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + with self.assertRaises(ValueError) as exception_context: + self._stream_unary(request_iterator, metadata=metadata) + self.assertIn(expected_error_details, str(exception_context.exception)) + + def testStreamRequestBlockingUnaryResponseWithCall(self): + request_iterator = ( + b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) + metadata = (('InVaLiD', 'StreamRequestBlockingUnaryResponseWithCall'),) + expected_error_details = "metadata was invalid: %s" % metadata + multi_callable = _stream_unary_multi_callable(self._channel) + with self.assertRaises(ValueError) as exception_context: + multi_callable.with_call(request_iterator, metadata=metadata) + self.assertIn(expected_error_details, str(exception_context.exception)) + + def testStreamRequestFutureUnaryResponse(self): + request_iterator = ( + b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) + metadata = (('InVaLiD', 'StreamRequestFutureUnaryResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + response_future = self._stream_unary.future( + request_iterator, metadata=metadata) + with self.assertRaises(grpc.RpcError) as exception_context: + response_future.result() + self.assertEqual( + exception_context.exception.details(), expected_error_details) + self.assertEqual( + exception_context.exception.code(), grpc.StatusCode.INTERNAL) + self.assertEqual(response_future.details(), expected_error_details) + self.assertEqual(response_future.code(), grpc.StatusCode.INTERNAL) + + def testStreamRequestStreamResponse(self): + request_iterator = ( + b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) + metadata = (('InVaLiD', 'StreamRequestStreamResponse'),) + expected_error_details = "metadata was invalid: %s" % metadata + response_iterator = self._stream_stream(request_iterator, metadata=metadata) + with self.assertRaises(grpc.RpcError) as exception_context: + next(response_iterator) + self.assertEqual( + exception_context.exception.details(), expected_error_details) + self.assertEqual( + exception_context.exception.code(), grpc.StatusCode.INTERNAL) + self.assertEqual(response_iterator.details(), expected_error_details) + self.assertEqual(response_iterator.code(), grpc.StatusCode.INTERNAL) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py b/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py new file mode 100644 index 0000000000..4312679bb9 --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py @@ -0,0 +1,247 @@ +# Copyright 2016, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# 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. + +import itertools +import threading +import unittest +from concurrent import futures + +import grpc +from grpc.framework.foundation import logging_pool + +from tests.unit.framework.common import test_constants +from tests.unit.framework.common import test_control + +_SERIALIZE_REQUEST = lambda bytestring: bytestring * 2 +_DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:] +_SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3 +_DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3] + +_UNARY_UNARY = '/test/UnaryUnary' +_UNARY_STREAM = '/test/UnaryStream' +_STREAM_UNARY = '/test/StreamUnary' +_STREAM_STREAM = '/test/StreamStream' + + +class _Callback(object): + def __init__(self): + self._condition = threading.Condition() + self._value = None + self._called = False + + def __call__(self, value): + with self._condition: + self._value = value + self._called = True + self._condition.notify_all() + + def value(self): + with self._condition: + while not self._called: + self._condition.wait() + return self._value + + +class _Handler(object): + def __init__(self, control): + self._control = control + + def handle_unary_unary(self, request, servicer_context): + self._control.control() + if servicer_context is not None: + servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) + return request + + def handle_unary_stream(self, request, servicer_context): + for _ in range(test_constants.STREAM_LENGTH): + self._control.control() + yield request + self._control.control() + if servicer_context is not None: + servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) + + def handle_stream_unary(self, request_iterator, servicer_context): + if servicer_context is not None: + servicer_context.invocation_metadata() + self._control.control() + response_elements = [] + for request in request_iterator: + self._control.control() + response_elements.append(request) + self._control.control() + if servicer_context is not None: + servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) + return b''.join(response_elements) + + def handle_stream_stream(self, request_iterator, servicer_context): + self._control.control() + if servicer_context is not None: + servicer_context.set_trailing_metadata((('testkey', 'testvalue',),)) + for request in request_iterator: + self._control.control() + yield request + self._control.control() + + +class _MethodHandler(grpc.RpcMethodHandler): + def __init__( + self, request_streaming, response_streaming, request_deserializer, + response_serializer, unary_unary, unary_stream, stream_unary, + stream_stream): + self.request_streaming = request_streaming + self.response_streaming = response_streaming + self.request_deserializer = request_deserializer + self.response_serializer = response_serializer + self.unary_unary = unary_unary + self.unary_stream = unary_stream + self.stream_unary = stream_unary + self.stream_stream = stream_stream + + +class _GenericHandler(grpc.GenericRpcHandler): + def __init__(self, handler): + self._handler = handler + + def service(self, handler_call_details): + if handler_call_details.method == _UNARY_UNARY: + return _MethodHandler( + False, False, None, None, self._handler.handle_unary_unary, None, + None, None) + elif handler_call_details.method == _UNARY_STREAM: + return _MethodHandler( + False, True, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, + self._handler.handle_unary_stream, None, None) + elif handler_call_details.method == _STREAM_UNARY: + return _MethodHandler( + True, False, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, None, + self._handler.handle_stream_unary, None) + elif handler_call_details.method == _STREAM_STREAM: + return _MethodHandler( + True, True, None, None, None, None, None, + self._handler.handle_stream_stream) + else: + return None + + +class FailAfterFewIterationsCounter(object): + def __init__(self, high, bytestring): + self._current = 0 + self._high = high + self._bytestring = bytestring + + def __iter__(self): + return self + + def __next__(self): + if self._current >= self._high: + raise Exception("This is a deliberate failure in a unit test.") + else: + self._current += 1 + return self._bytestring + + +def _unary_unary_multi_callable(channel): + return channel.unary_unary(_UNARY_UNARY) + + +def _unary_stream_multi_callable(channel): + return channel.unary_stream( + _UNARY_STREAM, + request_serializer=_SERIALIZE_REQUEST, + response_deserializer=_DESERIALIZE_RESPONSE) + + +def _stream_unary_multi_callable(channel): + return channel.stream_unary( + _STREAM_UNARY, + request_serializer=_SERIALIZE_REQUEST, + response_deserializer=_DESERIALIZE_RESPONSE) + + +def _stream_stream_multi_callable(channel): + return channel.stream_stream(_STREAM_STREAM) + + +class InvocationDefectsTest(unittest.TestCase): + def setUp(self): + self._control = test_control.PauseFailControl() + self._handler = _Handler(self._control) + self._server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY) + + self._server = grpc.server(self._server_pool) + port = self._server.add_insecure_port('[::]:0') + self._server.add_generic_rpc_handlers((_GenericHandler(self._handler),)) + self._server.start() + + self._channel = grpc.insecure_channel('localhost:%d' % port) + + def tearDown(self): + self._server.stop(0) + + def testIterableStreamRequestBlockingUnaryResponse(self): + requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)] + multi_callable = _stream_unary_multi_callable(self._channel) + + with self.assertRaises(grpc.RpcError): + response = multi_callable( + requests, + metadata=(('test', 'IterableStreamRequestBlockingUnaryResponse'),)) + + def testIterableStreamRequestFutureUnaryResponse(self): + requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)] + multi_callable = _stream_unary_multi_callable(self._channel) + response_future = multi_callable.future( + requests, + metadata=( + ('test', 'IterableStreamRequestFutureUnaryResponse'),)) + + with self.assertRaises(grpc.RpcError): + response = response_future.result() + + def testIterableStreamRequestStreamResponse(self): + requests = [b'\x77\x58' for _ in range(test_constants.STREAM_LENGTH)] + multi_callable = _stream_stream_multi_callable(self._channel) + response_iterator = multi_callable( + requests, + metadata=(('test', 'IterableStreamRequestStreamResponse'),)) + + with self.assertRaises(grpc.RpcError): + next(response_iterator) + + def testIteratorStreamRequestStreamResponse(self): + requests_iterator = FailAfterFewIterationsCounter( + test_constants.STREAM_LENGTH // 2, b'\x07\x08') + multi_callable = _stream_stream_multi_callable(self._channel) + response_iterator = multi_callable( + requests_iterator, + metadata=(('test', 'IteratorStreamRequestStreamResponse'),)) + + with self.assertRaises(grpc.RpcError): + for _ in range(test_constants.STREAM_LENGTH // 2 + 1): + next(response_iterator) diff --git a/src/python/grpcio_tests/tests/unit/_metadata_test.py b/src/python/grpcio_tests/tests/unit/_metadata_test.py index da73476929..caba53ffcc 100644 --- a/src/python/grpcio_tests/tests/unit/_metadata_test.py +++ b/src/python/grpcio_tests/tests/unit/_metadata_test.py @@ -193,7 +193,7 @@ class MetadataTest(unittest.TestCase): def testStreamUnary(self): multi_callable = self._channel.stream_unary(_STREAM_UNARY) unused_response, call = multi_callable.with_call( - [_REQUEST] * test_constants.STREAM_LENGTH, + iter([_REQUEST] * test_constants.STREAM_LENGTH), metadata=_CLIENT_METADATA) self.assertTrue(test_common.metadata_transmitted( _SERVER_INITIAL_METADATA, call.initial_metadata())) @@ -202,7 +202,7 @@ class MetadataTest(unittest.TestCase): def testStreamStream(self): multi_callable = self._channel.stream_stream(_STREAM_STREAM) - call = multi_callable([_REQUEST] * test_constants.STREAM_LENGTH, + call = multi_callable(iter([_REQUEST] * test_constants.STREAM_LENGTH), metadata=_CLIENT_METADATA) self.assertTrue(test_common.metadata_transmitted( _SERVER_INITIAL_METADATA, call.initial_metadata())) diff --git a/src/python/grpcio_tests/tests/unit/beta/_utilities_test.py b/src/python/grpcio_tests/tests/unit/beta/_utilities_test.py index 90fe10c77c..9cce96cc85 100644 --- a/src/python/grpcio_tests/tests/unit/beta/_utilities_test.py +++ b/src/python/grpcio_tests/tests/unit/beta/_utilities_test.py @@ -66,7 +66,7 @@ class ChannelConnectivityTest(unittest.TestCase): ready_future = utilities.channel_ready_future(channel) ready_future.add_done_callback(callback.accept_value) with self.assertRaises(future.TimeoutError): - ready_future.result(test_constants.SHORT_TIMEOUT) + ready_future.result(timeout=test_constants.SHORT_TIMEOUT) self.assertFalse(ready_future.cancelled()) self.assertFalse(ready_future.done()) self.assertTrue(ready_future.running()) @@ -88,7 +88,7 @@ class ChannelConnectivityTest(unittest.TestCase): ready_future = utilities.channel_ready_future(channel) ready_future.add_done_callback(callback.accept_value) self.assertIsNone( - ready_future.result(test_constants.SHORT_TIMEOUT)) + ready_future.result(timeout=test_constants.LONG_TIMEOUT)) value_passed_to_callback = callback.block_until_called() self.assertIs(ready_future, value_passed_to_callback) self.assertFalse(ready_future.cancelled()) |