Merge master

72 files changed, 1453 insertions, 209 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 3f214bf3b0..2e6839ef2d 100644
--- a/src/python/grpcio/grpc/BUILD.bazel
+++ b/src/python/grpcio/grpc/BUILD.bazel
@@ -2,7 +2,7 @@ load("@grpc_python_dependencies//:requirements.bzl", "requirement")
 
 package(default_visibility = ["//visibility:public"])
 
-py_binary(
+py_library(
     name = "grpcio",
     srcs = ["__init__.py"],
     deps = [
@@ -22,7 +22,6 @@ py_binary(
     data = [
         "//:grpc",
     ],
-    main = "__init__.py",
     imports = ["../",],
 )
 
diff --git a/src/python/grpcio/grpc/__init__.py b/src/python/grpcio/grpc/__init__.py
index 863696d236..88c5b6d5be 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
diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py
index 6876601785..ab154d8512 100644
--- a/src/python/grpcio/grpc/_channel.py
+++ b/src/python/grpcio/grpc/_channel.py
@@ -466,10 +466,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),
         ),
@@ -477,15 +478,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
@@ -494,15 +499,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),
@@ -511,9 +519,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:
@@ -527,17 +536,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:
@@ -552,6 +578,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
@@ -560,16 +587,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),
@@ -588,6 +623,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
@@ -596,13 +632,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:
@@ -618,32 +658,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)
@@ -651,6 +697,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
@@ -663,12 +710,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),),
@@ -683,6 +734,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):
@@ -979,5 +1048,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)
-        _moot(self._connectivity_state)
+        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 _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 8358cbec5b..f69127e38e 100644
--- a/src/python/grpcio/grpc/_common.py
+++ b/src/python/grpcio/grpc/_common.py
@@ -65,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 7124e83dee..cfd3a51d9b 100644
--- a/src/python/grpcio/grpc/_cython/BUILD.bazel
+++ b/src/python/grpcio/grpc/_cython/BUILD.bazel
@@ -8,6 +8,7 @@ pyx_library(
         "__init__.py",
         "cygrpc.pxd",
         "cygrpc.pyx",
+        "_cygrpc/_hooks.pyx.pxi",
         "_cygrpc/grpc_string.pyx.pxi",
         "_cygrpc/arguments.pyx.pxi",
         "_cygrpc/call.pyx.pxi",
@@ -15,6 +16,7 @@ pyx_library(
         "_cygrpc/credentials.pyx.pxi",
         "_cygrpc/completion_queue.pyx.pxi",
         "_cygrpc/event.pyx.pxi",
+        "_cygrpc/fork_posix.pyx.pxi",
         "_cygrpc/metadata.pyx.pxi",
         "_cygrpc/operation.pyx.pxi",
         "_cygrpc/records.pyx.pxi",
@@ -24,12 +26,14 @@ pyx_library(
         "_cygrpc/time.pyx.pxi",
         "_cygrpc/grpc_gevent.pyx.pxi",
         "_cygrpc/grpc.pxi",
+        "_cygrpc/_hooks.pxd.pxi",
         "_cygrpc/arguments.pxd.pxi",
         "_cygrpc/call.pxd.pxi",
         "_cygrpc/channel.pxd.pxi",
         "_cygrpc/credentials.pxd.pxi",
         "_cygrpc/completion_queue.pxd.pxi",
         "_cygrpc/event.pxd.pxi",
+        "_cygrpc/fork_posix.pxd.pxi",
         "_cygrpc/metadata.pxd.pxi",
         "_cygrpc/operation.pxd.pxi",
         "_cygrpc/records.pxd.pxi",
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
index 6cb1bc0c05..e0e068e452 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi
@@ -32,7 +32,7 @@ cdef class _ArgumentProcessor:
 
   cdef grpc_arg c_argument
 
-  cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references)
+  cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except *
 
 
 cdef class _ArgumentsProcessor:
@@ -42,5 +42,5 @@ cdef class _ArgumentsProcessor:
   cdef readonly list _references
   cdef grpc_channel_args _c_arguments
 
-  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable)
+  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except *
   cdef un_c(self)
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
index 2239e26b32..b7a4277ff6 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi
@@ -52,7 +52,7 @@ cdef grpc_arg _unwrap_grpc_arg(tuple wrapped_arg):
 
 cdef class _ArgumentProcessor:
 
-  cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references):
+  cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except *:
     key, value = argument
     cdef bytes encoded_key = _encode(key)
     if encoded_key is not key:
