diff options
Diffstat (limited to 'src/python')
103 files changed, 2295 insertions, 456 deletions
diff --git a/src/python/.gitignore b/src/python/.gitignore index 7b520579a0..41813129bd 100644 --- a/src/python/.gitignore +++ b/src/python/.gitignore @@ -1,3 +1,4 @@ gens/ *_pb2.py *_pb2_grpc.py +*.egg-info/ diff --git a/src/python/grpcio/_parallel_compile_patch.py b/src/python/grpcio/_parallel_compile_patch.py new file mode 100644 index 0000000000..4d03ef49ba --- /dev/null +++ b/src/python/grpcio/_parallel_compile_patch.py @@ -0,0 +1,63 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Patches the compile() to allow enable parallel compilation of C/C++. + +build_ext has lots of C/C++ files and normally them one by one. +Enabling parallel build helps a lot. +""" + +import distutils.ccompiler +import os + +try: + BUILD_EXT_COMPILER_JOBS = int( + os.environ.get('GRPC_PYTHON_BUILD_EXT_COMPILER_JOBS', '1')) +except ValueError: + BUILD_EXT_COMPILER_JOBS = 1 + + +# monkey-patch for parallel compilation +def _parallel_compile(self, + sources, + output_dir=None, + macros=None, + include_dirs=None, + debug=0, + extra_preargs=None, + extra_postargs=None, + depends=None): + # setup the same way as distutils.ccompiler.CCompiler + # https://github.com/python/cpython/blob/31368a4f0e531c19affe2a1becd25fc316bc7501/Lib/distutils/ccompiler.py#L564 + macros, objects, extra_postargs, pp_opts, build = self._setup_compile( + output_dir, macros, include_dirs, sources, depends, extra_postargs) + cc_args = self._get_cc_args(pp_opts, debug, extra_preargs) + + def _compile_single_file(obj): + try: + src, ext = build[obj] + except KeyError: + return + self._compile(obj, src, ext, cc_args, extra_postargs, pp_opts) + + # run compilation of individual files in parallel + import multiprocessing.pool + multiprocessing.pool.ThreadPool(BUILD_EXT_COMPILER_JOBS).map( + _compile_single_file, objects) + return objects + + +def monkeypatch_compile_maybe(): + """Monkeypatching is dumb, but the build speed gain is worth it.""" + if BUILD_EXT_COMPILER_JOBS > 1: + distutils.ccompiler.CCompiler.compile = _parallel_compile diff --git a/src/python/grpcio/commands.py b/src/python/grpcio/commands.py index 0a3097111f..b805f4277b 100644 --- a/src/python/grpcio/commands.py +++ b/src/python/grpcio/commands.py @@ -39,36 +39,6 @@ PROTO_STEM = os.path.join(GRPC_STEM, 'src', 'proto') PROTO_GEN_STEM = os.path.join(GRPC_STEM, 'src', 'python', 'gens') CYTHON_STEM = os.path.join(PYTHON_STEM, 'grpc', '_cython') -CONF_PY_ADDENDUM = """ -extensions.append('sphinx.ext.napoleon') -napoleon_google_docstring = True -napoleon_numpy_docstring = True -napoleon_include_special_with_doc = True - -html_theme = 'sphinx_rtd_theme' -copyright = "2016, The gRPC Authors" -""" - -API_GLOSSARY = """ - -Glossary -================ - -.. glossary:: - - metadatum - A key-value pair included in the HTTP header. It is a - 2-tuple where the first entry is the key and the - second is the value, i.e. (key, value). The metadata key is an ASCII str, - and must be a valid HTTP header name. The metadata value can be - either a valid HTTP ASCII str, or bytes. If bytes are provided, - the key must end with '-bin', i.e. - ``('binary-metadata-bin', b'\\x00\\xFF')`` - - metadata - A sequence of metadatum. -""" - class CommandError(Exception): """Simple exception class for GRPC custom commands.""" @@ -124,25 +94,14 @@ class SphinxDocumentation(setuptools.Command): def run(self): # We import here to ensure that setup.py has had a chance to install the # relevant package eggs first. - import sphinx - import sphinx.apidoc - metadata = self.distribution.metadata - src_dir = os.path.join(PYTHON_STEM, 'grpc') - sys.path.append(src_dir) - sphinx.apidoc.main([ - '', '--force', '--full', '-H', metadata.name, '-A', metadata.author, - '-V', metadata.version, '-R', metadata.version, '-o', - os.path.join('doc', 'src'), src_dir - ]) - conf_filepath = os.path.join('doc', 'src', 'conf.py') - with open(conf_filepath, 'a') as conf_file: - conf_file.write(CONF_PY_ADDENDUM) - glossary_filepath = os.path.join('doc', 'src', 'grpc.rst') - with open(glossary_filepath, 'a') as glossary_filepath: - glossary_filepath.write(API_GLOSSARY) - sphinx.main( - ['', os.path.join('doc', 'src'), - os.path.join('doc', 'build')]) + import sphinx.cmd.build + source_dir = os.path.join(GRPC_STEM, 'doc', 'python', 'sphinx') + target_dir = os.path.join(GRPC_STEM, 'doc', 'build') + exit_code = sphinx.cmd.build.build_main( + ['-b', 'html', '-W', '--keep-going', source_dir, target_dir]) + if exit_code is not 0: + raise CommandError( + "Documentation generation has warnings or errors") class BuildProjectMetadata(setuptools.Command): @@ -253,6 +212,12 @@ class BuildExt(build_ext.build_ext): LINK_OPTIONS = {} def build_extensions(self): + # This special conditioning is here due to difference of compiler + # behavior in gcc and clang. The clang doesn't take --stdc++11 + # flags but gcc does. Since the setuptools of Python only support + # all C or all C++ compilation, the mix of C and C++ will crash. + # *By default*, the macOS use clang and Linux use gcc, that's why + # the special condition here is checking platform. if "darwin" in sys.platform: config = os.environ.get('CONFIG', 'opt') target_path = os.path.abspath( @@ -274,8 +239,14 @@ class BuildExt(build_ext.build_ext): extra_defines = [ 'EXTRA_DEFINES="GRPC_POSIX_FORK_ALLOW_PTHREAD_ATFORK=1"' ] + # Ensure the BoringSSL are built instead of using system provided + # libraries. It prevents dependency issues while distributing to + # Mac users who use MacPorts to manage their libraries. #17002 + mod_env = dict(os.environ) + mod_env['REQUIRE_CUSTOM_LIBRARIES_opt'] = '1' make_process = subprocess.Popen( ['make'] + extra_defines + targets, + env=mod_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE) make_out, make_err = make_process.communicate() diff --git a/src/python/grpcio/grpc/BUILD.bazel b/src/python/grpcio/grpc/BUILD.bazel index 2e6839ef2d..6958ccdfb6 100644 --- a/src/python/grpcio/grpc/BUILD.bazel +++ b/src/python/grpcio/grpc/BUILD.bazel @@ -13,7 +13,6 @@ py_library( ":interceptor", ":server", "//src/python/grpcio/grpc/_cython:cygrpc", - "//src/python/grpcio/grpc/beta", "//src/python/grpcio/grpc/experimental", "//src/python/grpcio/grpc/framework", requirement('enum34'), diff --git a/src/python/grpcio/grpc/__init__.py b/src/python/grpcio/grpc/__init__.py index 863696d236..6022fc3ef2 100644 --- a/src/python/grpcio/grpc/__init__.py +++ b/src/python/grpcio/grpc/__init__.py @@ -15,12 +15,14 @@ import abc import enum +import logging import sys - import six from grpc._cython import cygrpc as _cygrpc +logging.getLogger(__name__).addHandler(logging.NullHandler()) + ############################## Future Interface ############################### @@ -48,11 +50,13 @@ class Future(six.with_metaclass(abc.ABCMeta)): Returns: bool: Returns True if the computation was canceled. + Returns False under all other circumstances, for example: + 1. computation has begun and could not be canceled. 2. computation has finished 3. computation is scheduled for execution and it is impossible - to determine its state without blocking. + to determine its state without blocking. """ raise NotImplementedError() @@ -66,7 +70,9 @@ class Future(six.with_metaclass(abc.ABCMeta)): bool: Returns True if the computation was cancelled before its result became available. - False under all other circumstances, for example: + + Returns False under all other circumstances, for example: + 1. computation was not cancelled. 2. computation's result is available. """ @@ -79,9 +85,9 @@ class Future(six.with_metaclass(abc.ABCMeta)): This method does not block. Returns: - bool: Returns True if the computation is scheduled for execution or currently executing. + Returns False if the computation already executed or was cancelled. """ raise NotImplementedError() @@ -210,7 +216,33 @@ class ChannelConnectivity(enum.Enum): @enum.unique class StatusCode(enum.Enum): - """Mirrors grpc_status_code in the gRPC Core.""" + """Mirrors grpc_status_code in the gRPC Core. + + Attributes: + OK: Not an error; returned on success + CANCELLED: The operation was cancelled (typically by the caller). + UNKNOWN: Unknown error. + INVALID_ARGUMENT: Client specified an invalid argument. + DEADLINE_EXCEEDED: Deadline expired before operation could complete. + NOT_FOUND: Some requested entity (e.g., file or directory) was not found. + ALREADY_EXISTS: Some entity that we attempted to create (e.g., file or directory) + already exists. + PERMISSION_DENIED: The caller does not have permission to execute the specified + operation. + UNAUTHENTICATED: The request does not have valid authentication credentials for the + operation. + RESOURCE_EXHAUSTED: Some resource has been exhausted, perhaps a per-user quota, or + perhaps the entire file system is out of space. + FAILED_PRECONDITION: Operation was rejected because the system is not in a state + required for the operation's execution. + ABORTED: The operation was aborted, typically due to a concurrency issue + like sequencer check failures, transaction aborts, etc. + UNIMPLEMENTED: Operation is not implemented or not supported/enabled in this service. + INTERNAL: Internal errors. Means some invariants expected by underlying + system has been broken. + UNAVAILABLE: The service is currently unavailable. + DATA_LOSS: Unrecoverable data loss or corruption. + """ OK = (_cygrpc.StatusCode.ok, 'ok') CANCELLED = (_cygrpc.StatusCode.cancelled, 'cancelled') UNKNOWN = (_cygrpc.StatusCode.unknown, 'unknown') @@ -357,6 +389,8 @@ class ClientCallDetails(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional flag t + enable wait for ready mechanism. """ @@ -450,8 +484,7 @@ class StreamUnaryClientInterceptor(six.with_metaclass(abc.ABCMeta)): actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use - `response_future = continuation(client_call_details, - request_iterator)` + `response_future = continuation(client_call_details, request_iterator)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message @@ -462,11 +495,11 @@ class StreamUnaryClientInterceptor(six.with_metaclass(abc.ABCMeta)): request_iterator: An iterator that yields request values for the RPC. Returns: - An object that is both a Call for the RPC and a Future. - In the event of RPC completion, the return Call-Future's - result value will be the response message of the RPC. - Should the event terminate with non-OK status, the returned - Call-Future's exception value will be an RpcError. + An object that is both a Call for the RPC and a Future. + In the event of RPC completion, the return Call-Future's + result value will be the response message of the RPC. + Should the event terminate with non-OK status, the returned + Call-Future's exception value will be an RpcError. """ raise NotImplementedError() @@ -482,13 +515,13 @@ class StreamStreamClientInterceptor(six.with_metaclass(abc.ABCMeta)): request_iterator): """Intercepts a stream-stream invocation. + Args: continuation: A function that proceeds with the invocation by executing the next interceptor in chain or invoking the actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use - `response_iterator = continuation(client_call_details, - request_iterator)` + `response_iterator = continuation(client_call_details, request_iterator)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and an iterator for response values. Drawing response values from the returned Call-iterator may @@ -499,10 +532,10 @@ class StreamStreamClientInterceptor(six.with_metaclass(abc.ABCMeta)): request_iterator: An iterator that yields request values for the RPC. Returns: - An object that is both a Call for the RPC and an iterator of - response values. Drawing response values from the returned - Call-iterator may raise RpcError indicating termination of - the RPC with non-OK status. + An object that is both a Call for the RPC and an iterator of + response values. Drawing response values from the returned + Call-iterator may raise RpcError indicating termination of + the RPC with non-OK status. """ raise NotImplementedError() @@ -609,7 +642,12 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): """Affords invoking a unary-unary RPC from client-side.""" @abc.abstractmethod - def __call__(self, request, timeout=None, metadata=None, credentials=None): + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): """Synchronously invokes the underlying RPC. Args: @@ -619,6 +657,8 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: The response value for the RPC. @@ -631,7 +671,12 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() @abc.abstractmethod - def with_call(self, request, timeout=None, metadata=None, credentials=None): + def with_call(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): """Synchronously invokes the underlying RPC. Args: @@ -641,6 +686,8 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: The response value for the RPC and a Call value for the RPC. @@ -653,7 +700,12 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): raise NotImplementedError() @abc.abstractmethod - def future(self, request, timeout=None, metadata=None, credentials=None): + def future(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): """Asynchronously invokes the underlying RPC. Args: @@ -663,6 +715,8 @@ class UnaryUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: An object that is both a Call for the RPC and a Future. @@ -678,7 +732,12 @@ class UnaryStreamMultiCallable(six.with_metaclass(abc.ABCMeta)): """Affords invoking a unary-stream RPC from client-side.""" @abc.abstractmethod - def __call__(self, request, timeout=None, metadata=None, credentials=None): + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): """Invokes the underlying RPC. Args: @@ -688,6 +747,8 @@ class UnaryStreamMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: An optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: An object that is both a Call for the RPC and an iterator of @@ -706,7 +767,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): """Synchronously invokes the underlying RPC. Args: @@ -717,6 +779,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: The response value for the RPC. @@ -733,7 +797,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): """Synchronously invokes the underlying RPC on the client. Args: @@ -744,6 +809,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: The response value for the RPC and a Call object for the RPC. @@ -760,7 +827,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): """Asynchronously invokes the underlying RPC on the client. Args: @@ -770,6 +838,8 @@ class StreamUnaryMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: An object that is both a Call for the RPC and a Future. @@ -789,7 +859,8 @@ class StreamStreamMultiCallable(six.with_metaclass(abc.ABCMeta)): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): """Invokes the underlying RPC on the client. Args: @@ -799,6 +870,8 @@ class StreamStreamMultiCallable(six.with_metaclass(abc.ABCMeta)): metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. + wait_for_ready: This is an EXPERIMENTAL argument. An optional + flag to enable wait for ready mechanism Returns: An object that is both a Call for the RPC and an iterator of @@ -972,8 +1045,7 @@ class ServicerContext(six.with_metaclass(abc.ABCMeta, RpcContext)): """Gets one or more peer identity(s). Equivalent to - servicer_context.auth_context().get( - servicer_context.peer_identity_key()) + servicer_context.auth_context().get(servicer_context.peer_identity_key()) Returns: An iterable of the identities, or None if the call is not @@ -1651,7 +1723,7 @@ def server(thread_pool, handlers. The interceptors are given control in the order they are specified. This is an EXPERIMENTAL API. options: An optional list of key-value pairs (channel args in gRPC runtime) - to configure the channel. + to configure the channel. maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server will service before returning RESOURCE_EXHAUSTED status, or None to indicate no limit. diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py index eeeb4ddb33..35fa82d56b 100644 --- a/src/python/grpcio/grpc/_channel.py +++ b/src/python/grpcio/grpc/_channel.py @@ -24,7 +24,6 @@ from grpc import _grpcio_metadata from grpc._cython import cygrpc from grpc.framework.foundation import callable_util -logging.basicConfig() _LOGGER = logging.getLogger(__name__) _USER_AGENT = 'grpc-python/{}'.format(_grpcio_metadata.__version__) @@ -176,6 +175,7 @@ def _event_handler(state, response_deserializer): return handle_event +#pylint: disable=too-many-statements def _consume_request_iterator(request_iterator, state, call, request_serializer, event_handler): if cygrpc.is_fork_support_enabled(): @@ -467,10 +467,11 @@ def _end_unary_response_blocking(state, call, with_call, deadline): raise _Rendezvous(state, None, None, deadline) -def _stream_unary_invocation_operationses(metadata): +def _stream_unary_invocation_operationses(metadata, initial_metadata_flags): return ( ( - cygrpc.SendInitialMetadataOperation(metadata, _EMPTY_FLAGS), + cygrpc.SendInitialMetadataOperation(metadata, + initial_metadata_flags), cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ), @@ -478,15 +479,19 @@ def _stream_unary_invocation_operationses(metadata): ) -def _stream_unary_invocation_operationses_and_tags(metadata): +def _stream_unary_invocation_operationses_and_tags(metadata, + initial_metadata_flags): return tuple(( operations, None, - ) for operations in _stream_unary_invocation_operationses(metadata)) + ) + for operations in _stream_unary_invocation_operationses( + metadata, initial_metadata_flags)) class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): + # pylint: disable=too-many-arguments def __init__(self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel @@ -495,15 +500,18 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): self._request_serializer = request_serializer self._response_deserializer = response_deserializer - def _prepare(self, request, timeout, metadata): + def _prepare(self, request, timeout, metadata, wait_for_ready): deadline, serialized_request, rendezvous = _start_unary_request( request, timeout, self._request_serializer) + initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( + wait_for_ready) if serialized_request is None: return None, None, None, rendezvous else: state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None) operations = ( - cygrpc.SendInitialMetadataOperation(metadata, _EMPTY_FLAGS), + cygrpc.SendInitialMetadataOperation(metadata, + initial_metadata_flags), cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS), cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS), @@ -512,9 +520,10 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): ) return state, operations, deadline, None - def _blocking(self, request, timeout, metadata, credentials): + def _blocking(self, request, timeout, metadata, credentials, + wait_for_ready): state, operations, deadline, rendezvous = self._prepare( - request, timeout, metadata) + request, timeout, metadata, wait_for_ready) if state is None: raise rendezvous else: @@ -528,17 +537,34 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): _handle_event(event, state, self._response_deserializer) return state, call, - def __call__(self, request, timeout=None, metadata=None, credentials=None): - state, call, = self._blocking(request, timeout, metadata, credentials) + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): + state, call, = self._blocking(request, timeout, metadata, credentials, + wait_for_ready) return _end_unary_response_blocking(state, call, False, None) - def with_call(self, request, timeout=None, metadata=None, credentials=None): - state, call, = self._blocking(request, timeout, metadata, credentials) + def with_call(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): + state, call, = self._blocking(request, timeout, metadata, credentials, + wait_for_ready) return _end_unary_response_blocking(state, call, True, None) - def future(self, request, timeout=None, metadata=None, credentials=None): + def future(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): state, operations, deadline, rendezvous = self._prepare( - request, timeout, metadata) + request, timeout, metadata, wait_for_ready) if state is None: raise rendezvous else: @@ -553,6 +579,7 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): + # pylint: disable=too-many-arguments def __init__(self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel @@ -561,16 +588,24 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): self._request_serializer = request_serializer self._response_deserializer = response_deserializer - def __call__(self, request, timeout=None, metadata=None, credentials=None): + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): deadline, serialized_request, rendezvous = _start_unary_request( request, timeout, self._request_serializer) + initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( + wait_for_ready) if serialized_request is None: raise rendezvous else: state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) operationses = ( ( - cygrpc.SendInitialMetadataOperation(metadata, _EMPTY_FLAGS), + cygrpc.SendInitialMetadataOperation(metadata, + initial_metadata_flags), cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS), cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), @@ -589,6 +624,7 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): + # pylint: disable=too-many-arguments def __init__(self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel @@ -597,13 +633,17 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): self._request_serializer = request_serializer self._response_deserializer = response_deserializer - def _blocking(self, request_iterator, timeout, metadata, credentials): + def _blocking(self, request_iterator, timeout, metadata, credentials, + wait_for_ready): deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) + initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( + wait_for_ready) call = self._channel.segregated_call( 0, self._method, None, deadline, metadata, None if credentials is None else credentials._credentials, - _stream_unary_invocation_operationses_and_tags(metadata)) + _stream_unary_invocation_operationses_and_tags( + metadata, initial_metadata_flags)) _consume_request_iterator(request_iterator, state, call, self._request_serializer, None) while True: @@ -619,32 +659,38 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): state, call, = self._blocking(request_iterator, timeout, metadata, - credentials) + credentials, wait_for_ready) return _end_unary_response_blocking(state, call, False, None) def with_call(self, request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): state, call, = self._blocking(request_iterator, timeout, metadata, - credentials) + credentials, wait_for_ready) return _end_unary_response_blocking(state, call, True, None) def future(self, request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) event_handler = _event_handler(state, self._response_deserializer) + initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( + wait_for_ready) call = self._managed_call( 0, self._method, None, deadline, metadata, None if credentials is None else credentials._credentials, - _stream_unary_invocation_operationses(metadata), event_handler) + _stream_unary_invocation_operationses( + metadata, initial_metadata_flags), event_handler) _consume_request_iterator(request_iterator, state, call, self._request_serializer, event_handler) return _Rendezvous(state, call, self._response_deserializer, deadline) @@ -652,6 +698,7 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): + # pylint: disable=too-many-arguments def __init__(self, channel, managed_call, method, request_serializer, response_deserializer): self._channel = channel @@ -664,12 +711,16 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): deadline = _deadline(timeout) state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None) + initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( + wait_for_ready) operationses = ( ( - cygrpc.SendInitialMetadataOperation(metadata, _EMPTY_FLAGS), + cygrpc.SendInitialMetadataOperation(metadata, + initial_metadata_flags), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ), (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),), @@ -684,6 +735,24 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): return _Rendezvous(state, call, self._response_deserializer, deadline) +class _InitialMetadataFlags(int): + """Stores immutable initial metadata flags""" + + def __new__(cls, value=_EMPTY_FLAGS): + value &= cygrpc.InitialMetadataFlags.used_mask + return super(_InitialMetadataFlags, cls).__new__(cls, value) + + def with_wait_for_ready(self, wait_for_ready): + if wait_for_ready is not None: + if wait_for_ready: + self = self.__class__(self | cygrpc.InitialMetadataFlags.wait_for_ready | \ + cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set) + elif not wait_for_ready: + self = self.__class__(self & ~cygrpc.InitialMetadataFlags.wait_for_ready | \ + cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set) + return self + + class _ChannelCallState(object): def __init__(self, channel): @@ -980,8 +1049,9 @@ class Channel(grpc.Channel): # for as long as they are in use and to close them after using them, # then deletion of this grpc._channel.Channel instance can be made to # effect closure of the underlying cygrpc.Channel instance. - cygrpc.fork_unregister_channel(self) + if cygrpc is not None: # Globals may have already been collected. + cygrpc.fork_unregister_channel(self) # This prevent the failed-at-initializing object removal from failing. # Though the __init__ failed, the removal will still trigger __del__. - if hasattr(self, "_connectivity_state"): + if _moot is not None and hasattr(self, "_connectivity_state"): _moot(self._connectivity_state) diff --git a/src/python/grpcio/grpc/_common.py b/src/python/grpcio/grpc/_common.py index 3805c7e82a..f69127e38e 100644 --- a/src/python/grpcio/grpc/_common.py +++ b/src/python/grpcio/grpc/_common.py @@ -20,7 +20,6 @@ import six import grpc from grpc._cython import cygrpc -logging.basicConfig() _LOGGER = logging.getLogger(__name__) CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { @@ -66,18 +65,13 @@ def encode(s): if isinstance(s, bytes): return s else: - return s.encode('ascii') + return s.encode('utf8') def decode(b): - if isinstance(b, str): - return b - else: - try: - return b.decode('utf8') - except UnicodeDecodeError: - _LOGGER.exception('Invalid encoding on %s', b) - return b.decode('latin1') + if isinstance(b, bytes): + return b.decode('utf-8', 'replace') + return b def _transform(message, transformer, exception_message): diff --git a/src/python/grpcio/grpc/_cython/BUILD.bazel b/src/python/grpcio/grpc/_cython/BUILD.bazel index cfd3a51d9b..e318298d0a 100644 --- a/src/python/grpcio/grpc/_cython/BUILD.bazel +++ b/src/python/grpcio/grpc/_cython/BUILD.bazel @@ -12,6 +12,7 @@ pyx_library( "_cygrpc/grpc_string.pyx.pxi", "_cygrpc/arguments.pyx.pxi", "_cygrpc/call.pyx.pxi", + "_cygrpc/channelz.pyx.pxi", "_cygrpc/channel.pyx.pxi", "_cygrpc/credentials.pyx.pxi", "_cygrpc/completion_queue.pyx.pxi", diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channelz.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channelz.pyx.pxi new file mode 100644 index 0000000000..113f7976dd --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channelz.pyx.pxi @@ -0,0 +1,69 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +def channelz_get_top_channels(start_channel_id): + cdef char *c_returned_str = grpc_channelz_get_top_channels( + start_channel_id, + ) + if c_returned_str == NULL: + raise ValueError('Failed to get top channels, please ensure your' \ + ' start_channel_id==%s is valid' % start_channel_id) + return c_returned_str + +def channelz_get_servers(start_server_id): + cdef char *c_returned_str = grpc_channelz_get_servers(start_server_id) + if c_returned_str == NULL: + raise ValueError('Failed to get servers, please ensure your' \ + ' start_server_id==%s is valid' % start_server_id) + return c_returned_str + +def channelz_get_server(server_id): + cdef char *c_returned_str = grpc_channelz_get_server(server_id) + if c_returned_str == NULL: + raise ValueError('Failed to get the server, please ensure your' \ + ' server_id==%s is valid' % server_id) + return c_returned_str + +def channelz_get_server_sockets(server_id, start_socket_id): + cdef char *c_returned_str = grpc_channelz_get_server_sockets( + server_id, + start_socket_id, + ) + if c_returned_str == NULL: + raise ValueError('Failed to get server sockets, please ensure your' \ + ' server_id==%s and start_socket_id==%s is valid' % + (server_id, start_socket_id)) + return c_returned_str + +def channelz_get_channel(channel_id): + cdef char *c_returned_str = grpc_channelz_get_channel(channel_id) + if c_returned_str == NULL: + raise ValueError('Failed to get the channel, please ensure your' \ + ' channel_id==%s is valid' % (channel_id)) + return c_returned_str + +def channelz_get_subchannel(subchannel_id): + cdef char *c_returned_str = grpc_channelz_get_subchannel(subchannel_id) + if c_returned_str == NULL: + raise ValueError('Failed to get the subchannel, please ensure your' \ + ' subchannel_id==%s is valid' % (subchannel_id)) + return c_returned_str + +def channelz_get_socket(socket_id): + cdef char *c_returned_str = grpc_channelz_get_socket(socket_id) + if c_returned_str == NULL: + raise ValueError('Failed to get the socket, please ensure your' \ + ' socket_id==%s is valid' % (socket_id)) + return c_returned_str diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pxd.pxi index 8d73215247..1cef726970 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pxd.pxi @@ -15,7 +15,7 @@ cdef class CallCredentials: - cdef grpc_call_credentials *c(self) + cdef grpc_call_credentials *c(self) except * # TODO(https://github.com/grpc/grpc/issues/12531): remove. cdef grpc_call_credentials *c_credentials @@ -36,7 +36,7 @@ cdef class MetadataPluginCallCredentials(CallCredentials): cdef readonly object _metadata_plugin cdef readonly bytes _name - cdef grpc_call_credentials *c(self) + cdef grpc_call_credentials *c(self) except * cdef grpc_call_credentials *_composition(call_credentialses) @@ -46,12 +46,12 @@ cdef class CompositeCallCredentials(CallCredentials): cdef readonly tuple _call_credentialses - cdef grpc_call_credentials *c(self) + cdef grpc_call_credentials *c(self) except * cdef class ChannelCredentials: - cdef grpc_channel_credentials *c(self) + cdef grpc_channel_credentials *c(self) except * # TODO(https://github.com/grpc/grpc/issues/12531): remove. cdef grpc_channel_credentials *c_credentials @@ -68,7 +68,7 @@ cdef class SSLChannelCredentials(ChannelCredentials): cdef readonly object _private_key cdef readonly object _certificate_chain - cdef grpc_channel_credentials *c(self) + cdef grpc_channel_credentials *c(self) except * cdef class CompositeChannelCredentials(ChannelCredentials): @@ -76,7 +76,7 @@ cdef class CompositeChannelCredentials(ChannelCredentials): cdef readonly tuple _call_credentialses cdef readonly ChannelCredentials _channel_credentials - cdef grpc_channel_credentials *c(self) + cdef grpc_channel_credentials *c(self) except * cdef class ServerCertificateConfig: diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi index 63048e8da0..2f51be40ce 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi @@ -35,7 +35,7 @@ def _spawn_callback_async(callback, args): cdef class CallCredentials: - cdef grpc_call_credentials *c(self): + cdef grpc_call_credentials *c(self) except *: raise NotImplementedError() @@ -61,6 +61,7 @@ cdef int _get_metadata( cdef void _destroy(void *state) with gil: cpython.Py_DECREF(<object>state) + grpc_shutdown() cdef class MetadataPluginCallCredentials(CallCredentials): @@ -69,13 +70,14 @@ cdef class MetadataPluginCallCredentials(CallCredentials): self._metadata_plugin = metadata_plugin self._name = name - cdef grpc_call_credentials *c(self): + cdef grpc_call_credentials *c(self) except *: cdef grpc_metadata_credentials_plugin c_metadata_plugin c_metadata_plugin.get_metadata = _get_metadata c_metadata_plugin.destroy = _destroy c_metadata_plugin.state = <void *>self._metadata_plugin c_metadata_plugin.type = self._name cpython.Py_INCREF(self._metadata_plugin) + fork_handlers_and_grpc_init() return grpc_metadata_credentials_create_from_plugin(c_metadata_plugin, NULL) @@ -101,13 +103,13 @@ cdef class CompositeCallCredentials(CallCredentials): def __cinit__(self, call_credentialses): self._call_credentialses = call_credentialses - cdef grpc_call_credentials *c(self): + cdef grpc_call_credentials *c(self) except *: return _composition(self._call_credentialses) cdef class ChannelCredentials: - cdef grpc_channel_credentials *c(self): + cdef grpc_channel_credentials *c(self) except *: raise NotImplementedError() @@ -129,11 +131,13 @@ cdef class SSLSessionCacheLRU: cdef class SSLChannelCredentials(ChannelCredentials): def __cinit__(self, pem_root_certificates, private_key, certificate_chain): + if pem_root_certificates is not None and not isinstance(pem_root_certificates, bytes): + raise TypeError('expected certificate to be bytes, got %s' % (type(pem_root_certificates))) self._pem_root_certificates = pem_root_certificates self._private_key = private_key self._certificate_chain = certificate_chain - cdef grpc_channel_credentials *c(self): + cdef grpc_channel_credentials *c(self) except *: cdef const char *c_pem_root_certificates cdef grpc_ssl_pem_key_cert_pair c_pem_key_certificate_pair if self._pem_root_certificates is None: @@ -162,7 +166,7 @@ cdef class CompositeChannelCredentials(ChannelCredentials): self._call_credentialses = call_credentialses self._channel_credentials = channel_credentials - cdef grpc_channel_credentials *c(self): + cdef grpc_channel_credentials *c(self) except *: cdef grpc_channel_credentials *c_channel_credentials c_channel_credentials = self._channel_credentials.c() cdef grpc_call_credentials *c_call_credentials_composition = _composition( diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi index 4781219319..5bbc10af25 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi @@ -13,6 +13,7 @@ # limitations under the License. cimport libc.time +from libc.stdint cimport intptr_t # Typedef types with approximately the same semantics to provide their names to @@ -121,7 +122,6 @@ cdef extern from "grpc/grpc.h": GRPC_STATUS_DATA_LOSS GRPC_STATUS__DO_NOT_USE - const char *GRPC_ARG_PRIMARY_USER_AGENT_STRING const char *GRPC_ARG_ENABLE_CENSUS const char *GRPC_ARG_MAX_CONCURRENT_STREAMS const char *GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH @@ -140,6 +140,10 @@ cdef extern from "grpc/grpc.h": const int GRPC_WRITE_NO_COMPRESS const int GRPC_WRITE_USED_MASK + const int GRPC_INITIAL_METADATA_WAIT_FOR_READY + const int GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET + const int GRPC_INITIAL_METADATA_USED_MASK + const int GRPC_MAX_COMPLETION_QUEUE_PLUCKERS ctypedef struct grpc_completion_queue: @@ -186,12 +190,6 @@ cdef extern from "grpc/grpc.h": size_t arguments_length "num_args" grpc_arg *arguments "args" - ctypedef enum grpc_compression_level: - GRPC_COMPRESS_LEVEL_NONE - GRPC_COMPRESS_LEVEL_LOW - GRPC_COMPRESS_LEVEL_MED - GRPC_COMPRESS_LEVEL_HIGH - ctypedef enum grpc_stream_compression_level: GRPC_STREAM_COMPRESS_LEVEL_NONE GRPC_STREAM_COMPRESS_LEVEL_LOW @@ -387,6 +385,15 @@ cdef extern from "grpc/grpc.h": void grpc_server_cancel_all_calls(grpc_server *server) nogil void grpc_server_destroy(grpc_server *server) nogil + char* grpc_channelz_get_top_channels(intptr_t start_channel_id) + char* grpc_channelz_get_servers(intptr_t start_server_id) + char* grpc_channelz_get_server(intptr_t server_id) + char* grpc_channelz_get_server_sockets(intptr_t server_id, + intptr_t start_socket_id) + char* grpc_channelz_get_channel(intptr_t channel_id) + char* grpc_channelz_get_subchannel(intptr_t subchannel_id) + char* grpc_channelz_get_socket(intptr_t socket_id) + cdef extern from "grpc/grpc_security.h": diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_gevent.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_gevent.pyx.pxi index f9a1b2856d..a1618d04d0 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_gevent.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_gevent.pyx.pxi @@ -266,7 +266,7 @@ cdef grpc_error* socket_listen(grpc_custom_socket* socket) with gil: (<SocketWrapper>socket.impl).socket.listen(50) return grpc_error_none() -cdef void accept_callback_cython(SocketWrapper s): +cdef void accept_callback_cython(SocketWrapper s) except *: try: conn, address = s.socket.accept() sw = SocketWrapper() diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi index 334e561baa..00a1b23a67 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi @@ -14,7 +14,6 @@ import logging -logging.basicConfig() _LOGGER = logging.getLogger(__name__) # This function will ascii encode unicode string inputs if neccesary. diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pxd.pxi index a18c365807..fc72ac1576 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pxd.pxi @@ -14,10 +14,10 @@ cdef void _store_c_metadata( - metadata, grpc_metadata **c_metadata, size_t *c_count) + metadata, grpc_metadata **c_metadata, size_t *c_count) except * -cdef void _release_c_metadata(grpc_metadata *c_metadata, int count) +cdef void _release_c_metadata(grpc_metadata *c_metadata, int count) except * cdef tuple _metadatum(grpc_slice key_slice, grpc_slice value_slice) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi index c39fef08fa..caf867b569 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi @@ -15,11 +15,17 @@ import collections +class InitialMetadataFlags: + used_mask = GRPC_INITIAL_METADATA_USED_MASK + wait_for_ready = GRPC_INITIAL_METADATA_WAIT_FOR_READY + wait_for_ready_explicitly_set = GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET + + _Metadatum = collections.namedtuple('_Metadatum', ('key', 'value',)) cdef void _store_c_metadata( - metadata, grpc_metadata **c_metadata, size_t *c_count): + metadata, grpc_metadata **c_metadata, size_t *c_count) except *: if metadata is None: c_count[0] = 0 c_metadata[0] = NULL @@ -39,7 +45,7 @@ cdef void _store_c_metadata( c_metadata[0][index].value = _slice_from_bytes(encoded_value) -cdef void _release_c_metadata(grpc_metadata *c_metadata, int count): +cdef void _release_c_metadata(grpc_metadata *c_metadata, int count) except *: if 0 < count: for index in range(count): grpc_slice_unref(c_metadata[index].key) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/operation.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/operation.pxd.pxi index 69a2a4989e..c9df32dadf 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/operation.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/operation.pxd.pxi @@ -15,8 +15,8 @@ cdef class Operation: - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * # TODO(https://github.com/grpc/grpc/issues/7950): Eliminate this! cdef grpc_op c_op @@ -29,8 +29,8 @@ cdef class SendInitialMetadataOperation(Operation): cdef grpc_metadata *_c_initial_metadata cdef size_t _c_initial_metadata_count - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class SendMessageOperation(Operation): @@ -39,16 +39,16 @@ cdef class SendMessageOperation(Operation): cdef readonly int _flags cdef grpc_byte_buffer *_c_message_byte_buffer - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class SendCloseFromClientOperation(Operation): cdef readonly int _flags - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class SendStatusFromServerOperation(Operation): @@ -61,8 +61,8 @@ cdef class SendStatusFromServerOperation(Operation): cdef size_t _c_trailing_metadata_count cdef grpc_slice _c_details - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class ReceiveInitialMetadataOperation(Operation): @@ -71,8 +71,8 @@ cdef class ReceiveInitialMetadataOperation(Operation): cdef tuple _initial_metadata cdef grpc_metadata_array _c_initial_metadata - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class ReceiveMessageOperation(Operation): @@ -81,8 +81,8 @@ cdef class ReceiveMessageOperation(Operation): cdef grpc_byte_buffer *_c_message_byte_buffer cdef bytes _message - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class ReceiveStatusOnClientOperation(Operation): @@ -97,8 +97,8 @@ cdef class ReceiveStatusOnClientOperation(Operation): cdef str _details cdef str _error_string - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * cdef class ReceiveCloseOnServerOperation(Operation): @@ -107,5 +107,5 @@ cdef class ReceiveCloseOnServerOperation(Operation): cdef object _cancelled cdef int _c_cancelled - cdef void c(self) - cdef void un_c(self) + cdef void c(self) except * + cdef void un_c(self) except * diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/operation.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/operation.pyx.pxi index 454627f570..c8a390106a 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/operation.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/operation.pyx.pxi @@ -15,10 +15,10 @@ cdef class Operation: - cdef void c(self): + cdef void c(self) except *: raise NotImplementedError() - cdef void un_c(self): + cdef void un_c(self) except *: raise NotImplementedError() @@ -31,7 +31,7 @@ cdef class SendInitialMetadataOperation(Operation): def type(self): return GRPC_OP_SEND_INITIAL_METADATA - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_SEND_INITIAL_METADATA self.c_op.flags = self._flags _store_c_metadata( @@ -41,7 +41,7 @@ cdef class SendInitialMetadataOperation(Operation): self.c_op.data.send_initial_metadata.count = self._c_initial_metadata_count self.c_op.data.send_initial_metadata.maybe_compression_level.is_set = 0 - cdef void un_c(self): + cdef void un_c(self) except *: _release_c_metadata( self._c_initial_metadata, self._c_initial_metadata_count) @@ -55,7 +55,7 @@ cdef class SendMessageOperation(Operation): def type(self): return GRPC_OP_SEND_MESSAGE - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_SEND_MESSAGE self.c_op.flags = self._flags cdef grpc_slice message_slice = grpc_slice_from_copied_buffer( @@ -65,7 +65,7 @@ cdef class SendMessageOperation(Operation): grpc_slice_unref(message_slice) self.c_op.data.send_message.send_message = self._c_message_byte_buffer - cdef void un_c(self): + cdef void un_c(self) except *: grpc_byte_buffer_destroy(self._c_message_byte_buffer) @@ -77,11 +77,11 @@ cdef class SendCloseFromClientOperation(Operation): def type(self): return GRPC_OP_SEND_CLOSE_FROM_CLIENT - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_SEND_CLOSE_FROM_CLIENT self.c_op.flags = self._flags - cdef void un_c(self): + cdef void un_c(self) except *: pass @@ -96,7 +96,7 @@ cdef class SendStatusFromServerOperation(Operation): def type(self): return GRPC_OP_SEND_STATUS_FROM_SERVER - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_SEND_STATUS_FROM_SERVER self.c_op.flags = self._flags _store_c_metadata( @@ -110,7 +110,7 @@ cdef class SendStatusFromServerOperation(Operation): self._c_details = _slice_from_bytes(_encode(self._details)) self.c_op.data.send_status_from_server.status_details = &self._c_details - cdef void un_c(self): + cdef void un_c(self) except *: grpc_slice_unref(self._c_details) _release_c_metadata( self._c_trailing_metadata, self._c_trailing_metadata_count) @@ -124,14 +124,14 @@ cdef class ReceiveInitialMetadataOperation(Operation): def type(self): return GRPC_OP_RECV_INITIAL_METADATA - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_RECV_INITIAL_METADATA self.c_op.flags = self._flags grpc_metadata_array_init(&self._c_initial_metadata) self.c_op.data.receive_initial_metadata.receive_initial_metadata = ( &self._c_initial_metadata) - cdef void un_c(self): + cdef void un_c(self) except *: self._initial_metadata = _metadata(&self._c_initial_metadata) grpc_metadata_array_destroy(&self._c_initial_metadata) @@ -147,13 +147,13 @@ cdef class ReceiveMessageOperation(Operation): def type(self): return GRPC_OP_RECV_MESSAGE - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_RECV_MESSAGE self.c_op.flags = self._flags self.c_op.data.receive_message.receive_message = ( &self._c_message_byte_buffer) - cdef void un_c(self): + cdef void un_c(self) except *: cdef grpc_byte_buffer_reader message_reader cdef bint message_reader_status cdef grpc_slice message_slice @@ -189,7 +189,7 @@ cdef class ReceiveStatusOnClientOperation(Operation): def type(self): return GRPC_OP_RECV_STATUS_ON_CLIENT - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_RECV_STATUS_ON_CLIENT self.c_op.flags = self._flags grpc_metadata_array_init(&self._c_trailing_metadata) @@ -202,7 +202,7 @@ cdef class ReceiveStatusOnClientOperation(Operation): self.c_op.data.receive_status_on_client.error_string = ( &self._c_error_string) - cdef void un_c(self): + cdef void un_c(self) except *: self._trailing_metadata = _metadata(&self._c_trailing_metadata) grpc_metadata_array_destroy(&self._c_trailing_metadata) self._code = self._c_code @@ -235,12 +235,12 @@ cdef class ReceiveCloseOnServerOperation(Operation): def type(self): return GRPC_OP_RECV_CLOSE_ON_SERVER - cdef void c(self): + cdef void c(self) except *: self.c_op.type = GRPC_OP_RECV_CLOSE_ON_SERVER self.c_op.flags = self._flags self.c_op.data.receive_close_on_server.cancelled = &self._c_cancelled - cdef void un_c(self): + cdef void un_c(self) except *: self._cancelled = bool(self._c_cancelled) def cancelled(self): diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/security.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/security.pyx.pxi index 7decae95bb..e17ca6d335 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/security.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/security.pyx.pxi @@ -14,14 +14,16 @@ from libc.string cimport memcpy -import pkg_resources +import pkgutil cdef grpc_ssl_roots_override_result ssl_roots_override_callback( char **pem_root_certs) nogil: with gil: - temporary_pem_root_certs = pkg_resources.resource_string( - __name__.rstrip('.cygrpc'), '_credentials/roots.pem') + pkg = __name__ + if pkg.endswith('.cygrpc'): + pkg = pkg[:-len('.cygrpc')] + temporary_pem_root_certs = pkgutil.get_data(pkg, '_credentials/roots.pem') pem_root_certs[0] = <char *>gpr_malloc(len(temporary_pem_root_certs) + 1) memcpy( pem_root_certs[0], <char *>temporary_pem_root_certs, diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index 5779437b92..ce701724fd 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -18,7 +18,6 @@ import logging import time import grpc -logging.basicConfig() _LOGGER = logging.getLogger(__name__) cdef class Server: diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi index f9a3b5e8f4..d8ba1ea9bd 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi @@ -32,7 +32,7 @@ cdef class _RequestCallTag(_Tag): cdef CallDetails call_details cdef grpc_metadata_array c_invocation_metadata - cdef void prepare(self) + cdef void prepare(self) except * cdef RequestCallEvent event(self, grpc_event c_event) @@ -44,7 +44,7 @@ cdef class _BatchOperationTag(_Tag): cdef grpc_op *c_ops cdef size_t c_nops - cdef void prepare(self) + cdef void prepare(self) except * cdef BatchOperationEvent event(self, grpc_event c_event) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi index aaca458442..be5013c8f7 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi @@ -35,7 +35,7 @@ cdef class _RequestCallTag(_Tag): self.call = None self.call_details = None - cdef void prepare(self): + cdef void prepare(self) except *: self.call = Call() self.call_details = CallDetails() grpc_metadata_array_init(&self.c_invocation_metadata) @@ -55,7 +55,7 @@ cdef class _BatchOperationTag: self._operations = operations self._retained_call = call - cdef void prepare(self): + cdef void prepare(self) except *: self.c_nops = 0 if self._operations is None else len(self._operations) if 0 < self.c_nops: self.c_ops = <grpc_op *>gpr_malloc(sizeof(grpc_op) * self.c_nops) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi index ce67c61eaf..1319ac0481 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi @@ -16,4 +16,4 @@ cdef gpr_timespec _timespec_from_time(object time) -cdef double _time_from_timespec(gpr_timespec timespec) +cdef double _time_from_timespec(gpr_timespec timespec) except * diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi index 7a668680b8..c452dd54f8 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi @@ -24,7 +24,7 @@ cdef gpr_timespec _timespec_from_time(object time): return timespec -cdef double _time_from_timespec(gpr_timespec timespec): +cdef double _time_from_timespec(gpr_timespec timespec) except *: cdef gpr_timespec real_timespec = gpr_convert_clock_type( timespec, GPR_CLOCK_REALTIME) return <double>real_timespec.seconds + <double>real_timespec.nanoseconds / 1e9 diff --git a/src/python/grpcio/grpc/_cython/cygrpc.pyx b/src/python/grpcio/grpc/_cython/cygrpc.pyx index 026f7ba2e3..9ab919375c 100644 --- a/src/python/grpcio/grpc/_cython/cygrpc.pyx +++ b/src/python/grpcio/grpc/_cython/cygrpc.pyx @@ -15,7 +15,6 @@ cimport cpython -import pkg_resources import os.path import sys @@ -36,6 +35,7 @@ include "_cygrpc/server.pyx.pxi" include "_cygrpc/tag.pyx.pxi" include "_cygrpc/time.pyx.pxi" include "_cygrpc/_hooks.pyx.pxi" +include "_cygrpc/channelz.pyx.pxi" include "_cygrpc/grpc_gevent.pyx.pxi" diff --git a/src/python/grpcio/grpc/_grpcio_metadata.py b/src/python/grpcio/grpc/_grpcio_metadata.py index 42b3a1ad49..7a9f173947 100644 --- a/src/python/grpcio/grpc/_grpcio_metadata.py +++ b/src/python/grpcio/grpc/_grpcio_metadata.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio/grpc/_grpcio_metadata.py.template`!!! -__version__ = """1.17.0.dev0""" +__version__ = """1.18.0.dev0""" diff --git a/src/python/grpcio/grpc/_interceptor.py b/src/python/grpcio/grpc/_interceptor.py index 1d2d374ad1..fc0ad77eb9 100644 --- a/src/python/grpcio/grpc/_interceptor.py +++ b/src/python/grpcio/grpc/_interceptor.py @@ -46,7 +46,7 @@ def service_pipeline(interceptors): class _ClientCallDetails( collections.namedtuple( '_ClientCallDetails', - ('method', 'timeout', 'metadata', 'credentials')), + ('method', 'timeout', 'metadata', 'credentials', 'wait_for_ready')), grpc.ClientCallDetails): pass @@ -72,7 +72,12 @@ def _unwrap_client_call_details(call_details, default_details): except AttributeError: credentials = default_details.credentials - return method, timeout, metadata, credentials + try: + wait_for_ready = call_details.wait_for_ready + except AttributeError: + wait_for_ready = default_details.wait_for_ready + + return method, timeout, metadata, credentials, wait_for_ready class _FailureOutcome(grpc.RpcError, grpc.Future, grpc.Call): @@ -130,9 +135,12 @@ class _FailureOutcome(grpc.RpcError, grpc.Future, grpc.Call): def __iter__(self): return self - def next(self): + def __next__(self): raise self._exception + def next(self): + return self.__next__() + class _UnaryOutcome(grpc.Call, grpc.Future): @@ -193,31 +201,42 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): self._method = method self._interceptor = interceptor - def __call__(self, request, timeout=None, metadata=None, credentials=None): + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): response, ignored_call = self._with_call( request, timeout=timeout, metadata=metadata, - credentials=credentials) + credentials=credentials, + wait_for_ready=wait_for_ready) return response - def _with_call(self, request, timeout=None, metadata=None, - credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + def _with_call(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) try: response, call = self._thunk(new_method).with_call( request, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) return _UnaryOutcome(response, call) - except grpc.RpcError: - raise + except grpc.RpcError as rpc_error: + return rpc_error except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) @@ -225,25 +244,37 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): continuation, client_call_details, request) return call.result(), call - def with_call(self, request, timeout=None, metadata=None, credentials=None): + def with_call(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): return self._with_call( request, timeout=timeout, metadata=metadata, - credentials=credentials) + credentials=credentials, + wait_for_ready=wait_for_ready) - def future(self, request, timeout=None, metadata=None, credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + def future(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) return self._thunk(new_method).future( request, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) try: return self._interceptor.intercept_unary_unary( @@ -259,18 +290,24 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): self._method = method self._interceptor = interceptor - def __call__(self, request, timeout=None, metadata=None, credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + def __call__(self, + request, + timeout=None, + metadata=None, + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) return self._thunk(new_method)( request, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) try: return self._interceptor.intercept_unary_stream( @@ -290,34 +327,38 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): response, ignored_call = self._with_call( request_iterator, timeout=timeout, metadata=metadata, - credentials=credentials) + credentials=credentials, + wait_for_ready=wait_for_ready) return response def _with_call(self, request_iterator, timeout=None, metadata=None, - credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request_iterator): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) try: response, call = self._thunk(new_method).with_call( request_iterator, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) return _UnaryOutcome(response, call) - except grpc.RpcError: - raise + except grpc.RpcError as rpc_error: + return rpc_error except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) @@ -329,29 +370,33 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): request_iterator, timeout=None, metadata=None, - credentials=None): + credentials=None, + wait_for_ready=None): return self._with_call( request_iterator, timeout=timeout, metadata=metadata, - credentials=credentials) + credentials=credentials, + wait_for_ready=wait_for_ready) def future(self, request_iterator, timeout=None, metadata=None, - credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request_iterator): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) return self._thunk(new_method).future( request_iterator, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) try: return self._interceptor.intercept_stream_unary( @@ -371,18 +416,20 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): request_iterator, timeout=None, metadata=None, - credentials=None): - client_call_details = _ClientCallDetails(self._method, timeout, - metadata, credentials) + credentials=None, + wait_for_ready=None): + client_call_details = _ClientCallDetails( + self._method, timeout, metadata, credentials, wait_for_ready) def continuation(new_details, request_iterator): - new_method, new_timeout, new_metadata, new_credentials = ( + new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready = ( _unwrap_client_call_details(new_details, client_call_details)) return self._thunk(new_method)( request_iterator, timeout=new_timeout, metadata=new_metadata, - credentials=new_credentials) + credentials=new_credentials, + wait_for_ready=new_wait_for_ready) try: return self._interceptor.intercept_stream_stream( diff --git a/src/python/grpcio/grpc/_plugin_wrapping.py b/src/python/grpcio/grpc/_plugin_wrapping.py index 88ab4d8371..916ee080b6 100644 --- a/src/python/grpcio/grpc/_plugin_wrapping.py +++ b/src/python/grpcio/grpc/_plugin_wrapping.py @@ -20,7 +20,6 @@ import grpc from grpc import _common from grpc._cython import cygrpc -logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/_server.py b/src/python/grpcio/grpc/_server.py index daa000a6e1..7276a7fd90 100644 --- a/src/python/grpcio/grpc/_server.py +++ b/src/python/grpcio/grpc/_server.py @@ -27,7 +27,6 @@ from grpc import _interceptor from grpc._cython import cygrpc from grpc.framework.foundation import callable_util -logging.basicConfig() _LOGGER = logging.getLogger(__name__) _SHUTDOWN_TAG = 'shutdown' diff --git a/src/python/grpcio/grpc/beta/BUILD.bazel b/src/python/grpcio/grpc/beta/BUILD.bazel deleted file mode 100644 index 731be5cb25..0000000000 --- a/src/python/grpcio/grpc/beta/BUILD.bazel +++ /dev/null @@ -1,58 +0,0 @@ -load("@grpc_python_dependencies//:requirements.bzl", "requirement") -package(default_visibility = ["//visibility:public"]) - -py_library( - name = "beta", - srcs = ["__init__.py",], - deps = [ - ":client_adaptations", - ":metadata", - ":server_adaptations", - ":implementations", - ":interfaces", - ":utilities", - ], -) - -py_library( - name = "client_adaptations", - srcs = ["_client_adaptations.py"], - imports=["../../",] -) - -py_library( - name = "metadata", - srcs = ["_metadata.py"], -) - -py_library( - name = "server_adaptations", - srcs = ["_server_adaptations.py"], - imports=["../../",], -) - -py_library( - name = "implementations", - srcs = ["implementations.py"], - imports=["../../",], -) - -py_library( - name = "interfaces", - srcs = ["interfaces.py"], - deps = [ - requirement("six"), - ], - imports=["../../",], -) - -py_library( - name = "utilities", - srcs = ["utilities.py"], - deps = [ - ":implementations", - ":interfaces", - "//src/python/grpcio/grpc/framework/foundation", - ], -) - diff --git a/src/python/grpcio/grpc/framework/foundation/callable_util.py b/src/python/grpcio/grpc/framework/foundation/callable_util.py index fb8d5f7c1e..24daf3406f 100644 --- a/src/python/grpcio/grpc/framework/foundation/callable_util.py +++ b/src/python/grpcio/grpc/framework/foundation/callable_util.py @@ -21,7 +21,6 @@ import logging import six -logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/framework/foundation/logging_pool.py b/src/python/grpcio/grpc/framework/foundation/logging_pool.py index 7702d1785f..216e3990db 100644 --- a/src/python/grpcio/grpc/framework/foundation/logging_pool.py +++ b/src/python/grpcio/grpc/framework/foundation/logging_pool.py @@ -17,7 +17,6 @@ import logging from concurrent import futures -logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/framework/foundation/stream_util.py b/src/python/grpcio/grpc/framework/foundation/stream_util.py index 9184f95873..1faaf29bd7 100644 --- a/src/python/grpcio/grpc/framework/foundation/stream_util.py +++ b/src/python/grpcio/grpc/framework/foundation/stream_util.py @@ -19,7 +19,6 @@ import threading from grpc.framework.foundation import stream _NO_VALUE = object() -logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index cb48b9f62c..ce65c594fe 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -208,12 +208,14 @@ CORE_SOURCE_FILES = [ 'src/core/lib/transport/timeout_encoding.cc', 'src/core/lib/transport/transport.cc', 'src/core/lib/transport/transport_op_string.cc', + 'src/core/lib/uri/uri_parser.cc', 'src/core/lib/debug/trace.cc', 'src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc', 'src/core/ext/transport/chttp2/transport/bin_decoder.cc', 'src/core/ext/transport/chttp2/transport/bin_encoder.cc', 'src/core/ext/transport/chttp2/transport/chttp2_plugin.cc', 'src/core/ext/transport/chttp2/transport/chttp2_transport.cc', + 'src/core/ext/transport/chttp2/transport/context_list.cc', 'src/core/ext/transport/chttp2/transport/flow_control.cc', 'src/core/ext/transport/chttp2/transport/frame_data.cc', 'src/core/ext/transport/chttp2/transport/frame_goaway.cc', @@ -280,7 +282,7 @@ CORE_SOURCE_FILES = [ 'src/core/tsi/alts/frame_protector/alts_unseal_privacy_integrity_crypter.cc', 'src/core/tsi/alts/frame_protector/frame_handler.cc', 'src/core/tsi/alts/handshaker/alts_handshaker_client.cc', - 'src/core/tsi/alts/handshaker/alts_tsi_event.cc', + 'src/core/tsi/alts/handshaker/alts_shared_resource.cc', 'src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc', 'src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_integrity_only_record_protocol.cc', 'src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc', @@ -322,19 +324,17 @@ CORE_SOURCE_FILES = [ 'src/core/ext/filters/client_channel/lb_policy.cc', 'src/core/ext/filters/client_channel/lb_policy_factory.cc', 'src/core/ext/filters/client_channel/lb_policy_registry.cc', - 'src/core/ext/filters/client_channel/method_params.cc', 'src/core/ext/filters/client_channel/parse_address.cc', 'src/core/ext/filters/client_channel/proxy_mapper.cc', 'src/core/ext/filters/client_channel/proxy_mapper_registry.cc', 'src/core/ext/filters/client_channel/resolver.cc', 'src/core/ext/filters/client_channel/resolver_registry.cc', + 'src/core/ext/filters/client_channel/resolver_result_parsing.cc', 'src/core/ext/filters/client_channel/retry_throttle.cc', 'src/core/ext/filters/client_channel/subchannel.cc', 'src/core/ext/filters/client_channel/subchannel_index.cc', - 'src/core/ext/filters/client_channel/uri_parser.cc', 'src/core/ext/filters/deadline/deadline_filter.cc', 'src/core/ext/filters/client_channel/health/health.pb.c', - 'src/core/tsi/alts_transport_security.cc', 'src/core/tsi/fake_transport_security.cc', 'src/core/tsi/local_transport_security.cc', 'src/core/tsi/ssl/session_cache/ssl_session_boringssl.cc', diff --git a/src/python/grpcio/grpc_version.py b/src/python/grpcio/grpc_version.py index 71113e68d9..2e91818d2c 100644 --- a/src/python/grpcio/grpc_version.py +++ b/src/python/grpcio/grpc_version.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio/grpc_version.py.template`!!! -VERSION = '1.17.0.dev0' +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_channelz/.gitignore b/src/python/grpcio_channelz/.gitignore new file mode 100644 index 0000000000..0c5da6b5af --- /dev/null +++ b/src/python/grpcio_channelz/.gitignore @@ -0,0 +1,6 @@ +*.proto +*_pb2.py +*_pb2_grpc.py +build/ +grpcio_channelz.egg-info/ +dist/ diff --git a/src/python/grpcio_channelz/MANIFEST.in b/src/python/grpcio_channelz/MANIFEST.in new file mode 100644 index 0000000000..ee93e21a69 --- /dev/null +++ b/src/python/grpcio_channelz/MANIFEST.in @@ -0,0 +1,4 @@ +include grpc_version.py +recursive-include grpc_channelz *.py +global-exclude *.pyc +include LICENSE diff --git a/src/python/grpcio_channelz/README.rst b/src/python/grpcio_channelz/README.rst new file mode 100644 index 0000000000..efeaa56064 --- /dev/null +++ b/src/python/grpcio_channelz/README.rst @@ -0,0 +1,9 @@ +gRPC Python Channelz package +============================== + +Channelz is a live debug tool in gRPC Python. + +Dependencies +------------ + +Depends on the `grpcio` package, available from PyPI via `pip install grpcio`. diff --git a/src/python/grpcio_channelz/channelz_commands.py b/src/python/grpcio_channelz/channelz_commands.py new file mode 100644 index 0000000000..7f158c2a4b --- /dev/null +++ b/src/python/grpcio_channelz/channelz_commands.py @@ -0,0 +1,67 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Provides distutils command classes for the GRPC Python setup process.""" + +import os +import shutil + +import setuptools + +ROOT_DIR = os.path.abspath(os.path.dirname(os.path.abspath(__file__))) +CHANNELZ_PROTO = os.path.join(ROOT_DIR, + '../../proto/grpc/channelz/channelz.proto') +LICENSE = os.path.join(ROOT_DIR, '../../../LICENSE') + + +class Preprocess(setuptools.Command): + """Command to copy proto modules from grpc/src/proto and LICENSE from + the root directory""" + + description = '' + user_options = [] + + def initialize_options(self): + pass + + def finalize_options(self): + pass + + def run(self): + if os.path.isfile(CHANNELZ_PROTO): + shutil.copyfile(CHANNELZ_PROTO, + os.path.join(ROOT_DIR, + 'grpc_channelz/v1/channelz.proto')) + if os.path.isfile(LICENSE): + shutil.copyfile(LICENSE, os.path.join(ROOT_DIR, 'LICENSE')) + + +class BuildPackageProtos(setuptools.Command): + """Command to generate project *_pb2.py modules from proto files.""" + + description = 'build grpc protobuf modules' + user_options = [] + + def initialize_options(self): + pass + + def finalize_options(self): + pass + + def run(self): + # due to limitations of the proto generator, we require that only *one* + # 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 + command.build_package_protos(self.distribution.package_dir['']) diff --git a/src/python/grpcio_channelz/grpc_channelz/__init__.py b/src/python/grpcio_channelz/grpc_channelz/__init__.py new file mode 100644 index 0000000000..38fdfc9c5c --- /dev/null +++ b/src/python/grpcio_channelz/grpc_channelz/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/src/python/grpcio_channelz/grpc_channelz/v1/BUILD.bazel b/src/python/grpcio_channelz/grpc_channelz/v1/BUILD.bazel new file mode 100644 index 0000000000..aae8cedb76 --- /dev/null +++ b/src/python/grpcio_channelz/grpc_channelz/v1/BUILD.bazel @@ -0,0 +1,38 @@ +load("@grpc_python_dependencies//:requirements.bzl", "requirement") +load("@org_pubref_rules_protobuf//python:rules.bzl", "py_proto_library") + +package(default_visibility = ["//visibility:public"]) + +genrule( + name = "mv_channelz_proto", + srcs = [ + "//src/proto/grpc/channelz:channelz_proto_file", + ], + outs = ["channelz.proto",], + cmd = "cp $< $@", +) + +py_proto_library( + name = "py_channelz_proto", + protos = ["mv_channelz_proto",], + imports = [ + "external/com_google_protobuf/src/", + ], + inputs = [ + "@com_google_protobuf//:well_known_protos", + ], + with_grpc = True, + deps = [ + requirement('protobuf'), + ], +) + +py_library( + name = "grpc_channelz", + srcs = ["channelz.py",], + deps = [ + ":py_channelz_proto", + "//src/python/grpcio/grpc:grpcio", + ], + imports=["../../",], +) diff --git a/src/python/grpcio_channelz/grpc_channelz/v1/__init__.py b/src/python/grpcio_channelz/grpc_channelz/v1/__init__.py new file mode 100644 index 0000000000..38fdfc9c5c --- /dev/null +++ b/src/python/grpcio_channelz/grpc_channelz/v1/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/src/python/grpcio_channelz/grpc_channelz/v1/channelz.py b/src/python/grpcio_channelz/grpc_channelz/v1/channelz.py new file mode 100644 index 0000000000..573b9d0d5a --- /dev/null +++ b/src/python/grpcio_channelz/grpc_channelz/v1/channelz.py @@ -0,0 +1,141 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Channelz debug service implementation in gRPC Python.""" + +import grpc +from grpc._cython import cygrpc + +import grpc_channelz.v1.channelz_pb2 as _channelz_pb2 +import grpc_channelz.v1.channelz_pb2_grpc as _channelz_pb2_grpc + +from google.protobuf import json_format + + +class ChannelzServicer(_channelz_pb2_grpc.ChannelzServicer): + """Servicer handling RPCs for service statuses.""" + + @staticmethod + def GetTopChannels(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_top_channels(request.start_channel_id), + _channelz_pb2.GetTopChannelsResponse(), + ) + except (ValueError, json_format.ParseError) as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetServers(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_servers(request.start_server_id), + _channelz_pb2.GetServersResponse(), + ) + except (ValueError, json_format.ParseError) as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetServer(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_server(request.server_id), + _channelz_pb2.GetServerResponse(), + ) + except ValueError as e: + context.set_code(grpc.StatusCode.NOT_FOUND) + context.set_details(str(e)) + except json_format.ParseError as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetServerSockets(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_server_sockets(request.server_id, + request.start_socket_id), + _channelz_pb2.GetServerSocketsResponse(), + ) + except ValueError as e: + context.set_code(grpc.StatusCode.NOT_FOUND) + context.set_details(str(e)) + except json_format.ParseError as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetChannel(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_channel(request.channel_id), + _channelz_pb2.GetChannelResponse(), + ) + except ValueError as e: + context.set_code(grpc.StatusCode.NOT_FOUND) + context.set_details(str(e)) + except json_format.ParseError as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetSubchannel(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_subchannel(request.subchannel_id), + _channelz_pb2.GetSubchannelResponse(), + ) + except ValueError as e: + context.set_code(grpc.StatusCode.NOT_FOUND) + context.set_details(str(e)) + except json_format.ParseError as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + @staticmethod + def GetSocket(request, context): + try: + return json_format.Parse( + cygrpc.channelz_get_socket(request.socket_id), + _channelz_pb2.GetSocketResponse(), + ) + except ValueError as e: + context.set_code(grpc.StatusCode.NOT_FOUND) + context.set_details(str(e)) + except json_format.ParseError as e: + context.set_code(grpc.StatusCode.INTERNAL) + context.set_details(str(e)) + + +def add_channelz_servicer(server): + """Add Channelz servicer to a server. Channelz servicer is in charge of + pulling information from C-Core for entire process. It will allow the + server to response to Channelz queries. + + The Channelz statistic is enabled by default inside C-Core. Whether the + statistic is enabled or not is isolated from adding Channelz servicer. + That means you can query Channelz info with a Channelz-disabled channel, + and you can add Channelz servicer to a Channelz-disabled server. + + The Channelz statistic can be enabled or disabled by channel option + 'grpc.enable_channelz'. Set to 1 to enable, set to 0 to disable. + + This is an EXPERIMENTAL API. + + Args: + server: grpc.Server to which Channelz service will be added. + """ + _channelz_pb2_grpc.add_ChannelzServicer_to_server(ChannelzServicer(), + server) diff --git a/src/python/grpcio_channelz/grpc_version.py b/src/python/grpcio_channelz/grpc_version.py new file mode 100644 index 0000000000..16356ea402 --- /dev/null +++ b/src/python/grpcio_channelz/grpc_version.py @@ -0,0 +1,17 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_channelz/grpc_version.py.template`!!! + +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_channelz/setup.py b/src/python/grpcio_channelz/setup.py new file mode 100644 index 0000000000..f8c0e93913 --- /dev/null +++ b/src/python/grpcio_channelz/setup.py @@ -0,0 +1,96 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Setup module for the GRPC Python package's Channelz.""" + +import os +import sys + +import setuptools + +# Ensure we're in the proper directory whether or not we're being used by pip. +os.chdir(os.path.dirname(os.path.abspath(__file__))) + +# Break import-style to ensure we can actually find our local modules. +import grpc_version + + +class _NoOpCommand(setuptools.Command): + """No-op command.""" + + description = '' + user_options = [] + + def initialize_options(self): + pass + + def finalize_options(self): + pass + + def run(self): + pass + + +CLASSIFIERS = [ + 'Development Status :: 5 - Production/Stable', + 'Programming Language :: Python', + 'Programming Language :: Python :: 2', + 'Programming Language :: Python :: 2.7', + 'Programming Language :: Python :: 3', + 'Programming Language :: Python :: 3.4', + 'Programming Language :: Python :: 3.5', + 'Programming Language :: Python :: 3.6', + 'License :: OSI Approved :: Apache Software License', +] + +PACKAGE_DIRECTORIES = { + '': '.', +} + +INSTALL_REQUIRES = ( + 'protobuf>=3.6.0', + 'grpcio>={version}'.format(version=grpc_version.VERSION), +) + +try: + import channelz_commands as _channelz_commands + # we are in the build environment, otherwise the above import fails + SETUP_REQUIRES = ( + 'grpcio-tools=={version}'.format(version=grpc_version.VERSION),) + COMMAND_CLASS = { + # Run preprocess from the repository *before* doing any packaging! + 'preprocess': _channelz_commands.Preprocess, + 'build_package_protos': _channelz_commands.BuildPackageProtos, + } +except ImportError: + SETUP_REQUIRES = () + COMMAND_CLASS = { + # wire up commands to no-op not to break the external dependencies + 'preprocess': _NoOpCommand, + 'build_package_protos': _NoOpCommand, + } + +setuptools.setup( + name='grpcio-channelz', + version=grpc_version.VERSION, + license='Apache License 2.0', + description='Channel Level Live Debug Information Service for gRPC', + author='The gRPC Authors', + author_email='grpc-io@googlegroups.com', + classifiers=CLASSIFIERS, + url='https://grpc.io', + package_dir=PACKAGE_DIRECTORIES, + packages=setuptools.find_packages('.'), + install_requires=INSTALL_REQUIRES, + setup_requires=SETUP_REQUIRES, + cmdclass=COMMAND_CLASS) diff --git a/src/python/grpcio_health_checking/MANIFEST.in b/src/python/grpcio_health_checking/MANIFEST.in index 996c74a9d4..3a22311b8e 100644 --- a/src/python/grpcio_health_checking/MANIFEST.in +++ b/src/python/grpcio_health_checking/MANIFEST.in @@ -1,3 +1,4 @@ include grpc_version.py recursive-include grpc_health *.py global-exclude *.pyc +include LICENSE diff --git a/src/python/grpcio_health_checking/grpc_version.py b/src/python/grpcio_health_checking/grpc_version.py index a30aac2e0b..85fa762f7e 100644 --- a/src/python/grpcio_health_checking/grpc_version.py +++ b/src/python/grpcio_health_checking/grpc_version.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_health_checking/grpc_version.py.template`!!! -VERSION = '1.17.0.dev0' +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_health_checking/health_commands.py b/src/python/grpcio_health_checking/health_commands.py index 933f965aa2..3820ef0bba 100644 --- a/src/python/grpcio_health_checking/health_commands.py +++ b/src/python/grpcio_health_checking/health_commands.py @@ -20,10 +20,12 @@ import setuptools ROOT_DIR = os.path.abspath(os.path.dirname(os.path.abspath(__file__))) HEALTH_PROTO = os.path.join(ROOT_DIR, '../../proto/grpc/health/v1/health.proto') +LICENSE = os.path.join(ROOT_DIR, '../../../LICENSE') -class CopyProtoModules(setuptools.Command): - """Command to copy proto modules from grpc/src/proto.""" +class Preprocess(setuptools.Command): + """Command to copy proto modules from grpc/src/proto and LICENSE from + the root directory""" description = '' user_options = [] @@ -39,6 +41,8 @@ class CopyProtoModules(setuptools.Command): shutil.copyfile(HEALTH_PROTO, os.path.join(ROOT_DIR, 'grpc_health/v1/health.proto')) + if os.path.isfile(LICENSE): + shutil.copyfile(LICENSE, os.path.join(ROOT_DIR, 'LICENSE')) class BuildPackageProtos(setuptools.Command): diff --git a/src/python/grpcio_health_checking/setup.py b/src/python/grpcio_health_checking/setup.py index db2edae2ce..5a09a80f6a 100644 --- a/src/python/grpcio_health_checking/setup.py +++ b/src/python/grpcio_health_checking/setup.py @@ -68,7 +68,7 @@ try: 'grpcio-tools=={version}'.format(version=grpc_version.VERSION),) COMMAND_CLASS = { # Run preprocess from the repository *before* doing any packaging! - 'preprocess': _health_commands.CopyProtoModules, + 'preprocess': _health_commands.Preprocess, 'build_package_protos': _health_commands.BuildPackageProtos, } except ImportError: diff --git a/src/python/grpcio_reflection/MANIFEST.in b/src/python/grpcio_reflection/MANIFEST.in index d6fb6ce73a..10b01fa41d 100644 --- a/src/python/grpcio_reflection/MANIFEST.in +++ b/src/python/grpcio_reflection/MANIFEST.in @@ -1,3 +1,4 @@ include grpc_version.py recursive-include grpc_reflection *.py global-exclude *.pyc +include LICENSE diff --git a/src/python/grpcio_reflection/grpc_reflection/v1alpha/BUILD.bazel b/src/python/grpcio_reflection/grpc_reflection/v1alpha/BUILD.bazel new file mode 100644 index 0000000000..3a2ba26371 --- /dev/null +++ b/src/python/grpcio_reflection/grpc_reflection/v1alpha/BUILD.bazel @@ -0,0 +1,34 @@ +load("@grpc_python_dependencies//:requirements.bzl", "requirement") +load("@org_pubref_rules_protobuf//python:rules.bzl", "py_proto_library") + +package(default_visibility = ["//visibility:public"]) + +genrule( + name = "mv_reflection_proto", + srcs = [ + "//src/proto/grpc/reflection/v1alpha:reflection_proto_file", + ], + outs = ["reflection.proto",], + cmd = "cp $< $@", +) + +py_proto_library( + name = "py_reflection_proto", + protos = [":mv_reflection_proto",], + with_grpc = True, + deps = [ + requirement('protobuf'), + ], +) + +py_library( + name = "grpc_reflection", + srcs = ["reflection.py",], + deps = [ + ":py_reflection_proto", + "//src/python/grpcio/grpc:grpcio", + requirement('protobuf'), + ], + imports=["../../",], +) + diff --git a/src/python/grpcio_reflection/grpc_version.py b/src/python/grpcio_reflection/grpc_version.py index aafea9fe76..e62ab169a2 100644 --- a/src/python/grpcio_reflection/grpc_version.py +++ b/src/python/grpcio_reflection/grpc_version.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_reflection/grpc_version.py.template`!!! -VERSION = '1.17.0.dev0' +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_reflection/reflection_commands.py b/src/python/grpcio_reflection/reflection_commands.py index 6f91f6b875..311ca4c4db 100644 --- a/src/python/grpcio_reflection/reflection_commands.py +++ b/src/python/grpcio_reflection/reflection_commands.py @@ -21,10 +21,12 @@ import setuptools ROOT_DIR = os.path.abspath(os.path.dirname(os.path.abspath(__file__))) REFLECTION_PROTO = os.path.join( ROOT_DIR, '../../proto/grpc/reflection/v1alpha/reflection.proto') +LICENSE = os.path.join(ROOT_DIR, '../../../LICENSE') -class CopyProtoModules(setuptools.Command): - """Command to copy proto modules from grpc/src/proto.""" +class Preprocess(setuptools.Command): + """Command to copy proto modules from grpc/src/proto and LICENSE from + the root directory""" description = '' user_options = [] @@ -41,6 +43,8 @@ class CopyProtoModules(setuptools.Command): REFLECTION_PROTO, os.path.join(ROOT_DIR, 'grpc_reflection/v1alpha/reflection.proto')) + if os.path.isfile(LICENSE): + shutil.copyfile(LICENSE, os.path.join(ROOT_DIR, 'LICENSE')) class BuildPackageProtos(setuptools.Command): diff --git a/src/python/grpcio_reflection/setup.py b/src/python/grpcio_reflection/setup.py index b4087d87b4..f205069acd 100644 --- a/src/python/grpcio_reflection/setup.py +++ b/src/python/grpcio_reflection/setup.py @@ -69,7 +69,7 @@ try: 'grpcio-tools=={version}'.format(version=grpc_version.VERSION),) COMMAND_CLASS = { # Run preprocess from the repository *before* doing any packaging! - 'preprocess': _reflection_commands.CopyProtoModules, + 'preprocess': _reflection_commands.Preprocess, 'build_package_protos': _reflection_commands.BuildPackageProtos, } except ImportError: diff --git a/src/python/grpcio_testing/MANIFEST.in b/src/python/grpcio_testing/MANIFEST.in index 39b3565217..559dfaf786 100644 --- a/src/python/grpcio_testing/MANIFEST.in +++ b/src/python/grpcio_testing/MANIFEST.in @@ -1,3 +1,4 @@ include grpc_version.py recursive-include grpc_testing *.py global-exclude *.pyc +include LICENSE diff --git a/src/python/grpcio_testing/grpc_version.py b/src/python/grpcio_testing/grpc_version.py index 876acd3142..7b4c1695fa 100644 --- a/src/python/grpcio_testing/grpc_version.py +++ b/src/python/grpcio_testing/grpc_version.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_testing/grpc_version.py.template`!!! -VERSION = '1.17.0.dev0' +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_testing/setup.py b/src/python/grpcio_testing/setup.py index 6ceb1fc5c9..18db71e0f0 100644 --- a/src/python/grpcio_testing/setup.py +++ b/src/python/grpcio_testing/setup.py @@ -24,6 +24,23 @@ os.chdir(os.path.dirname(os.path.abspath(__file__))) # Break import style to ensure that we can find same-directory modules. import grpc_version + +class _NoOpCommand(setuptools.Command): + """No-op command.""" + + description = '' + user_options = [] + + def initialize_options(self): + pass + + def finalize_options(self): + pass + + def run(self): + pass + + PACKAGE_DIRECTORIES = { '': '.', } @@ -33,6 +50,19 @@ INSTALL_REQUIRES = ( 'grpcio>={version}'.format(version=grpc_version.VERSION), ) +try: + import testing_commands as _testing_commands + # we are in the build environment, otherwise the above import fails + COMMAND_CLASS = { + # Run preprocess from the repository *before* doing any packaging! + 'preprocess': _testing_commands.Preprocess, + } +except ImportError: + COMMAND_CLASS = { + # wire up commands to no-op not to break the external dependencies + 'preprocess': _NoOpCommand, + } + setuptools.setup( name='grpcio-testing', version=grpc_version.VERSION, @@ -43,4 +73,5 @@ setuptools.setup( url='https://grpc.io', package_dir=PACKAGE_DIRECTORIES, packages=setuptools.find_packages('.'), - install_requires=INSTALL_REQUIRES) + install_requires=INSTALL_REQUIRES, + cmdclass=COMMAND_CLASS) diff --git a/src/python/grpcio_testing/testing_commands.py b/src/python/grpcio_testing/testing_commands.py new file mode 100644 index 0000000000..fb40d37efb --- /dev/null +++ b/src/python/grpcio_testing/testing_commands.py @@ -0,0 +1,39 @@ +# Copyright 2018 gRPC Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Provides distutils command classes for the GRPC Python setup process.""" + +import os +import shutil + +import setuptools + +ROOT_DIR = os.path.abspath(os.path.dirname(os.path.abspath(__file__))) +LICENSE = os.path.join(ROOT_DIR, '../../../LICENSE') + + +class Preprocess(setuptools.Command): + """Command to copy LICENSE from root directory.""" + + description = '' + user_options = [] + + def initialize_options(self): + pass + + def finalize_options(self): + pass + + def run(self): + if os.path.isfile(LICENSE): + shutil.copyfile(LICENSE, os.path.join(ROOT_DIR, 'LICENSE')) diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py index 6931d93ef0..65e9a99950 100644 --- a/src/python/grpcio_tests/commands.py +++ b/src/python/grpcio_tests/commands.py @@ -130,6 +130,13 @@ class TestGevent(setuptools.Command): # Beta API is unsupported for gevent 'protoc_plugin.beta_python_plugin_test', 'unit.beta._beta_features_test', + # TODO(https://github.com/grpc/grpc/issues/15411) unpin gevent version + # This test will stuck while running higher version of gevent + 'unit._auth_context_test.AuthContextTest.testSessionResumption', + # TODO(https://github.com/grpc/grpc/issues/17330) enable these three tests + 'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels', + 'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels_and_sockets', + 'channelz._channelz_servicer_test.ChannelzServicerTest.test_streaming_rpc' ) description = 'run tests with gevent. Assumes grpc/gevent are installed' user_options = [] diff --git a/src/python/grpcio_tests/grpc_version.py b/src/python/grpcio_tests/grpc_version.py index cc9b41587c..2fcd1ad617 100644 --- a/src/python/grpcio_tests/grpc_version.py +++ b/src/python/grpcio_tests/grpc_version.py @@ -14,4 +14,4 @@ # AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_tests/grpc_version.py.template`!!! -VERSION = '1.17.0.dev0' +VERSION = '1.18.0.dev0' diff --git a/src/python/grpcio_tests/setup.py b/src/python/grpcio_tests/setup.py index 61c98fa038..f56425ac6d 100644 --- a/src/python/grpcio_tests/setup.py +++ b/src/python/grpcio_tests/setup.py @@ -39,6 +39,7 @@ PACKAGE_DIRECTORIES = { INSTALL_REQUIRES = ( 'coverage>=4.0', 'enum34>=1.0.4', 'grpcio>={version}'.format(version=grpc_version.VERSION), + 'grpcio-channelz>={version}'.format(version=grpc_version.VERSION), 'grpcio-tools>={version}'.format(version=grpc_version.VERSION), 'grpcio-health-checking>={version}'.format(version=grpc_version.VERSION), 'oauth2client>=1.4.7', 'protobuf>=3.6.0', 'six>=1.10', 'google-auth>=1.0.0', diff --git a/src/python/grpcio_tests/tests/_sanity/_sanity_test.py b/src/python/grpcio_tests/tests/_sanity/_sanity_test.py index b4079850ff..7da6e7b34c 100644 --- a/src/python/grpcio_tests/tests/_sanity/_sanity_test.py +++ b/src/python/grpcio_tests/tests/_sanity/_sanity_test.py @@ -13,9 +13,9 @@ # limitations under the License. import json +import pkgutil import unittest -import pkg_resources import six import tests @@ -35,7 +35,7 @@ class SanityTest(unittest.TestCase): loader.suite) }) - tests_json_string = pkg_resources.resource_string('tests', 'tests.json') + tests_json_string = pkgutil.get_data('tests', 'tests.json') tests_json = json.loads(tests_json_string.decode() if six.PY3 else tests_json_string) diff --git a/src/python/grpcio_tests/tests/channelz/BUILD.bazel b/src/python/grpcio_tests/tests/channelz/BUILD.bazel new file mode 100644 index 0000000000..63513616e7 --- /dev/null +++ b/src/python/grpcio_tests/tests/channelz/BUILD.bazel @@ -0,0 +1,15 @@ +package(default_visibility = ["//visibility:public"]) + +py_test( + name = "channelz_servicer_test", + srcs = ["_channelz_servicer_test.py"], + main = "_channelz_servicer_test.py", + size = "small", + deps = [ + "//src/python/grpcio/grpc:grpcio", + "//src/python/grpcio_channelz/grpc_channelz/v1:grpc_channelz", + "//src/python/grpcio_tests/tests/unit:test_common", + "//src/python/grpcio_tests/tests/unit/framework/common:common", + ], + imports = ["../../",], +) diff --git a/src/python/grpcio_tests/tests/channelz/__init__.py b/src/python/grpcio_tests/tests/channelz/__init__.py new file mode 100644 index 0000000000..38fdfc9c5c --- /dev/null +++ b/src/python/grpcio_tests/tests/channelz/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/src/python/grpcio_tests/tests/channelz/_channelz_servicer_test.py b/src/python/grpcio_tests/tests/channelz/_channelz_servicer_test.py new file mode 100644 index 0000000000..8ca5189522 --- /dev/null +++ b/src/python/grpcio_tests/tests/channelz/_channelz_servicer_test.py @@ -0,0 +1,470 @@ +# Copyright 2018 The gRPC Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests of grpc_channelz.v1.channelz.""" + +import unittest + +from concurrent import futures + +import grpc +from grpc_channelz.v1 import channelz +from grpc_channelz.v1 import channelz_pb2 +from grpc_channelz.v1 import channelz_pb2_grpc + +from tests.unit import test_common +from tests.unit.framework.common import test_constants + +_SUCCESSFUL_UNARY_UNARY = '/test/SuccessfulUnaryUnary' +_FAILED_UNARY_UNARY = '/test/FailedUnaryUnary' +_SUCCESSFUL_STREAM_STREAM = '/test/SuccessfulStreamStream' + +_REQUEST = b'\x00\x00\x00' +_RESPONSE = b'\x01\x01\x01' + +_DISABLE_REUSE_PORT = (('grpc.so_reuseport', 0),) +_ENABLE_CHANNELZ = (('grpc.enable_channelz', 1),) +_DISABLE_CHANNELZ = (('grpc.enable_channelz', 0),) + + +def _successful_unary_unary(request, servicer_context): + return _RESPONSE + + +def _failed_unary_unary(request, servicer_context): + servicer_context.set_code(grpc.StatusCode.INTERNAL) + servicer_context.set_details("Channelz Test Intended Failure") + + +def _successful_stream_stream(request_iterator, servicer_context): + for _ in request_iterator: + yield _RESPONSE + + +class _GenericHandler(grpc.GenericRpcHandler): + + def service(self, handler_call_details): + if handler_call_details.method == _SUCCESSFUL_UNARY_UNARY: + return grpc.unary_unary_rpc_method_handler(_successful_unary_unary) + elif handler_call_details.method == _FAILED_UNARY_UNARY: + return grpc.unary_unary_rpc_method_handler(_failed_unary_unary) + elif handler_call_details.method == _SUCCESSFUL_STREAM_STREAM: + return grpc.stream_stream_rpc_method_handler( + _successful_stream_stream) + else: + return None + + +class _ChannelServerPair(object): + + def __init__(self): + # Server will enable channelz service + self.server = grpc.server( + futures.ThreadPoolExecutor(max_workers=3), + options=_DISABLE_REUSE_PORT + _ENABLE_CHANNELZ) + port = self.server.add_insecure_port('[::]:0') + self.server.add_generic_rpc_handlers((_GenericHandler(),)) + self.server.start() + + # Channel will enable channelz service... + self.channel = grpc.insecure_channel('localhost:%d' % port, + _ENABLE_CHANNELZ) + + +def _generate_channel_server_pairs(n): + return [_ChannelServerPair() for i in range(n)] + + +def _close_channel_server_pairs(pairs): + for pair in pairs: + pair.server.stop(None) + # TODO(ericgribkoff) This del should not be required + del pair.server + pair.channel.close() + + +class ChannelzServicerTest(unittest.TestCase): + + def _send_successful_unary_unary(self, idx): + _, r = self._pairs[idx].channel.unary_unary( + _SUCCESSFUL_UNARY_UNARY).with_call(_REQUEST) + self.assertEqual(r.code(), grpc.StatusCode.OK) + + def _send_failed_unary_unary(self, idx): + try: + self._pairs[idx].channel.unary_unary(_FAILED_UNARY_UNARY).with_call( + _REQUEST) + except grpc.RpcError: + return + else: + self.fail("This call supposed to fail") + + def _send_successful_stream_stream(self, idx): + response_iterator = self._pairs[idx].channel.stream_stream( + _SUCCESSFUL_STREAM_STREAM).__call__( + iter([_REQUEST] * test_constants.STREAM_LENGTH)) + cnt = 0 + for _ in response_iterator: + cnt += 1 + self.assertEqual(cnt, test_constants.STREAM_LENGTH) + + def _get_channel_id(self, idx): + """Channel id may not be consecutive""" + resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=0)) + self.assertGreater(len(resp.channel), idx) + return resp.channel[idx].ref.channel_id + + def setUp(self): + self._pairs = [] + # This server is for Channelz info fetching only + # It self should not enable Channelz + self._server = grpc.server( + futures.ThreadPoolExecutor(max_workers=3), + options=_DISABLE_REUSE_PORT + _DISABLE_CHANNELZ) + port = self._server.add_insecure_port('[::]:0') + channelz.add_channelz_servicer(self._server) + self._server.start() + + # This channel is used to fetch Channelz info only + # Channelz should not be enabled + self._channel = grpc.insecure_channel('localhost:%d' % port, + _DISABLE_CHANNELZ) + self._channelz_stub = channelz_pb2_grpc.ChannelzStub(self._channel) + + def tearDown(self): + self._server.stop(None) + self._channel.close() + _close_channel_server_pairs(self._pairs) + + def test_get_top_channels_basic(self): + self._pairs = _generate_channel_server_pairs(1) + resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=0)) + self.assertEqual(len(resp.channel), 1) + self.assertEqual(resp.end, True) + + def test_get_top_channels_high_start_id(self): + self._pairs = _generate_channel_server_pairs(1) + resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=10000)) + self.assertEqual(len(resp.channel), 0) + self.assertEqual(resp.end, True) + + def test_successful_request(self): + self._pairs = _generate_channel_server_pairs(1) + self._send_successful_unary_unary(0) + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(0))) + self.assertEqual(resp.channel.data.calls_started, 1) + self.assertEqual(resp.channel.data.calls_succeeded, 1) + self.assertEqual(resp.channel.data.calls_failed, 0) + + def test_failed_request(self): + self._pairs = _generate_channel_server_pairs(1) + self._send_failed_unary_unary(0) + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(0))) + self.assertEqual(resp.channel.data.calls_started, 1) + self.assertEqual(resp.channel.data.calls_succeeded, 0) + self.assertEqual(resp.channel.data.calls_failed, 1) + + def test_many_requests(self): + self._pairs = _generate_channel_server_pairs(1) + k_success = 7 + k_failed = 9 + for i in range(k_success): + self._send_successful_unary_unary(0) + for i in range(k_failed): + self._send_failed_unary_unary(0) + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(0))) + self.assertEqual(resp.channel.data.calls_started, k_success + k_failed) + self.assertEqual(resp.channel.data.calls_succeeded, k_success) + self.assertEqual(resp.channel.data.calls_failed, k_failed) + + def test_many_channel(self): + k_channels = 4 + self._pairs = _generate_channel_server_pairs(k_channels) + resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=0)) + self.assertEqual(len(resp.channel), k_channels) + + def test_many_requests_many_channel(self): + k_channels = 4 + self._pairs = _generate_channel_server_pairs(k_channels) + k_success = 11 + k_failed = 13 + for i in range(k_success): + self._send_successful_unary_unary(0) + self._send_successful_unary_unary(2) + for i in range(k_failed): + self._send_failed_unary_unary(1) + self._send_failed_unary_unary(2) + + # The first channel saw only successes + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(0))) + self.assertEqual(resp.channel.data.calls_started, k_success) + self.assertEqual(resp.channel.data.calls_succeeded, k_success) + self.assertEqual(resp.channel.data.calls_failed, 0) + + # The second channel saw only failures + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(1))) + self.assertEqual(resp.channel.data.calls_started, k_failed) + self.assertEqual(resp.channel.data.calls_succeeded, 0) + self.assertEqual(resp.channel.data.calls_failed, k_failed) + + # The third channel saw both successes and failures + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(2))) + self.assertEqual(resp.channel.data.calls_started, k_success + k_failed) + self.assertEqual(resp.channel.data.calls_succeeded, k_success) + self.assertEqual(resp.channel.data.calls_failed, k_failed) + + # The fourth channel saw nothing + resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(3))) + self.assertEqual(resp.channel.data.calls_started, 0) + self.assertEqual(resp.channel.data.calls_succeeded, 0) + self.assertEqual(resp.channel.data.calls_failed, 0) + + def test_many_subchannels(self): + k_channels = 4 + self._pairs = _generate_channel_server_pairs(k_channels) + k_success = 17 + k_failed = 19 + for i in range(k_success): + self._send_successful_unary_unary(0) + self._send_successful_unary_unary(2) + for i in range(k_failed): + self._send_failed_unary_unary(1) + self._send_failed_unary_unary(2) + + gtc_resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=0)) + self.assertEqual(len(gtc_resp.channel), k_channels) + for i in range(k_channels): + # If no call performed in the channel, there shouldn't be any subchannel + if gtc_resp.channel[i].data.calls_started == 0: + self.assertEqual(len(gtc_resp.channel[i].subchannel_ref), 0) + continue + + # Otherwise, the subchannel should exist + self.assertGreater(len(gtc_resp.channel[i].subchannel_ref), 0) + gsc_resp = self._channelz_stub.GetSubchannel( + channelz_pb2.GetSubchannelRequest( + subchannel_id=gtc_resp.channel[i].subchannel_ref[ + 0].subchannel_id)) + self.assertEqual(gtc_resp.channel[i].data.calls_started, + gsc_resp.subchannel.data.calls_started) + self.assertEqual(gtc_resp.channel[i].data.calls_succeeded, + gsc_resp.subchannel.data.calls_succeeded) + self.assertEqual(gtc_resp.channel[i].data.calls_failed, + gsc_resp.subchannel.data.calls_failed) + + def test_server_basic(self): + self._pairs = _generate_channel_server_pairs(1) + resp = self._channelz_stub.GetServers( + channelz_pb2.GetServersRequest(start_server_id=0)) + self.assertEqual(len(resp.server), 1) + + def test_get_one_server(self): + self._pairs = _generate_channel_server_pairs(1) + gss_resp = self._channelz_stub.GetServers( + channelz_pb2.GetServersRequest(start_server_id=0)) + self.assertEqual(len(gss_resp.server), 1) + gs_resp = self._channelz_stub.GetServer( + channelz_pb2.GetServerRequest( + server_id=gss_resp.server[0].ref.server_id)) + self.assertEqual(gss_resp.server[0].ref.server_id, + gs_resp.server.ref.server_id) + + def test_server_call(self): + self._pairs = _generate_channel_server_pairs(1) + k_success = 23 + k_failed = 29 + for i in range(k_success): + self._send_successful_unary_unary(0) + for i in range(k_failed): + self._send_failed_unary_unary(0) + + resp = self._channelz_stub.GetServers( + channelz_pb2.GetServersRequest(start_server_id=0)) + self.assertEqual(len(resp.server), 1) + self.assertEqual(resp.server[0].data.calls_started, + k_success + k_failed) + self.assertEqual(resp.server[0].data.calls_succeeded, k_success) + self.assertEqual(resp.server[0].data.calls_failed, k_failed) + + def test_many_subchannels_and_sockets(self): + k_channels = 4 + self._pairs = _generate_channel_server_pairs(k_channels) + k_success = 3 + k_failed = 5 + for i in range(k_success): + self._send_successful_unary_unary(0) + self._send_successful_unary_unary(2) + for i in range(k_failed): + self._send_failed_unary_unary(1) + self._send_failed_unary_unary(2) + + gtc_resp = self._channelz_stub.GetTopChannels( + channelz_pb2.GetTopChannelsRequest(start_channel_id=0)) + self.assertEqual(len(gtc_resp.channel), k_channels) + for i in range(k_channels): + # If no call performed in the channel, there shouldn't be any subchannel + if gtc_resp.channel[i].data.calls_started == 0: + self.assertEqual(len(gtc_resp.channel[i].subchannel_ref), 0) + continue + + # Otherwise, the subchannel should exist + self.assertGreater(len(gtc_resp.channel[i].subchannel_ref), 0) + gsc_resp = self._channelz_stub.GetSubchannel( + channelz_pb2.GetSubchannelRequest( + subchannel_id=gtc_resp.channel[i].subchannel_ref[ + 0].subchannel_id)) + self.assertEqual(len(gsc_resp.subchannel.socket_ref), 1) + + gs_resp = self._channelz_stub.GetSocket( + channelz_pb2.GetSocketRequest( + socket_id=gsc_resp.subchannel.socket_ref[0].socket_id)) + self.assertEqual(gsc_resp.subchannel.data.calls_started, + gs_resp.socket.data.streams_started) + self.assertEqual(gsc_resp.subchannel.data.calls_started, + gs_resp.socket.data.streams_succeeded) + # Calls started == messages sent, only valid for unary calls + self.assertEqual(gsc_resp.subchannel.data.calls_started, + gs_resp.socket.data.messages_sent) + # Only receive responses when the RPC was successful + self.assertEqual(gsc_resp.subchannel.data.calls_succeeded, + gs_resp.socket.data.messages_received) + + def test_streaming_rpc(self): + self._pairs = _generate_channel_server_pairs(1) + # In C++, the argument for _send_successful_stream_stream is message length. + # Here the argument is still channel idx, to be consistent with the other two. + self._send_successful_stream_stream(0) + + gc_resp = self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=self._get_channel_id(0))) + self.assertEqual(gc_resp.channel.data.calls_started, 1) + self.assertEqual(gc_resp.channel.data.calls_succeeded, 1) + self.assertEqual(gc_resp.channel.data.calls_failed, 0) + # Subchannel exists + self.assertGreater(len(gc_resp.channel.subchannel_ref), 0) + + gsc_resp = self._channelz_stub.GetSubchannel( + channelz_pb2.GetSubchannelRequest( + subchannel_id=gc_resp.channel.subchannel_ref[0].subchannel_id)) + self.assertEqual(gsc_resp.subchannel.data.calls_started, 1) + self.assertEqual(gsc_resp.subchannel.data.calls_succeeded, 1) + self.assertEqual(gsc_resp.subchannel.data.calls_failed, 0) + # Socket exists + self.assertEqual(len(gsc_resp.subchannel.socket_ref), 1) + + gs_resp = self._channelz_stub.GetSocket( + channelz_pb2.GetSocketRequest( + socket_id=gsc_resp.subchannel.socket_ref[0].socket_id)) + self.assertEqual(gs_resp.socket.data.streams_started, 1) + self.assertEqual(gs_resp.socket.data.streams_succeeded, 1) + self.assertEqual(gs_resp.socket.data.streams_failed, 0) + self.assertEqual(gs_resp.socket.data.messages_sent, + test_constants.STREAM_LENGTH) + self.assertEqual(gs_resp.socket.data.messages_received, + test_constants.STREAM_LENGTH) + + def test_server_sockets(self): + self._pairs = _generate_channel_server_pairs(1) + self._send_successful_unary_unary(0) + self._send_failed_unary_unary(0) + + gs_resp = self._channelz_stub.GetServers( + channelz_pb2.GetServersRequest(start_server_id=0)) + self.assertEqual(len(gs_resp.server), 1) + self.assertEqual(gs_resp.server[0].data.calls_started, 2) + self.assertEqual(gs_resp.server[0].data.calls_succeeded, 1) + self.assertEqual(gs_resp.server[0].data.calls_failed, 1) + + gss_resp = self._channelz_stub.GetServerSockets( + channelz_pb2.GetServerSocketsRequest( + server_id=gs_resp.server[0].ref.server_id, start_socket_id=0)) + # If the RPC call failed, it will raise a grpc.RpcError + # So, if there is no exception raised, considered pass + + def test_server_listen_sockets(self): + self._pairs = _generate_channel_server_pairs(1) + + gss_resp = self._channelz_stub.GetServers( + channelz_pb2.GetServersRequest(start_server_id=0)) + self.assertEqual(len(gss_resp.server), 1) + self.assertEqual(len(gss_resp.server[0].listen_socket), 1) + + gs_resp = self._channelz_stub.GetSocket( + channelz_pb2.GetSocketRequest( + socket_id=gss_resp.server[0].listen_socket[0].socket_id)) + # If the RPC call failed, it will raise a grpc.RpcError + # So, if there is no exception raised, considered pass + + def test_invalid_query_get_server(self): + try: + self._channelz_stub.GetServer( + channelz_pb2.GetServerRequest(server_id=10000)) + except BaseException as e: + self.assertIn('StatusCode.NOT_FOUND', str(e)) + else: + self.fail('Invalid query not detected') + + def test_invalid_query_get_channel(self): + try: + self._channelz_stub.GetChannel( + channelz_pb2.GetChannelRequest(channel_id=10000)) + except BaseException as e: + self.assertIn('StatusCode.NOT_FOUND', str(e)) + else: + self.fail('Invalid query not detected') + + def test_invalid_query_get_subchannel(self): + try: + self._channelz_stub.GetSubchannel( + channelz_pb2.GetSubchannelRequest(subchannel_id=10000)) + except BaseException as e: + self.assertIn('StatusCode.NOT_FOUND', str(e)) + else: + self.fail('Invalid query not detected') + + def test_invalid_query_get_socket(self): + try: + self._channelz_stub.GetSocket( + channelz_pb2.GetSocketRequest(socket_id=10000)) + except BaseException as e: + self.assertIn('StatusCode.NOT_FOUND', str(e)) + else: + self.fail('Invalid query not detected') + + def test_invalid_query_get_server_sockets(self): + try: + self._channelz_stub.GetServerSockets( + channelz_pb2.GetServerSocketsRequest( + server_id=10000, + start_socket_id=0, + )) + except BaseException as e: + self.assertIn('StatusCode.NOT_FOUND', str(e)) + else: + self.fail('Invalid query not detected') + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/interop/BUILD.bazel b/src/python/grpcio_tests/tests/interop/BUILD.bazel new file mode 100644 index 0000000000..aebdbf67eb --- /dev/null +++ b/src/python/grpcio_tests/tests/interop/BUILD.bazel @@ -0,0 +1,101 @@ +load("@grpc_python_dependencies//:requirements.bzl", "requirement") + +package(default_visibility = ["//visibility:public"]) + +py_library( + name = "_intraop_test_case", + srcs = ["_intraop_test_case.py"], + deps = [ + ":methods", + ], + imports=["../../",], +) + +py_library( + name = "client", + srcs = ["client.py"], + deps = [ + "//src/python/grpcio/grpc:grpcio", + ":methods", + ":resources", + "//src/proto/grpc/testing:py_test_proto", + requirement('google-auth'), + ], + imports=["../../",], +) + +py_library( + name = "methods", + srcs = ["methods.py"], + deps = [ + "//src/python/grpcio/grpc:grpcio", + "//src/proto/grpc/testing:py_empty_proto", + "//src/proto/grpc/testing:py_messages_proto", + "//src/proto/grpc/testing:py_test_proto", + requirement('google-auth'), + requirement('requests'), + requirement('enum34'), + requirement('urllib3'), + requirement('chardet'), + requirement('certifi'), + requirement('idna'), + ], + imports=["../../",], +) + +py_library( + name = "resources", + srcs = ["resources.py"], + data = [ + "//src/python/grpcio_tests/tests/interop/credentials", + ], +) + +py_library( + name = "server", + srcs = ["server.py"], + deps = [ + "//src/python/grpcio/grpc:grpcio", + ":methods", + ":resources", + "//src/python/grpcio_tests/tests/unit:test_common", + "//src/proto/grpc/testing:py_test_proto", + ], + imports=["../../",], +) + +py_test( + name="_insecure_intraop_test", + size="small", + srcs=["_insecure_intraop_test.py",], + main="_insecure_intraop_test.py", + deps=[ + "//src/python/grpcio/grpc:grpcio", + ":_intraop_test_case", + ":methods", + ":server", + "//src/python/grpcio_tests/tests/unit:test_common", + "//src/proto/grpc/testing:py_test_proto", + ], + imports=["../../",], + data=[ + "//src/python/grpcio_tests/tests/unit/credentials", + ], +) + +py_test( + name="_secure_intraop_test", + size="small", + srcs=["_secure_intraop_test.py",], + main="_secure_intraop_test.py", + deps=[ + "//src/python/grpcio/grpc:grpcio", + ":_intraop_test_case", + ":methods", + ":server", + "//src/python/grpcio_tests/tests/unit:test_common", + "//src/proto/grpc/testing:py_test_proto", + ], + imports=["../../",], +) + diff --git a/src/python/grpcio_tests/tests/interop/credentials/BUILD.bazel b/src/python/grpcio_tests/tests/interop/credentials/BUILD.bazel new file mode 100644 index 0000000000..bc2b997292 --- /dev/null +++ b/src/python/grpcio_tests/tests/interop/credentials/BUILD.bazel @@ -0,0 +1,9 @@ +package(default_visibility = ["//visibility:public"]) + +filegroup( + name="credentials", + srcs=glob([ + "**", + ]), +) + diff --git a/src/python/grpcio_tests/tests/interop/methods.py b/src/python/grpcio_tests/tests/interop/methods.py index cda15a68a3..c11f6c8fad 100644 --- a/src/python/grpcio_tests/tests/interop/methods.py +++ b/src/python/grpcio_tests/tests/interop/methods.py @@ -23,7 +23,6 @@ from google.auth import environment_vars as google_auth_environment_vars from google.auth.transport import grpc as google_auth_transport_grpc from google.auth.transport import requests as google_auth_transport_requests import grpc -from grpc.beta import implementations from src.proto.grpc.testing import empty_pb2 from src.proto.grpc.testing import messages_pb2 @@ -377,7 +376,7 @@ def _unimplemented_service(unimplemented_service_stub): def _custom_metadata(stub): initial_metadata_value = "test_initial_metadata_value" - trailing_metadata_value = "\x0a\x0b\x0a\x0b\x0a\x0b" + trailing_metadata_value = b"\x0a\x0b\x0a\x0b\x0a\x0b" metadata = ((_INITIAL_METADATA_KEY, initial_metadata_value), (_TRAILING_METADATA_KEY, trailing_metadata_value)) @@ -391,7 +390,7 @@ def _custom_metadata(stub): if trailing_metadata[_TRAILING_METADATA_KEY] != trailing_metadata_value: raise ValueError('expected trailing metadata %s, got %s' % (trailing_metadata_value, - initial_metadata[_TRAILING_METADATA_KEY])) + trailing_metadata[_TRAILING_METADATA_KEY])) # Testing with UnaryCall request = messages_pb2.SimpleRequest( @@ -422,7 +421,7 @@ def _compute_engine_creds(stub, args): def _oauth2_auth_token(stub, args): json_key_filename = os.environ[google_auth_environment_vars.CREDENTIALS] - wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] + wanted_email = json.load(open(json_key_filename, 'r'))['client_email'] response = _large_unary_common_behavior(stub, True, True, None) if wanted_email != response.username: raise ValueError('expected username %s, got %s' % (wanted_email, @@ -435,7 +434,7 @@ def _oauth2_auth_token(stub, args): def _jwt_token_creds(stub, args): json_key_filename = os.environ[google_auth_environment_vars.CREDENTIALS] - wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] + wanted_email = json.load(open(json_key_filename, 'r'))['client_email'] response = _large_unary_common_behavior(stub, True, False, None) if wanted_email != response.username: raise ValueError('expected username %s, got %s' % (wanted_email, @@ -444,7 +443,7 @@ def _jwt_token_creds(stub, args): def _per_rpc_creds(stub, args): json_key_filename = os.environ[google_auth_environment_vars.CREDENTIALS] - wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] + wanted_email = json.load(open(json_key_filename, 'r'))['client_email'] google_credentials, unused_project_id = google_auth.default( scopes=[args.oauth_scope]) call_credentials = grpc.metadata_call_credentials( @@ -457,6 +456,22 @@ def _per_rpc_creds(stub, args): response.username)) +def _special_status_message(stub, args): + details = b'\t\ntest with whitespace\r\nand Unicode BMP \xe2\x98\xba and non-BMP \xf0\x9f\x98\x88\t\n'.decode( + 'utf-8') + code = 2 + status = grpc.StatusCode.UNKNOWN # code = 2 + + # Test with a UnaryCall + request = messages_pb2.SimpleRequest( + response_type=messages_pb2.COMPRESSABLE, + response_size=1, + payload=messages_pb2.Payload(body=b'\x00'), + response_status=messages_pb2.EchoStatus(code=code, message=details)) + response_future = stub.UnaryCall.future(request) + _validate_status_code_and_details(response_future, status, details) + + @enum.unique class TestCase(enum.Enum): EMPTY_UNARY = 'empty_unary' @@ -476,6 +491,7 @@ class TestCase(enum.Enum): JWT_TOKEN_CREDS = 'jwt_token_creds' PER_RPC_CREDS = 'per_rpc_creds' TIMEOUT_ON_SLEEPING_SERVER = 'timeout_on_sleeping_server' + SPECIAL_STATUS_MESSAGE = 'special_status_message' def test_interoperability(self, stub, args): if self is TestCase.EMPTY_UNARY: @@ -512,6 +528,8 @@ class TestCase(enum.Enum): _jwt_token_creds(stub, args) elif self is TestCase.PER_RPC_CREDS: _per_rpc_creds(stub, args) + elif self is TestCase.SPECIAL_STATUS_MESSAGE: + _special_status_message(stub, args) else: raise NotImplementedError( 'Test case "%s" not implemented!' % self.name) diff --git a/src/python/grpcio_tests/tests/interop/resources.py b/src/python/grpcio_tests/tests/interop/resources.py index 2f76cf5db6..a55919a60a 100644 --- a/src/python/grpcio_tests/tests/interop/resources.py +++ b/src/python/grpcio_tests/tests/interop/resources.py @@ -14,27 +14,24 @@ """Constants and functions for data used in interoperability testing.""" import argparse +import pkgutil import os -import pkg_resources - _ROOT_CERTIFICATES_RESOURCE_PATH = 'credentials/ca.pem' _PRIVATE_KEY_RESOURCE_PATH = 'credentials/server1.key' _CERTIFICATE_CHAIN_RESOURCE_PATH = 'credentials/server1.pem' def test_root_certificates(): - return pkg_resources.resource_string(__name__, - _ROOT_CERTIFICATES_RESOURCE_PATH) + return pkgutil.get_data(__name__, _ROOT_CERTIFICATES_RESOURCE_PATH) def private_key(): - return pkg_resources.resource_string(__name__, _PRIVATE_KEY_RESOURCE_PATH) + return pkgutil.get_data(__name__, _PRIVATE_KEY_RESOURCE_PATH) def certificate_chain(): - return pkg_resources.resource_string(__name__, - _CERTIFICATE_CHAIN_RESOURCE_PATH) + return pkgutil.get_data(__name__, _CERTIFICATE_CHAIN_RESOURCE_PATH) def parse_bool(value): diff --git a/src/python/grpcio_tests/tests/interop/server.py b/src/python/grpcio_tests/tests/interop/server.py index 768cdaf468..72f88a1c98 100644 --- a/src/python/grpcio_tests/tests/interop/server.py +++ b/src/python/grpcio_tests/tests/interop/server.py @@ -25,8 +25,8 @@ from tests.interop import methods from tests.interop import resources from tests.unit import test_common -logging.basicConfig() _ONE_DAY_IN_SECONDS = 60 * 60 * 24 +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio_tests/tests/qps/worker_server.py b/src/python/grpcio_tests/tests/qps/worker_server.py index 740bdcf1eb..337a94b546 100644 --- a/src/python/grpcio_tests/tests/qps/worker_server.py +++ b/src/python/grpcio_tests/tests/qps/worker_server.py @@ -39,7 +39,7 @@ class WorkerServer(worker_service_pb2_grpc.WorkerServiceServicer): self._quit_event = threading.Event() def RunServer(self, request_iterator, context): - config = next(request_iterator).setup + config = next(request_iterator).setup #pylint: disable=stop-iteration-return server, port = self._create_server(config) cores = multiprocessing.cpu_count() server.start() @@ -102,7 +102,7 @@ class WorkerServer(worker_service_pb2_grpc.WorkerServiceServicer): return (server, port) def RunClient(self, request_iterator, context): - config = next(request_iterator).setup + config = next(request_iterator).setup #pylint: disable=stop-iteration-return client_runners = [] qps_data = histogram.Histogram(config.histogram_params.resolution, config.histogram_params.max_possible) diff --git a/src/python/grpcio_tests/tests/reflection/BUILD.bazel b/src/python/grpcio_tests/tests/reflection/BUILD.bazel new file mode 100644 index 0000000000..c0efb0b7ce --- /dev/null +++ b/src/python/grpcio_tests/tests/reflection/BUILD.bazel @@ -0,0 +1,21 @@ +load("@grpc_python_dependencies//:requirements.bzl", "requirement") + +package(default_visibility = ["//visibility:public"]) + +py_test( + name="_reflection_servicer_test", + size="small", + timeout="moderate", + srcs=["_reflection_servicer_test.py",], + main="_reflection_servicer_test.py", + deps=[ + "//src/python/grpcio/grpc:grpcio", + "//src/python/grpcio_reflection/grpc_reflection/v1alpha:grpc_reflection", + "//src/python/grpcio_tests/tests/unit:test_common", + "//src/proto/grpc/testing:py_empty_proto", + "//src/proto/grpc/testing/proto2:empty2_extensions_proto", + requirement('protobuf'), + ], + imports=["../../",], +) + diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json index 76d5d22d57..9cffd3df19 100644 --- a/src/python/grpcio_tests/tests/tests.json +++ b/src/python/grpcio_tests/tests/tests.json @@ -1,5 +1,6 @@ [ "_sanity._sanity_test.SanityTest", + "channelz._channelz_servicer_test.ChannelzServicerTest", "health_check._health_servicer_test.HealthServicerTest", "interop._insecure_intraop_test.InsecureIntraopTest", "interop._secure_intraop_test.SecureIntraopTest", @@ -42,11 +43,14 @@ "unit._cython.cygrpc_test.SecureServerSecureClient", "unit._cython.cygrpc_test.TypeSmokeTest", "unit._empty_message_test.EmptyMessageTest", + "unit._error_message_encoding_test.ErrorMessageEncodingTest", "unit._exit_test.ExitTest", "unit._interceptor_test.InterceptorTest", "unit._invalid_metadata_test.InvalidMetadataTest", "unit._invocation_defects_test.InvocationDefectsTest", + "unit._logging_test.LoggingTest", "unit._metadata_code_details_test.MetadataCodeDetailsTest", + "unit._metadata_flags_test.MetadataFlagsTest", "unit._metadata_test.MetadataTest", "unit._reconnect_test.ReconnectTest", "unit._resource_exhausted_test.ResourceExhaustedTest", diff --git a/src/python/grpcio_tests/tests/unit/BUILD.bazel b/src/python/grpcio_tests/tests/unit/BUILD.bazel index dcd6d9fbb2..de33b81e32 100644 --- a/src/python/grpcio_tests/tests/unit/BUILD.bazel +++ b/src/python/grpcio_tests/tests/unit/BUILD.bazel @@ -17,6 +17,7 @@ GRPCIO_TESTS_UNIT = [ "_interceptor_test.py", "_invalid_metadata_test.py", "_invocation_defects_test.py", + "_logging_test.py", "_metadata_code_details_test.py", "_metadata_test.py", # TODO: Issue 16336 diff --git a/src/python/grpcio_tests/tests/unit/_api_test.py b/src/python/grpcio_tests/tests/unit/_api_test.py index f6245be77d..38072861a4 100644 --- a/src/python/grpcio_tests/tests/unit/_api_test.py +++ b/src/python/grpcio_tests/tests/unit/_api_test.py @@ -14,6 +14,7 @@ """Test of gRPC Python's application-layer API.""" import unittest +import logging import six @@ -102,4 +103,5 @@ class ChannelTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_auth_context_test.py b/src/python/grpcio_tests/tests/unit/_auth_context_test.py index d174051070..b1b5bbdcab 100644 --- a/src/python/grpcio_tests/tests/unit/_auth_context_test.py +++ b/src/python/grpcio_tests/tests/unit/_auth_context_test.py @@ -15,6 +15,7 @@ import pickle import unittest +import logging import grpc from grpc import _channel @@ -187,4 +188,5 @@ class AuthContextTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_auth_test.py b/src/python/grpcio_tests/tests/unit/_auth_test.py index e2cb938936..d9df2add4f 100644 --- a/src/python/grpcio_tests/tests/unit/_auth_test.py +++ b/src/python/grpcio_tests/tests/unit/_auth_test.py @@ -16,6 +16,7 @@ import collections import threading import unittest +import logging from grpc import _auth @@ -77,4 +78,5 @@ class AccessTokenAuthMetadataPluginTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_channel_args_test.py b/src/python/grpcio_tests/tests/unit/_channel_args_test.py index dd1d2969a2..228c0e4c16 100644 --- a/src/python/grpcio_tests/tests/unit/_channel_args_test.py +++ b/src/python/grpcio_tests/tests/unit/_channel_args_test.py @@ -15,6 +15,7 @@ from concurrent import futures import unittest +import logging import grpc @@ -62,4 +63,5 @@ class ChannelArgsTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_channel_close_test.py b/src/python/grpcio_tests/tests/unit/_channel_close_test.py index af3a9ee1ee..82fa165710 100644 --- a/src/python/grpcio_tests/tests/unit/_channel_close_test.py +++ b/src/python/grpcio_tests/tests/unit/_channel_close_test.py @@ -13,6 +13,7 @@ # limitations under the License. """Tests server and client side compression.""" +import logging import threading import time import unittest @@ -182,4 +183,5 @@ class ChannelCloseTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py b/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py index f9eb0011dc..727fb7d65f 100644 --- a/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py +++ b/src/python/grpcio_tests/tests/unit/_channel_connectivity_test.py @@ -13,6 +13,7 @@ # limitations under the License. """Tests of grpc._channel.Channel connectivity.""" +import logging import threading import time import unittest @@ -142,4 +143,5 @@ class ChannelConnectivityTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) 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 30b486079c..345460ef40 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 @@ -15,6 +15,7 @@ import threading import unittest +import logging import grpc from tests.unit.framework.common import test_constants @@ -85,4 +86,5 @@ class ChannelReadyFutureTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_compression_test.py b/src/python/grpcio_tests/tests/unit/_compression_test.py index 0b11f03adf..876d8e827e 100644 --- a/src/python/grpcio_tests/tests/unit/_compression_test.py +++ b/src/python/grpcio_tests/tests/unit/_compression_test.py @@ -15,6 +15,7 @@ import unittest +import logging import grpc from grpc import _grpcio_metadata @@ -117,4 +118,5 @@ class CompressionTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_credentials_test.py b/src/python/grpcio_tests/tests/unit/_credentials_test.py index f487fe66a2..187a6f0388 100644 --- a/src/python/grpcio_tests/tests/unit/_credentials_test.py +++ b/src/python/grpcio_tests/tests/unit/_credentials_test.py @@ -14,6 +14,8 @@ """Tests of credentials.""" import unittest +import logging +import six import grpc @@ -52,6 +54,17 @@ class CredentialsTest(unittest.TestCase): self.assertIsInstance(channel_first_second_and_third, grpc.ChannelCredentials) + @unittest.skipIf(six.PY2, 'only invalid in Python3') + def test_invalid_string_certificate(self): + self.assertRaises( + TypeError, + grpc.ssl_channel_credentials, + root_certificates='A Certificate', + private_key=None, + certificate_chain=None, + ) + if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) 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 c55ef61c13..3e8393b53c 100644 --- a/src/python/grpcio_tests/tests/unit/_empty_message_test.py +++ b/src/python/grpcio_tests/tests/unit/_empty_message_test.py @@ -13,6 +13,7 @@ # limitations under the License. import unittest +import logging import grpc @@ -118,4 +119,5 @@ class EmptyMessageTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py b/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py new file mode 100644 index 0000000000..6c551df3ec --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py @@ -0,0 +1,86 @@ +# Copyright 2018 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests 'utf-8' encoded error message.""" + +import unittest +import weakref + +import grpc + +from tests.unit import test_common +from tests.unit.framework.common import test_constants + +_UNICODE_ERROR_MESSAGES = [ + b'\xe2\x80\x9d'.decode('utf-8'), + b'abc\x80\xd0\xaf'.decode('latin-1'), + b'\xc3\xa9'.decode('utf-8'), +] + +_REQUEST = b'\x00\x00\x00' +_RESPONSE = b'\x00\x00\x00' + +_UNARY_UNARY = '/test/UnaryUnary' + + +class _MethodHandler(grpc.RpcMethodHandler): + + def __init__(self, request_streaming=None, response_streaming=None): + self.request_streaming = request_streaming + self.response_streaming = response_streaming + self.request_deserializer = None + self.response_serializer = None + self.unary_stream = None + self.stream_unary = None + self.stream_stream = None + + def unary_unary(self, request, servicer_context): + servicer_context.set_code(grpc.StatusCode.UNKNOWN) + servicer_context.set_details(request.decode('utf-8')) + return _RESPONSE + + +class _GenericHandler(grpc.GenericRpcHandler): + + def __init__(self, test): + self._test = test + + def service(self, handler_call_details): + return _MethodHandler() + + +class ErrorMessageEncodingTest(unittest.TestCase): + + def setUp(self): + self._server = test_common.test_server() + self._server.add_generic_rpc_handlers((_GenericHandler( + weakref.proxy(self)),)) + port = self._server.add_insecure_port('[::]:0') + self._server.start() + self._channel = grpc.insecure_channel('localhost:%d' % port) + + def tearDown(self): + self._server.stop(0) + + def testMessageEncoding(self): + for message in _UNICODE_ERROR_MESSAGES: + multi_callable = self._channel.unary_unary(_UNARY_UNARY) + with self.assertRaises(grpc.RpcError) as cm: + multi_callable(message.encode('utf-8')) + + self.assertEqual(cm.exception.code(), grpc.StatusCode.UNKNOWN) + self.assertEqual(cm.exception.details(), message) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py index 0a0239a63d..d1263c2c56 100644 --- a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py +++ b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py @@ -16,6 +16,7 @@ import argparse import threading import time +import logging import grpc @@ -86,7 +87,7 @@ def hang_stream_stream(request_iterator, servicer_context): def hang_partial_stream_stream(request_iterator, servicer_context): for _ in range(test_constants.STREAM_LENGTH // 2): - yield next(request_iterator) + yield next(request_iterator) #pylint: disable=stop-iteration-return time.sleep(WAIT_TIME) @@ -161,6 +162,7 @@ def infinite_request_iterator(): if __name__ == '__main__': + logging.basicConfig() parser = argparse.ArgumentParser() parser.add_argument('scenario', type=str) parser.add_argument( diff --git a/src/python/grpcio_tests/tests/unit/_exit_test.py b/src/python/grpcio_tests/tests/unit/_exit_test.py index f40f3ae07c..5226537579 100644 --- a/src/python/grpcio_tests/tests/unit/_exit_test.py +++ b/src/python/grpcio_tests/tests/unit/_exit_test.py @@ -26,6 +26,7 @@ import sys import threading import time import unittest +import logging from tests.unit import _exit_scenarios @@ -187,4 +188,5 @@ class ExitTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_interceptor_test.py b/src/python/grpcio_tests/tests/unit/_interceptor_test.py index 3d547b71cd..99db0ac58b 100644 --- a/src/python/grpcio_tests/tests/unit/_interceptor_test.py +++ b/src/python/grpcio_tests/tests/unit/_interceptor_test.py @@ -17,6 +17,7 @@ import collections import itertools import threading import unittest +import logging from concurrent import futures import grpc @@ -598,4 +599,5 @@ class InterceptorTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py b/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py index f153089a24..0ff49490d5 100644 --- a/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py +++ b/src/python/grpcio_tests/tests/unit/_invalid_metadata_test.py @@ -14,6 +14,7 @@ """Test of RPCs made against gRPC Python's application-layer API.""" import unittest +import logging import grpc @@ -129,6 +130,10 @@ class InvalidMetadataTest(unittest.TestCase): self._stream_stream(request_iterator, metadata=metadata) self.assertIn(expected_error_details, str(exception_context.exception)) + def testInvalidMetadata(self): + self.assertRaises(TypeError, self._unary_unary, b'', metadata=42) + if __name__ == '__main__': + logging.basicConfig() 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 index 93a5fdf9ff..00949e2236 100644 --- a/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py +++ b/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py @@ -15,6 +15,7 @@ import itertools import threading import unittest +import logging import grpc @@ -271,4 +272,5 @@ class InvocationDefectsTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_logging_test.py b/src/python/grpcio_tests/tests/unit/_logging_test.py new file mode 100644 index 0000000000..631b9de9db --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_logging_test.py @@ -0,0 +1,80 @@ +# Copyright 2018 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Test of gRPC Python's interaction with the python logging module""" + +import unittest +import six +from six.moves import reload_module +import logging +import grpc +import functools +import sys + + +def patch_stderr(f): + + @functools.wraps(f) + def _impl(*args, **kwargs): + old_stderr = sys.stderr + sys.stderr = six.StringIO() + try: + f(*args, **kwargs) + finally: + sys.stderr = old_stderr + + return _impl + + +def isolated_logging(f): + + @functools.wraps(f) + def _impl(*args, **kwargs): + reload_module(logging) + reload_module(grpc) + try: + f(*args, **kwargs) + finally: + reload_module(logging) + + return _impl + + +class LoggingTest(unittest.TestCase): + + @isolated_logging + def test_logger_not_occupied(self): + self.assertEqual(0, len(logging.getLogger().handlers)) + + @patch_stderr + @isolated_logging + def test_handler_found(self): + self.assertEqual(0, len(sys.stderr.getvalue())) + + @isolated_logging + def test_can_configure_logger(self): + intended_stream = six.StringIO() + logging.basicConfig(stream=intended_stream) + self.assertEqual(1, len(logging.getLogger().handlers)) + self.assertIs(logging.getLogger().handlers[0].stream, intended_stream) + + @isolated_logging + def test_grpc_logger(self): + self.assertIn("grpc", logging.Logger.manager.loggerDict) + root_logger = logging.getLogger("grpc") + self.assertEqual(1, len(root_logger.handlers)) + self.assertIsInstance(root_logger.handlers[0], logging.NullHandler) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_metadata_code_details_test.py b/src/python/grpcio_tests/tests/unit/_metadata_code_details_test.py index ca10bd4dab..0dafab827a 100644 --- a/src/python/grpcio_tests/tests/unit/_metadata_code_details_test.py +++ b/src/python/grpcio_tests/tests/unit/_metadata_code_details_test.py @@ -15,6 +15,7 @@ import threading import unittest +import logging import grpc @@ -656,4 +657,5 @@ class MetadataCodeDetailsTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_metadata_flags_test.py b/src/python/grpcio_tests/tests/unit/_metadata_flags_test.py new file mode 100644 index 0000000000..2d352e99d4 --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_metadata_flags_test.py @@ -0,0 +1,251 @@ +# Copyright 2018 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests metadata flags feature by testing wait-for-ready semantics""" + +import time +import weakref +import unittest +import threading +import socket +from six.moves import queue + +import grpc + +from tests.unit import test_common +from tests.unit.framework.common import test_constants + +_UNARY_UNARY = '/test/UnaryUnary' +_UNARY_STREAM = '/test/UnaryStream' +_STREAM_UNARY = '/test/StreamUnary' +_STREAM_STREAM = '/test/StreamStream' + +_REQUEST = b'\x00\x00\x00' +_RESPONSE = b'\x00\x00\x00' + + +def handle_unary_unary(test, request, servicer_context): + return _RESPONSE + + +def handle_unary_stream(test, request, servicer_context): + for _ in range(test_constants.STREAM_LENGTH): + yield _RESPONSE + + +def handle_stream_unary(test, request_iterator, servicer_context): + for _ in request_iterator: + pass + return _RESPONSE + + +def handle_stream_stream(test, request_iterator, servicer_context): + for _ in request_iterator: + yield _RESPONSE + + +class _MethodHandler(grpc.RpcMethodHandler): + + def __init__(self, test, request_streaming, response_streaming): + self.request_streaming = request_streaming + self.response_streaming = response_streaming + self.request_deserializer = None + self.response_serializer = None + self.unary_unary = None + self.unary_stream = None + self.stream_unary = None + self.stream_stream = None + if self.request_streaming and self.response_streaming: + self.stream_stream = lambda req, ctx: handle_stream_stream(test, req, ctx) + elif self.request_streaming: + self.stream_unary = lambda req, ctx: handle_stream_unary(test, req, ctx) + elif self.response_streaming: + self.unary_stream = lambda req, ctx: handle_unary_stream(test, req, ctx) + else: + self.unary_unary = lambda req, ctx: handle_unary_unary(test, req, ctx) + + +class _GenericHandler(grpc.GenericRpcHandler): + + def __init__(self, test): + self._test = test + + def service(self, handler_call_details): + if handler_call_details.method == _UNARY_UNARY: + return _MethodHandler(self._test, False, False) + elif handler_call_details.method == _UNARY_STREAM: + return _MethodHandler(self._test, False, True) + elif handler_call_details.method == _STREAM_UNARY: + return _MethodHandler(self._test, True, False) + elif handler_call_details.method == _STREAM_STREAM: + return _MethodHandler(self._test, True, True) + else: + return None + + +def get_free_loopback_tcp_port(): + tcp = socket.socket(socket.AF_INET6) + tcp.bind(('', 0)) + address_tuple = tcp.getsockname() + return tcp, "[::1]:%s" % (address_tuple[1]) + + +def create_dummy_channel(): + """Creating dummy channels is a workaround for retries""" + _, addr = get_free_loopback_tcp_port() + return grpc.insecure_channel(addr) + + +def perform_unary_unary_call(channel, wait_for_ready=None): + channel.unary_unary(_UNARY_UNARY).__call__( + _REQUEST, + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + + +def perform_unary_unary_with_call(channel, wait_for_ready=None): + channel.unary_unary(_UNARY_UNARY).with_call( + _REQUEST, + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + + +def perform_unary_unary_future(channel, wait_for_ready=None): + channel.unary_unary(_UNARY_UNARY).future( + _REQUEST, + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready).result( + timeout=test_constants.LONG_TIMEOUT) + + +def perform_unary_stream_call(channel, wait_for_ready=None): + response_iterator = channel.unary_stream(_UNARY_STREAM).__call__( + _REQUEST, + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + for _ in response_iterator: + pass + + +def perform_stream_unary_call(channel, wait_for_ready=None): + channel.stream_unary(_STREAM_UNARY).__call__( + iter([_REQUEST] * test_constants.STREAM_LENGTH), + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + + +def perform_stream_unary_with_call(channel, wait_for_ready=None): + channel.stream_unary(_STREAM_UNARY).with_call( + iter([_REQUEST] * test_constants.STREAM_LENGTH), + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + + +def perform_stream_unary_future(channel, wait_for_ready=None): + channel.stream_unary(_STREAM_UNARY).future( + iter([_REQUEST] * test_constants.STREAM_LENGTH), + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready).result( + timeout=test_constants.LONG_TIMEOUT) + + +def perform_stream_stream_call(channel, wait_for_ready=None): + response_iterator = channel.stream_stream(_STREAM_STREAM).__call__( + iter([_REQUEST] * test_constants.STREAM_LENGTH), + timeout=test_constants.LONG_TIMEOUT, + wait_for_ready=wait_for_ready) + for _ in response_iterator: + pass + + +_ALL_CALL_CASES = [ + perform_unary_unary_call, perform_unary_unary_with_call, + perform_unary_unary_future, perform_unary_stream_call, + perform_stream_unary_call, perform_stream_unary_with_call, + perform_stream_unary_future, perform_stream_stream_call +] + + +class MetadataFlagsTest(unittest.TestCase): + + def check_connection_does_failfast(self, fn, channel, wait_for_ready=None): + try: + fn(channel, wait_for_ready) + self.fail("The Call should fail") + except BaseException as e: # pylint: disable=broad-except + self.assertIn('StatusCode.UNAVAILABLE', str(e)) + + def test_call_wait_for_ready_default(self): + for perform_call in _ALL_CALL_CASES: + self.check_connection_does_failfast(perform_call, + create_dummy_channel()) + + def test_call_wait_for_ready_disabled(self): + for perform_call in _ALL_CALL_CASES: + self.check_connection_does_failfast( + perform_call, create_dummy_channel(), wait_for_ready=False) + + def test_call_wait_for_ready_enabled(self): + # To test the wait mechanism, Python thread is required to make + # client set up first without handling them case by case. + # Also, Python thread don't pass the unhandled exceptions to + # main thread. So, it need another method to store the + # exceptions and raise them again in main thread. + unhandled_exceptions = queue.Queue() + tcp, addr = get_free_loopback_tcp_port() + wg = test_common.WaitGroup(len(_ALL_CALL_CASES)) + + def wait_for_transient_failure(channel_connectivity): + if channel_connectivity == grpc.ChannelConnectivity.TRANSIENT_FAILURE: + wg.done() + + def test_call(perform_call): + try: + channel = grpc.insecure_channel(addr) + channel.subscribe(wait_for_transient_failure) + perform_call(channel, wait_for_ready=True) + except BaseException as e: # pylint: disable=broad-except + # If the call failed, the thread would be destroyed. The channel + # object can be collected before calling the callback, which + # will result in a deadlock. + wg.done() + unhandled_exceptions.put(e, True) + + test_threads = [] + for perform_call in _ALL_CALL_CASES: + test_thread = threading.Thread( + target=test_call, args=(perform_call,)) + test_thread.exception = None + test_thread.start() + test_threads.append(test_thread) + + # Start the server after the connections are waiting + wg.wait() + tcp.close() + server = test_common.test_server() + server.add_generic_rpc_handlers((_GenericHandler(weakref.proxy(self)),)) + server.add_insecure_port(addr) + server.start() + + for test_thread in test_threads: + test_thread.join() + + # Stop the server to make test end properly + server.stop(0) + + if not unhandled_exceptions.empty(): + raise unhandled_exceptions.get(True) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_metadata_test.py b/src/python/grpcio_tests/tests/unit/_metadata_test.py index 5908421011..777ab683e3 100644 --- a/src/python/grpcio_tests/tests/unit/_metadata_test.py +++ b/src/python/grpcio_tests/tests/unit/_metadata_test.py @@ -15,6 +15,7 @@ import unittest import weakref +import logging import grpc from grpc import _channel @@ -237,4 +238,5 @@ class MetadataTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_reconnect_test.py b/src/python/grpcio_tests/tests/unit/_reconnect_test.py index a708d8d862..f6d4fcbd0a 100644 --- a/src/python/grpcio_tests/tests/unit/_reconnect_test.py +++ b/src/python/grpcio_tests/tests/unit/_reconnect_test.py @@ -15,6 +15,7 @@ import socket import time +import logging import unittest import grpc @@ -100,4 +101,5 @@ class ReconnectTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_resource_exhausted_test.py b/src/python/grpcio_tests/tests/unit/_resource_exhausted_test.py index df4b129018..4fead8fcd5 100644 --- a/src/python/grpcio_tests/tests/unit/_resource_exhausted_test.py +++ b/src/python/grpcio_tests/tests/unit/_resource_exhausted_test.py @@ -15,6 +15,7 @@ import threading import unittest +import logging import grpc from grpc import _channel @@ -253,4 +254,5 @@ class ResourceExhaustedTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_rpc_test.py b/src/python/grpcio_tests/tests/unit/_rpc_test.py index 34e7831a98..a768d6c7c1 100644 --- a/src/python/grpcio_tests/tests/unit/_rpc_test.py +++ b/src/python/grpcio_tests/tests/unit/_rpc_test.py @@ -16,6 +16,7 @@ import itertools import threading import unittest +import logging from concurrent import futures import grpc @@ -846,4 +847,5 @@ class RPCTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_server_ssl_cert_config_test.py b/src/python/grpcio_tests/tests/unit/_server_ssl_cert_config_test.py index 0d78034b7b..9e4bd61816 100644 --- a/src/python/grpcio_tests/tests/unit/_server_ssl_cert_config_test.py +++ b/src/python/grpcio_tests/tests/unit/_server_ssl_cert_config_test.py @@ -35,6 +35,7 @@ import os import six import threading import unittest +import logging from concurrent import futures @@ -69,18 +70,11 @@ SERVER_CERT_CHAIN_2_PEM = (resources.cert_hier_2_server_1_cert() + Call = collections.namedtuple('Call', ['did_raise', 'returned_cert_config']) -def _create_client_stub( - port, - expect_success, - root_certificates=None, - private_key=None, - certificate_chain=None, -): - channel = grpc.secure_channel('localhost:{}'.format(port), - grpc.ssl_channel_credentials( - root_certificates=root_certificates, - private_key=private_key, - certificate_chain=certificate_chain)) +def _create_channel(port, credentials): + return grpc.secure_channel('localhost:{}'.format(port), credentials) + + +def _create_client_stub(channel, expect_success): if expect_success: # per Nathaniel: there's some robustness issue if we start # using a channel without waiting for it to be actually ready @@ -175,14 +169,13 @@ class _ServerSSLCertReloadTest( root_certificates=None, private_key=None, certificate_chain=None): - client_stub = _create_client_stub( - self.port, - expect_success, + credentials = grpc.ssl_channel_credentials( root_certificates=root_certificates, private_key=private_key, certificate_chain=certificate_chain) - self._perform_rpc(client_stub, expect_success) - del client_stub + with _create_channel(self.port, credentials) as client_channel: + client_stub = _create_client_stub(client_channel, expect_success) + self._perform_rpc(client_stub, expect_success) def _test(self): # things should work... @@ -258,12 +251,13 @@ class _ServerSSLCertReloadTest( # now create the "persistent" clients self.cert_config_fetcher.reset() self.cert_config_fetcher.configure(False, None) - persistent_client_stub_A = _create_client_stub( + channel_A = _create_channel( self.port, - True, - root_certificates=CA_1_PEM, - private_key=CLIENT_KEY_2_PEM, - certificate_chain=CLIENT_CERT_CHAIN_2_PEM) + grpc.ssl_channel_credentials( + root_certificates=CA_1_PEM, + private_key=CLIENT_KEY_2_PEM, + certificate_chain=CLIENT_CERT_CHAIN_2_PEM)) + persistent_client_stub_A = _create_client_stub(channel_A, True) self._perform_rpc(persistent_client_stub_A, True) actual_calls = self.cert_config_fetcher.getCalls() self.assertEqual(len(actual_calls), 1) @@ -272,12 +266,13 @@ class _ServerSSLCertReloadTest( self.cert_config_fetcher.reset() self.cert_config_fetcher.configure(False, None) - persistent_client_stub_B = _create_client_stub( + channel_B = _create_channel( self.port, - True, - root_certificates=CA_1_PEM, - private_key=CLIENT_KEY_2_PEM, - certificate_chain=CLIENT_CERT_CHAIN_2_PEM) + grpc.ssl_channel_credentials( + root_certificates=CA_1_PEM, + private_key=CLIENT_KEY_2_PEM, + certificate_chain=CLIENT_CERT_CHAIN_2_PEM)) + persistent_client_stub_B = _create_client_stub(channel_B, True) self._perform_rpc(persistent_client_stub_B, True) actual_calls = self.cert_config_fetcher.getCalls() self.assertEqual(len(actual_calls), 1) @@ -358,6 +353,9 @@ class _ServerSSLCertReloadTest( actual_calls = self.cert_config_fetcher.getCalls() self.assertEqual(len(actual_calls), 0) + channel_A.close() + channel_B.close() + class ServerSSLCertConfigFetcherParamsChecks(unittest.TestCase): @@ -518,4 +516,5 @@ class ServerSSLCertReloadTestCertConfigReuse(_ServerSSLCertReloadTest): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_server_test.py b/src/python/grpcio_tests/tests/unit/_server_test.py index acf4a17921..2c8205f365 100644 --- a/src/python/grpcio_tests/tests/unit/_server_test.py +++ b/src/python/grpcio_tests/tests/unit/_server_test.py @@ -14,6 +14,7 @@ from concurrent import futures import unittest +import logging import grpc @@ -49,4 +50,5 @@ class ServerTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/_session_cache_test.py b/src/python/grpcio_tests/tests/unit/_session_cache_test.py index b4e4670fa7..9223a6da03 100644 --- a/src/python/grpcio_tests/tests/unit/_session_cache_test.py +++ b/src/python/grpcio_tests/tests/unit/_session_cache_test.py @@ -15,6 +15,7 @@ import pickle import unittest +import logging import grpc from grpc import _channel @@ -142,4 +143,5 @@ class SSLSessionCacheTest(unittest.TestCase): if __name__ == '__main__': + logging.basicConfig() unittest.main(verbosity=2) diff --git a/src/python/grpcio_tests/tests/unit/beta/BUILD.bazel b/src/python/grpcio_tests/tests/unit/beta/BUILD.bazel deleted file mode 100644 index d3e0fe20eb..0000000000 --- a/src/python/grpcio_tests/tests/unit/beta/BUILD.bazel +++ /dev/null @@ -1,75 +0,0 @@ -load("@grpc_python_dependencies//:requirements.bzl", "requirement") - -package(default_visibility = ["//visibility:public"]) - -py_library( - name = "test_utilities", - srcs = ["test_utilities.py"], - deps = [ - "//src/python/grpcio/grpc:grpcio", - ], -) - -py_test( - name = "_beta_features_test", - srcs = ["_beta_features_test.py"], - main = "_beta_features_test.py", - size = "small", - deps = [ - "//src/python/grpcio/grpc:grpcio", - "//src/python/grpcio_tests/tests/unit:resources", - "//src/python/grpcio_tests/tests/unit/framework/common", - ":test_utilities", - ], - imports=["../../../",], -) - -py_test( - name = "_connectivity_channel_test", - srcs = ["_connectivity_channel_test.py"], - main = "_connectivity_channel_test.py", - size = "small", - deps = [ - "//src/python/grpcio/grpc:grpcio", - ], -) - -# TODO(ghostwriternr): To be added later. -#py_test( -# name = "_implementations_test", -# srcs = ["_implementations_test.py"], -# main = "_implementations_test.py", -# size = "small", -# deps = [ -# "//src/python/grpcio/grpc:grpcio", -# "//src/python/grpcio_tests/tests/unit:resources", -# requirement('oauth2client'), -# ], -# imports=["../../../",], -#) - -py_test( - name = "_not_found_test", - srcs = ["_not_found_test.py"], - main = "_not_found_test.py", - size = "small", - deps = [ - "//src/python/grpcio/grpc:grpcio", - "//src/python/grpcio_tests/tests/unit/framework/common", - ], - imports=["../../../",], -) - -py_test( - name = "_utilities_test", - srcs = ["_utilities_test.py"], - main = "_utilities_test.py", - size = "small", - deps = [ - "//src/python/grpcio/grpc:grpcio", - "//src/python/grpcio_tests/tests/unit/framework/common", - ], - imports=["../../../",], -) - - diff --git a/src/python/grpcio_tests/tests/unit/resources.py b/src/python/grpcio_tests/tests/unit/resources.py index 51a8979f58..6efd870fc8 100644 --- a/src/python/grpcio_tests/tests/unit/resources.py +++ b/src/python/grpcio_tests/tests/unit/resources.py @@ -14,8 +14,7 @@ """Constants and functions for data used in testing.""" import os - -import pkg_resources +import pkgutil _ROOT_CERTIFICATES_RESOURCE_PATH = 'credentials/ca.pem' _PRIVATE_KEY_RESOURCE_PATH = 'credentials/server1.key' @@ -23,94 +22,92 @@ _CERTIFICATE_CHAIN_RESOURCE_PATH = 'credentials/server1.pem' def test_root_certificates(): - return pkg_resources.resource_string(__name__, - _ROOT_CERTIFICATES_RESOURCE_PATH) + return pkgutil.get_data(__name__, _ROOT_CERTIFICATES_RESOURCE_PATH) def private_key(): - return pkg_resources.resource_string(__name__, _PRIVATE_KEY_RESOURCE_PATH) + return pkgutil.get_data(__name__, _PRIVATE_KEY_RESOURCE_PATH) def certificate_chain(): - return pkg_resources.resource_string(__name__, - _CERTIFICATE_CHAIN_RESOURCE_PATH) + return pkgutil.get_data(__name__, _CERTIFICATE_CHAIN_RESOURCE_PATH) def cert_hier_1_root_ca_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/certs/ca.cert.pem') def cert_hier_1_intermediate_ca_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/intermediate/certs/intermediate.cert.pem' ) def cert_hier_1_client_1_key(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/intermediate/private/client.key.pem' ) def cert_hier_1_client_1_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/intermediate/certs/client.cert.pem' ) def cert_hier_1_server_1_key(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/intermediate/private/localhost-1.key.pem' ) def cert_hier_1_server_1_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_1/intermediate/certs/localhost-1.cert.pem' ) def cert_hier_2_root_ca_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/certs/ca.cert.pem') def cert_hier_2_intermediate_ca_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/intermediate/certs/intermediate.cert.pem' ) def cert_hier_2_client_1_key(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/intermediate/private/client.key.pem' ) def cert_hier_2_client_1_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/intermediate/certs/client.cert.pem' ) def cert_hier_2_server_1_key(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/intermediate/private/localhost-1.key.pem' ) def cert_hier_2_server_1_cert(): - return pkg_resources.resource_string( + return pkgutil.get_data( __name__, 'credentials/certificate_hierarchy_2/intermediate/certs/localhost-1.cert.pem' ) diff --git a/src/python/grpcio_tests/tests/unit/test_common.py b/src/python/grpcio_tests/tests/unit/test_common.py index 61717ae135..bc3b24862d 100644 --- a/src/python/grpcio_tests/tests/unit/test_common.py +++ b/src/python/grpcio_tests/tests/unit/test_common.py @@ -14,6 +14,7 @@ """Common code used throughout tests of gRPC.""" import collections +import threading from concurrent import futures import grpc @@ -107,3 +108,28 @@ def test_server(max_workers=10): return grpc.server( futures.ThreadPoolExecutor(max_workers=max_workers), options=(('grpc.so_reuseport', 0),)) + + +class WaitGroup(object): + + def __init__(self, n=0): + self.count = n + self.cv = threading.Condition() + + def add(self, n): + self.cv.acquire() + self.count += n + self.cv.release() + + def done(self): + self.cv.acquire() + self.count -= 1 + if self.count == 0: + self.cv.notify_all() + self.cv.release() + + def wait(self): + self.cv.acquire() + while self.count > 0: + self.cv.wait() + self.cv.release() |