@@ -89,7 +89,7 @@ cdef class _ArgumentsProcessor:
     self._argument_processors = []
     self._references = []
 
-  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable):
+  cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except *:
     self._c_arguments.arguments_length = len(self._arguments)
     if self._c_arguments.arguments_length == 0:
       return NULL
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi
index 38fd9e78b2..63048e8da0 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi
@@ -144,8 +144,14 @@ cdef class SSLChannelCredentials(ChannelCredentials):
       return grpc_ssl_credentials_create(
           c_pem_root_certificates, NULL, NULL, NULL)
     else:
-      c_pem_key_certificate_pair.private_key = self._private_key
-      c_pem_key_certificate_pair.certificate_chain = self._certificate_chain
+      if self._private_key:
+        c_pem_key_certificate_pair.private_key = self._private_key
+      else:
+        c_pem_key_certificate_pair.private_key = NULL
+      if self._certificate_chain:
+        c_pem_key_certificate_pair.certificate_chain = self._certificate_chain
+      else:
+        c_pem_key_certificate_pair.certificate_chain = NULL
       return grpc_ssl_credentials_create(
           c_pem_root_certificates, &c_pem_key_certificate_pair, NULL, NULL)
 
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
index 4781219319..23428f0b0c 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
@@ -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:
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi
index c39fef08fa..53f0c7f0bb 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi
@@ -15,6 +15,12 @@
 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',))
 
 
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.pyx b/src/python/grpcio/grpc/_cython/cygrpc.pyx
index 026f7ba2e3..ae5c07bfc8 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
 
diff --git a/src/python/grpcio/grpc/_grpcio_metadata.py b/src/python/grpcio/grpc/_grpcio_metadata.py
index c33911ebc1..42b3a1ad49 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.15.0.dev0"""
+__version__ = """1.17.0.dev0"""
diff --git a/src/python/grpcio/grpc/_interceptor.py b/src/python/grpcio/grpc/_interceptor.py
index 1d2d374ad1..4345114026 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):
@@ -193,28 +198,39 @@ 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
@@ -225,25 +241,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 +287,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,31 +324,35 @@ 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
@@ -329,29 +367,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 +413,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_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index 38f6b4d24c..2c929102cc 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -92,7 +92,6 @@ CORE_SOURCE_FILES = [
     'src/core/lib/iomgr/error.cc',
     'src/core/lib/iomgr/ev_epoll1_linux.cc',
     'src/core/lib/iomgr/ev_epollex_linux.cc',
-    'src/core/lib/iomgr/ev_epollsig_linux.cc',
     'src/core/lib/iomgr/ev_poll_posix.cc',
     'src/core/lib/iomgr/ev_posix.cc',
     'src/core/lib/iomgr/ev_windows.cc',
@@ -209,6 +208,7 @@ 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',
@@ -256,11 +256,14 @@ CORE_SOURCE_FILES = [
     'src/core/lib/security/credentials/oauth2/oauth2_credentials.cc',
     'src/core/lib/security/credentials/plugin/plugin_credentials.cc',
     'src/core/lib/security/credentials/ssl/ssl_credentials.cc',
-    'src/core/lib/security/security_connector/alts_security_connector.cc',
+    'src/core/lib/security/security_connector/alts/alts_security_connector.cc',
+    'src/core/lib/security/security_connector/fake/fake_security_connector.cc',
     'src/core/lib/security/security_connector/load_system_roots_fallback.cc',
     'src/core/lib/security/security_connector/load_system_roots_linux.cc',
-    'src/core/lib/security/security_connector/local_security_connector.cc',
+    'src/core/lib/security/security_connector/local/local_security_connector.cc',
     'src/core/lib/security/security_connector/security_connector.cc',
+    'src/core/lib/security/security_connector/ssl/ssl_security_connector.cc',
+    'src/core/lib/security/security_connector/ssl_utils.cc',
     'src/core/lib/security/transport/client_auth_filter.cc',
     'src/core/lib/security/transport/secure_endpoint.cc',
     'src/core/lib/security/transport/security_handshaker.cc',
@@ -279,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',
@@ -315,6 +318,7 @@ CORE_SOURCE_FILES = [
     'src/core/ext/filters/client_channel/client_channel_factory.cc',
     'src/core/ext/filters/client_channel/client_channel_plugin.cc',
     'src/core/ext/filters/client_channel/connector.cc',
+    'src/core/ext/filters/client_channel/health/health_check_client.cc',
     'src/core/ext/filters/client_channel/http_connect_handshaker.cc',
     'src/core/ext/filters/client_channel/http_proxy.cc',
     'src/core/ext/filters/client_channel/lb_policy.cc',
@@ -329,9 +333,8 @@ CORE_SOURCE_FILES = [
     '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/tsi/alts_transport_security.cc',
+    'src/core/ext/filters/client_channel/health/health.pb.c',
     'src/core/tsi/fake_transport_security.cc',
     'src/core/tsi/local_transport_security.cc',
     'src/core/tsi/ssl/session_cache/ssl_session_boringssl.cc',
@@ -350,10 +353,14 @@ CORE_SOURCE_FILES = [
     'src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc',
     'src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.cc',
     'src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc',
+    'src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc',
     'src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/google/protobuf/duration.pb.c',
     'src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/google/protobuf/timestamp.pb.c',
     'src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c',
-    'src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc',
+    'src/core/ext/filters/client_channel/lb_policy/xds/xds.cc',
+    'src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_secure.cc',
+    'src/core/ext/filters/client_channel/lb_policy/xds/xds_client_stats.cc',
+    'src/core/ext/filters/client_channel/lb_policy/xds/xds_load_balancer_api.cc',
     'src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc',
     'src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc',
     'src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc',
diff --git a/src/python/grpcio/grpc_version.py b/src/python/grpcio/grpc_version.py
index 9337800a33..71113e68d9 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.15.0.dev0'
+VERSION = '1.17.0.dev0'
diff --git a/src/python/grpcio_health_checking/grpc_health/v1/BUILD.bazel b/src/python/grpcio_health_checking/grpc_health/v1/BUILD.bazel
new file mode 100644
index 0000000000..ce3121fa90
--- /dev/null
+++ b/src/python/grpcio_health_checking/grpc_health/v1/BUILD.bazel
@@ -0,0 +1,33 @@
+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_health_proto",
+    srcs = [
+        "//src/proto/grpc/health/v1:health_proto_file",
+    ],
+    outs = ["health.proto",],
+    cmd = "cp $< $@",
+)
+
+py_proto_library(
+    name = "py_health_proto",
+    protos = [":mv_health_proto",],
+    with_grpc = True,
+    deps = [
+        requirement('protobuf'),
+    ],
+)
+
+py_library(
+    name = "grpc_health",
+    srcs = ["health.py",],
+    deps = [
+        ":py_health_proto",
+        "//src/python/grpcio/grpc:grpcio",
+    ],
+    imports=["../../",],
+)
+
diff --git a/src/python/grpcio_health_checking/grpc_version.py b/src/python/grpcio_health_checking/grpc_version.py
index 3b84f7a4c5..a30aac2e0b 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.15.0.dev0'
+VERSION = '1.17.0.dev0'
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 7b0e48ea23..aafea9fe76 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.15.0.dev0'
+VERSION = '1.17.0.dev0'
diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py
index 191b1c1726..d7205ca579 100644
--- a/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py
+++ b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py
@@ -18,6 +18,7 @@ import threading
 import grpc
 
 _NOT_YET_OBSERVED = object()
+logging.basicConfig()
 _LOGGER = logging.getLogger(__name__)
 
 
diff --git a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py
index b856da100f..736b714dc6 100644
--- a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py
+++ b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py
@@ -18,6 +18,7 @@ import threading
 import grpc
 from grpc_testing import _common
 
+logging.basicConfig()
 _LOGGER = logging.getLogger(__name__)
 
 
diff --git a/src/python/grpcio_testing/grpc_testing/_time.py b/src/python/grpcio_testing/grpc_testing/_time.py
index 75e6db3458..9692c34e6f 100644
--- a/src/python/grpcio_testing/grpc_testing/_time.py
+++ b/src/python/grpcio_testing/grpc_testing/_time.py
@@ -21,6 +21,7 @@ import time as _time
 import grpc
 import grpc_testing
 
+logging.basicConfig()
 _LOGGER = logging.getLogger(__name__)
 
 
diff --git a/src/python/grpcio_testing/grpc_version.py b/src/python/grpcio_testing/grpc_version.py
index df9953fa25..876acd3142 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.15.0.dev0'
+VERSION = '1.17.0.dev0'
diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py
index 0dfbf3180b..d163f6fb68 100644
--- a/src/python/grpcio_tests/commands.py
+++ b/src/python/grpcio_tests/commands.py
@@ -116,6 +116,8 @@ class TestGevent(setuptools.Command):
         # eventually succeed, but need to dig into performance issues.
         'unit._cython._no_messages_server_completion_queue_per_call_test.Test.test_rpcs',
         'unit._cython._no_messages_single_server_completion_queue_test.Test.test_rpcs',
+        # TODO(https://github.com/grpc/grpc/issues/16890) enable this test
+        'unit._cython._channel_test.ChannelTest.test_multiple_channels_lonely_connectivity',
         # I have no idea why this doesn't work in gevent, but it shouldn't even be
         # using the c-core
         'testing._client_test.ClientTest.test_infinite_request_stream_real_time',
@@ -128,7 +130,9 @@ 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')
     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 b2cf129e4f..cc9b41587c 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.15.0.dev0'
+VERSION = '1.17.0.dev0'
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/health_check/BUILD.bazel b/src/python/grpcio_tests/tests/health_check/BUILD.bazel
new file mode 100644
index 0000000000..19e1e1b2e1
--- /dev/null
+++ b/src/python/grpcio_tests/tests/health_check/BUILD.bazel
@@ -0,0 +1,15 @@
+package(default_visibility = ["//visibility:public"])
+
+py_test(
+    name = "health_servicer_test",
+    srcs = ["_health_servicer_test.py"],
+    main = "_health_servicer_test.py",
+    size = "small",
+    deps = [
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_health_checking/grpc_health/v1:grpc_health",
+        "//src/python/grpcio_tests/tests/unit:test_common",
+    ],
+    imports = ["../../",],
+)
+
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..721dedf0b7 100644
--- a/src/python/grpcio_tests/tests/interop/methods.py
+++ b/src/python/grpcio_tests/tests/interop/methods.py
@@ -457,6 +457,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 +492,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 +529,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 fd28d498a1..72f88a1c98 100644
--- a/src/python/grpcio_tests/tests/interop/server.py
+++ b/src/python/grpcio_tests/tests/interop/server.py
@@ -26,6 +26,7 @@ from tests.interop import resources
 from tests.unit import test_common
 
 _ONE_DAY_IN_SECONDS = 60 * 60 * 24
+logging.basicConfig()
 _LOGGER = logging.getLogger(__name__)
 
 
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/testing/BUILD.bazel b/src/python/grpcio_tests/tests/testing/BUILD.bazel
new file mode 100644
index 0000000000..9bdd616c56
--- /dev/null
+++ b/src/python/grpcio_tests/tests/testing/BUILD.bazel
@@ -0,0 +1,30 @@
+package(default_visibility = ["//visibility:public"])
+
+py_library(
+    name = "testing",
+    srcs = ["__init__.py",],
+    deps = [
+        # ":_application_common",
+        ":_server_application",
+    ],
+)
+
+# TODO(ghostwriternr): To be added later.
+# py_library(
+#     name = "_application_common",
+#     srcs = ["_application_common.py",],
+#     deps = [
+#         "//src/python/grpcio_tests/tests/testing/proto:requests",
+#         "//src/python/grpcio_tests/tests/testing/proto:services",
+#     ],
+#     imports = ["../../",],
+# )
+
+py_library(
+    name = "_server_application",
+    srcs = ["_server_application.py",],
+    imports = ["../../",],
+)
+
+
+
diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json
index 76d5d22d57..a3006d9afc 100644
--- a/src/python/grpcio_tests/tests/tests.json
+++ b/src/python/grpcio_tests/tests/tests.json
@@ -42,11 +42,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
new file mode 100644
index 0000000000..de33b81e32
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/BUILD.bazel
@@ -0,0 +1,84 @@
+load("@grpc_python_dependencies//:requirements.bzl", "requirement")
+
+package(default_visibility = ["//visibility:public"])
+
+GRPCIO_TESTS_UNIT = [
+    "_api_test.py",
+    "_auth_context_test.py",
+    "_auth_test.py",
+    "_channel_args_test.py",
+    "_channel_close_test.py",
+    "_channel_connectivity_test.py",
+    "_channel_ready_future_test.py",
+    "_compression_test.py",
+    "_credentials_test.py",
+    "_empty_message_test.py",
+    "_exit_test.py",
+    "_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
+    # "_reconnect_test.py",
+    "_resource_exhausted_test.py",
+    "_rpc_test.py",
+    # TODO(ghostwriternr): To be added later.
+    # "_server_ssl_cert_config_test.py",
+    "_server_test.py",
+    "_session_cache_test.py",
+]
+
+py_library(
+    name = "resources",
+    srcs = ["resources.py"],
+    data=[
+        "//src/python/grpcio_tests/tests/unit/credentials",
+    ],
+)
+
+py_library(
+    name = "test_common",
+    srcs = ["test_common.py"],
+)
+
+py_library(
+    name = "_exit_scenarios",
+    srcs = ["_exit_scenarios.py"],
+)
+
+py_library(
+    name = "_thread_pool",
+    srcs = ["_thread_pool.py"],
+)
+
+py_library(
+    name = "_from_grpc_import_star",
+    srcs = ["_from_grpc_import_star.py"],
+)
+
+[
+    py_test(
+        name=test_file_name[:-3],
+        size="small",
+        srcs=[test_file_name],
+        main=test_file_name,
+        deps=[
+            "//src/python/grpcio/grpc:grpcio",
+            ":resources",
+            ":test_common",
+            ":_exit_scenarios",
+            ":_thread_pool",
+            ":_from_grpc_import_star",
+            "//src/python/grpcio_tests/tests/unit/framework/common",
+            "//src/python/grpcio_tests/tests/testing",
+            requirement('six'),
+        ],
+        imports=["../../",],
+        data=[
+            "//src/python/grpcio_tests/tests/unit/credentials",
+        ],
+    ) for test_file_name in GRPCIO_TESTS_UNIT
+]
+
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 869c2f4d2f..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
 
@@ -33,6 +34,14 @@ TEST_CHANNEL_ARGS = (
     ('arg6', TestPointerWrapper()),
 )
 
+INVALID_TEST_CHANNEL_ARGS = [
+    {
+        'foo': 'bar'
+    },
+    (('key',),),
+    'str',
+]
+
 
 class ChannelArgsTest(unittest.TestCase):
 
@@ -44,6 +53,15 @@ class ChannelArgsTest(unittest.TestCase):
             futures.ThreadPoolExecutor(max_workers=1),
             options=TEST_CHANNEL_ARGS)
 
+    def test_invalid_client_args(self):
+        for invalid_arg in INVALID_TEST_CHANNEL_ARGS:
+            self.assertRaises(
+                ValueError,
+                grpc.insecure_channel,
+                'localhost:8080',
+                options=invalid_arg)
+
 
 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..be7378ecbc 100644
--- a/src/python/grpcio_tests/tests/unit/_credentials_test.py
+++ b/src/python/grpcio_tests/tests/unit/_credentials_test.py
@@ -14,6 +14,7 @@
 """Tests of credentials."""
 
 import unittest
+import logging
 
 import grpc
 
@@ -54,4 +55,5 @@ class CredentialsTest(unittest.TestCase):
 
 
 if __name__ == '__main__':
+    logging.basicConfig()
     unittest.main(verbosity=2)
diff --git a/src/python/grpcio_tests/tests/unit/_cython/BUILD.bazel b/src/python/grpcio_tests/tests/unit/_cython/BUILD.bazel
new file mode 100644
index 0000000000..458a6b1fb8
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_cython/BUILD.bazel
@@ -0,0 +1,46 @@
+load("@grpc_python_dependencies//:requirements.bzl", "requirement")
+
+package(default_visibility = ["//visibility:public"])
+
+GRPCIO_TESTS_UNIT_CYTHON = [
+    "_cancel_many_calls_test.py",
+    "_channel_test.py",
+    "_no_messages_server_completion_queue_per_call_test.py",
+    "_no_messages_single_server_completion_queue_test.py",
+    "_read_some_but_not_all_responses_test.py",
+    "_server_test.py",
+    "cygrpc_test.py",
+]
+
+py_library(
+    name = "common",
+    srcs = ["_common.py"],
+)
+
+py_library(
+    name = "test_utilities",
+    srcs = ["test_utilities.py"],
+)
+
+[
+    py_test(
+        name=test_file_name[:-3],
+        size="small",
+        srcs=[test_file_name],
+        main=test_file_name,
+        deps=[
+            "//src/python/grpcio/grpc:grpcio",
+            ":common",
+            ":test_utilities",
+            "//src/python/grpcio_tests/tests/unit/framework/common",
+            "//src/python/grpcio_tests/tests/unit:test_common",
+            "//src/python/grpcio_tests/tests/unit:resources",
+        ],
+        imports=["../../../",],
+        data=[
+            "//src/python/grpcio_tests/tests/unit/credentials",
+        ],
+    ) for test_file_name in GRPCIO_TESTS_UNIT_CYTHON
+]
+
+
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..f489db12cb 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
 
@@ -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..9771764635 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
 
@@ -131,4 +132,5 @@ class InvalidMetadataTest(unittest.TestCase):
 
 
 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..e733a59a5b 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
 
@@ -518,4 +519,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
new file mode 100644
index 0000000000..d3e0fe20eb
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/beta/BUILD.bazel
@@ -0,0 +1,75 @@
+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/credentials/BUILD.bazel b/src/python/grpcio_tests/tests/unit/credentials/BUILD.bazel
new file mode 100644
index 0000000000..358216db58
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/credentials/BUILD.bazel
@@ -0,0 +1,10 @@
+package(default_visibility = ["//visibility:public"])
+
+filegroup(
+    name="credentials",
+    srcs=glob([
+        "**",
+    ]),
+)
+
+
diff --git a/src/python/grpcio_tests/tests/unit/framework/common/BUILD.bazel b/src/python/grpcio_tests/tests/unit/framework/common/BUILD.bazel
new file mode 100644
index 0000000000..c206a04fad
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/framework/common/BUILD.bazel
@@ -0,0 +1,11 @@
+package(default_visibility = ["//visibility:public"])
+
+py_library(
+    name = "common",
+    srcs = [
+        "test_constants.py",
+        "test_control.py",
+        "test_coverage.py",
+    ],
+)
+
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()