Add insecure_channel_credentials function

src/python/grpcio/grpc/__init__.py

@@ -569,7 +569,6 @@ class StreamStreamClientInterceptor(six.with_metaclass(abc.ABCMeta)):
 
 ############  Authentication & Authorization Interfaces & Classes  #############
 
-
 class ChannelCredentials(object):
     """An encapsulation of the data required to create a secure Channel.
 
@@ -600,6 +599,13 @@ class CallCredentials(object):
         self._credentials = credentials
 
 
+_insecure_channel_credentials = object()
+
+
+def insecure_channel_credentials():
+    return ChannelCredentials(_insecure_channel_credentials)
+
+
 class AuthMetadataContext(six.with_metaclass(abc.ABCMeta)):
     """Provides information to call credentials metadata plugins.
 
@@ -1879,6 +1885,10 @@ def secure_channel(target, credentials, options=None, compression=None):
       A Channel.
     """
     from grpc import _channel  # pylint: disable=cyclic-import
+    if credentials._credentials is _insecure_channel_credentials:
+        raise ValueError(
+            "secure_channel cannot be called with insecure credentials." +
+            " Call insecure_channel instead.")
     return _channel.Channel(target, () if options is None else options,
                             credentials._credentials, compression)
 
@@ -1949,7 +1959,6 @@ def server(thread_pool,
                                  maximum_concurrent_rpcs, compression)
 
 
-
 @contextlib.contextmanager
 def _create_servicer_context(rpc_event, state, request_deserializer):
     from grpc import _server  # pylint: disable=cyclic-import
@@ -2022,6 +2031,7 @@ __all__ = (
     'access_token_call_credentials',
     'composite_call_credentials',
     'composite_channel_credentials',
+    'insecure_channel_credentials',
     'local_channel_credentials',
     'local_server_credentials',
     'ssl_server_credentials',

src/python/grpcio/grpc/_simple_stubs.py

@@ -9,13 +9,14 @@ import threading
 import grpc
 from typing import Any, AnyStr, Callable, Iterator, Optional, Sequence, Tuple, TypeVar, Union
 
-
 _LOGGER = logging.getLogger(__name__)
 
 _EVICTION_PERIOD_KEY = "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS"
 if _EVICTION_PERIOD_KEY in os.environ:
-    _EVICTION_PERIOD = datetime.timedelta(seconds=float(os.environ[_EVICTION_PERIOD_KEY]))
-    _LOGGER.info(f"Setting managed channel eviction period to {_EVICTION_PERIOD}")
+    _EVICTION_PERIOD = datetime.timedelta(
+        seconds=float(os.environ[_EVICTION_PERIOD_KEY]))
+    _LOGGER.info(
+        f"Setting managed channel eviction period to {_EVICTION_PERIOD}")
 else:
     _EVICTION_PERIOD = datetime.timedelta(minutes=10)
 
@@ -24,39 +25,40 @@ if _MAXIMUM_CHANNELS_KEY in os.environ:
     _MAXIMUM_CHANNELS = int(os.environ[_MAXIMUM_CHANNELS_KEY])
     _LOGGER.info(f"Setting maximum managed channels to {_MAXIMUM_CHANNELS}")
 else:
-    _MAXIMUM_CHANNELS = 2 ** 8
+    _MAXIMUM_CHANNELS = 2**8
+
 
-def _create_channel(target: str,
-                    options: Sequence[Tuple[str, str]],
-                    channel_credentials: Optional[grpc.ChannelCredentials],
+def _create_channel(target: str, options: Sequence[Tuple[str, str]],
+                    channel_credentials: grpc.ChannelCredentials,
                     compression: Optional[grpc.Compression]) -> grpc.Channel:
-    if channel_credentials is None:
+    if channel_credentials._credentials is grpc._insecure_channel_credentials:
         _LOGGER.info(f"Creating insecure channel with options '{options}' " +
-                       f"and compression '{compression}'")
+                     f"and compression '{compression}'")
         return grpc.insecure_channel(target,
                                      options=options,
                                      compression=compression)
     else:
-        _LOGGER.info(f"Creating secure channel with credentials '{channel_credentials}', " +
-                       f"options '{options}' and compression '{compression}'")
+        _LOGGER.info(
+            f"Creating secure channel with credentials '{channel_credentials}', "
+            + f"options '{options}' and compression '{compression}'")
         return grpc.secure_channel(target,
                                    credentials=channel_credentials,
                                    options=options,
                                    compression=compression)
 
+
 class ChannelCache:
     _singleton = None
     _lock = threading.RLock()
     _condition = threading.Condition(lock=_lock)
     _eviction_ready = threading.Event()
 
-
     def __init__(self):
         self._mapping = collections.OrderedDict()
-        self._eviction_thread = threading.Thread(target=ChannelCache._perform_evictions, daemon=True)
+        self._eviction_thread = threading.Thread(
+            target=ChannelCache._perform_evictions, daemon=True)
         self._eviction_thread.start()
 
-
     @staticmethod
     def get():
         with ChannelCache._lock:
@@ -72,7 +74,6 @@ class ChannelCache:
         channel.close()
         del channel
 
-
     # TODO: Refactor. Way too deeply nested.
     @staticmethod
     def _perform_evictions():
@@ -86,7 +87,8 @@ class ChannelCache:
                     ChannelCache._singleton._evict_locked(key)
                     # And immediately reevaluate.
                 else:
-                    key, (channel, eviction_time) = next(iter(ChannelCache._singleton._mapping.items()))
+                    key, (channel, eviction_time) = next(
+                        iter(ChannelCache._singleton._mapping.items()))
                     now = datetime.datetime.now()
                     if eviction_time <= now:
                         ChannelCache._singleton._evict_locked(key)
@@ -95,12 +97,8 @@ class ChannelCache:
                         time_to_eviction = (eviction_time - now).total_seconds()
                         ChannelCache._condition.wait(timeout=time_to_eviction)
 
-
-
-    def get_channel(self,
-                    target: str,
-                    options: Sequence[Tuple[str, str]],
-                    channel_credentials: Optional[grpc.ChannelCredentials],
+    def get_channel(self, target: str, options: Sequence[Tuple[str, str]],
+                    channel_credentials: grpc.ChannelCredentials,
                     compression: Optional[grpc.Compression]) -> grpc.Channel:
         key = (target, options, channel_credentials, compression)
         with self._lock:
@@ -109,12 +107,16 @@ class ChannelCache:
             if channel_data is not None:
                 channel = channel_data[0]
                 self._mapping.pop(key)
-                self._mapping[key] = (channel, datetime.datetime.now() + _EVICTION_PERIOD)
+                self._mapping[key] = (channel, datetime.datetime.now() +
+                                      _EVICTION_PERIOD)
                 return channel
             else:
-                channel = _create_channel(target, options, channel_credentials, compression)
-                self._mapping[key] = (channel, datetime.datetime.now() + _EVICTION_PERIOD)
-                if len(self._mapping) == 1 or len(self._mapping) >= _MAXIMUM_CHANNELS:
+                channel = _create_channel(target, options, channel_credentials,
+                                          compression)
+                self._mapping[key] = (channel, datetime.datetime.now() +
+                                      _EVICTION_PERIOD)
+                if len(self._mapping) == 1 or len(
+                        self._mapping) >= _MAXIMUM_CHANNELS:
                     self._condition.notify()
                 return channel
 
@@ -122,22 +124,38 @@ class ChannelCache:
         with self._lock:
             return len(self._mapping)
 
+
 RequestType = TypeVar('RequestType')
 ResponseType = TypeVar('ResponseType')
 
-def unary_unary(request: RequestType,
-                target: str,
-                method: str,
-                request_serializer: Optional[Callable[[Any], bytes]] = None,
-                request_deserializer: Optional[Callable[[bytes], Any]] = None,
-                options: Sequence[Tuple[AnyStr, AnyStr]] = (),
-                # TODO: Somehow make insecure_channel opt-in, not the default.
-                channel_credentials: Optional[grpc.ChannelCredentials] = None,
-                call_credentials: Optional[grpc.CallCredentials] = None,
-                compression: Optional[grpc.Compression] = None,
-                wait_for_ready: Optional[bool] = None,
-                timeout: Optional[float] = None,
-                metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None) -> ResponseType:
+
+# TODO(rbellevi): Consider a credential type that has the
+#   following functionality matrix:
+#
+#   +----------+-------+--------+
+#   |          | local | remote |
+#   |----------+-------+--------+
+#   | secure   | o     | o      |
+#   | insecure | o     | x      |
+#   +----------+-------+--------+
+#
+#  Make this the default option.
+
+# TODO: Make LocalChannelCredentials the default.
+def unary_unary(
+        request: RequestType,
+        target: str,
+        method: str,
+        request_serializer: Optional[Callable[[Any], bytes]] = None,
+        request_deserializer: Optional[Callable[[bytes], Any]] = None,
+        options: Sequence[Tuple[AnyStr, AnyStr]] = (),
+        channel_credentials: Optional[grpc.ChannelCredentials] = None,
+        call_credentials: Optional[grpc.CallCredentials] = None,
+        compression: Optional[grpc.Compression] = None,
+        wait_for_ready: Optional[bool] = None,
+        timeout: Optional[float] = None,
+        metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
+) -> ResponseType:
     """Invokes a unary-unary RPC without an explicitly specified channel.
 
     This is backed by a per-process cache of channels. Channels are evicted
@@ -147,7 +165,7 @@ def unary_unary(request: RequestType,
     The default eviction period is 10 minutes. One may set the environment
     variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this.
 
-    The default maximum maximum number of channels is 256. One may set the
+    The default maximum number of channels is 256. One may set the
     environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure
     this.
 
@@ -162,7 +180,8 @@ def unary_unary(request: RequestType,
       options: An optional list of key-value pairs (channel args in gRPC Core
         runtime) to configure the channel.
       channel_credentials: A credential applied to the whole channel, e.g. the
-        return value of grpc.ssl_channel_credentials().
+        return value of grpc.ssl_channel_credentials() or
+        grpc.insecure_channel_credentials().
       call_credentials: A call credential applied to each call individually,
         e.g. the output of grpc.metadata_call_credentials() or
         grpc.access_token_call_credentials().
@@ -180,8 +199,11 @@ def unary_unary(request: RequestType,
     Returns:
       The response to the RPC.
     """
-    channel = ChannelCache.get().get_channel(target, options, channel_credentials, compression)
-    multicallable = channel.unary_unary(method, request_serializer, request_deserializer)
+    channel_credentials = channel_credentials or grpc.local_channel_credentials()
+    channel = ChannelCache.get().get_channel(target, options,
+                                             channel_credentials, compression)
+    multicallable = channel.unary_unary(method, request_serializer,
+                                        request_deserializer)
     return multicallable(request,
                          metadata=metadata,
                          wait_for_ready=wait_for_ready,
@@ -189,18 +211,20 @@ def unary_unary(request: RequestType,
                          timeout=timeout)
 
 
-def unary_stream(request: RequestType,
-                 target: str,
-                 method: str,
-                 request_serializer: Optional[Callable[[Any], bytes]] = None,
-                 request_deserializer: Optional[Callable[[bytes], Any]] = None,
-                 options: Sequence[Tuple[AnyStr, AnyStr]] = (),
-                 channel_credentials: Optional[grpc.ChannelCredentials] = None,
-                 call_credentials: Optional[grpc.CallCredentials] = None,
-                 compression: Optional[grpc.Compression] = None,
-                 wait_for_ready: Optional[bool] = None,
-                 timeout: Optional[float] = None,
-                 metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None) -> Iterator[ResponseType]:
+def unary_stream(
+        request: RequestType,
+        target: str,
+        method: str,
+        request_serializer: Optional[Callable[[Any], bytes]] = None,
+        request_deserializer: Optional[Callable[[bytes], Any]] = None,
+        options: Sequence[Tuple[AnyStr, AnyStr]] = (),
+        channel_credentials: Optional[grpc.ChannelCredentials] = None,
+        call_credentials: Optional[grpc.CallCredentials] = None,
+        compression: Optional[grpc.Compression] = None,
+        wait_for_ready: Optional[bool] = None,
+        timeout: Optional[float] = None,
+        metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
+) -> Iterator[ResponseType]:
     """Invokes a unary-stream RPC without an explicitly specified channel.
 
     This is backed by a per-process cache of channels. Channels are evicted
@@ -210,7 +234,7 @@ def unary_stream(request: RequestType,
     The default eviction period is 10 minutes. One may set the environment
     variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this.
 
-    The default maximum maximum number of channels is 256. One may set the
+    The default maximum number of channels is 256. One may set the
     environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure
     this.
 
@@ -243,8 +267,11 @@ def unary_stream(request: RequestType,
     Returns:
       An iterator of responses.
     """
-    channel = ChannelCache.get().get_channel(target, options, channel_credentials, compression)
-    multicallable = channel.unary_stream(method, request_serializer, request_deserializer)
+    channel_credentials = channel_credentials or grpc.local_channel_credentials()
+    channel = ChannelCache.get().get_channel(target, options,
+                                             channel_credentials, compression)
+    multicallable = channel.unary_stream(method, request_serializer,
+                                         request_deserializer)
     return multicallable(request,
                          metadata=metadata,
                          wait_for_ready=wait_for_ready,
@@ -252,18 +279,20 @@ def unary_stream(request: RequestType,
                          timeout=timeout)
 
 
-def stream_unary(request_iterator: Iterator[RequestType],
-                 target: str,
-                 method: str,
-                 request_serializer: Optional[Callable[[Any], bytes]] = None,
-                 request_deserializer: Optional[Callable[[bytes], Any]] = None,
-                 options: Sequence[Tuple[AnyStr, AnyStr]] = (),
-                 channel_credentials: Optional[grpc.ChannelCredentials] = None,
-                 call_credentials: Optional[grpc.CallCredentials] = None,
-                 compression: Optional[grpc.Compression] = None,
-                 wait_for_ready: Optional[bool] = None,
-                 timeout: Optional[float] = None,
-                 metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None) -> ResponseType:
+def stream_unary(
+        request_iterator: Iterator[RequestType],
+        target: str,
+        method: str,
+        request_serializer: Optional[Callable[[Any], bytes]] = None,
+        request_deserializer: Optional[Callable[[bytes], Any]] = None,
+        options: Sequence[Tuple[AnyStr, AnyStr]] = (),
+        channel_credentials: Optional[grpc.ChannelCredentials] = None,
+        call_credentials: Optional[grpc.CallCredentials] = None,
+        compression: Optional[grpc.Compression] = None,
+        wait_for_ready: Optional[bool] = None,
+        timeout: Optional[float] = None,
+        metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
+) -> ResponseType:
     """Invokes a stream-unary RPC without an explicitly specified channel.
 
     This is backed by a per-process cache of channels. Channels are evicted
@@ -273,7 +302,7 @@ def stream_unary(request_iterator: Iterator[RequestType],
     The default eviction period is 10 minutes. One may set the environment
     variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this.
 
-    The default maximum maximum number of channels is 256. One may set the
+    The default maximum number of channels is 256. One may set the
     environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure
     this.
 
@@ -306,8 +335,11 @@ def stream_unary(request_iterator: Iterator[RequestType],
     Returns:
       The response to the RPC.
     """
-    channel = ChannelCache.get().get_channel(target, options, channel_credentials, compression)
-    multicallable = channel.stream_unary(method, request_serializer, request_deserializer)
+    channel_credentials = channel_credentials or grpc.local_channel_credentials()
+    channel = ChannelCache.get().get_channel(target, options,
+                                             channel_credentials, compression)
+    multicallable = channel.stream_unary(method, request_serializer,
+                                         request_deserializer)
     return multicallable(request_iterator,
                          metadata=metadata,
                          wait_for_ready=wait_for_ready,
@@ -315,18 +347,20 @@ def stream_unary(request_iterator: Iterator[RequestType],
                          timeout=timeout)
 
 
-def stream_stream(request_iterator: Iterator[RequestType],
-                  target: str,
-                  method: str,
-                  request_serializer: Optional[Callable[[Any], bytes]] = None,
-                  request_deserializer: Optional[Callable[[bytes], Any]] = None,
-                  options: Sequence[Tuple[AnyStr, AnyStr]] = (),
-                  channel_credentials: Optional[grpc.ChannelCredentials] = None,
-                  call_credentials: Optional[grpc.CallCredentials] = None,
-                  compression: Optional[grpc.Compression] = None,
-                  wait_for_ready: Optional[bool] = None,
-                  timeout: Optional[float] = None,
-                  metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None) -> Iterator[ResponseType]:
+def stream_stream(
+        request_iterator: Iterator[RequestType],
+        target: str,
+        method: str,
+        request_serializer: Optional[Callable[[Any], bytes]] = None,
+        request_deserializer: Optional[Callable[[bytes], Any]] = None,
+        options: Sequence[Tuple[AnyStr, AnyStr]] = (),
+        channel_credentials: Optional[grpc.ChannelCredentials] = None,
+        call_credentials: Optional[grpc.CallCredentials] = None,
+        compression: Optional[grpc.Compression] = None,
+        wait_for_ready: Optional[bool] = None,
+        timeout: Optional[float] = None,
+        metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None
+) -> Iterator[ResponseType]:
     """Invokes a stream-stream RPC without an explicitly specified channel.
 
     This is backed by a per-process cache of channels. Channels are evicted
@@ -336,7 +370,7 @@ def stream_stream(request_iterator: Iterator[RequestType],
     The default eviction period is 10 minutes. One may set the environment
     variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this.
 
-    The default maximum maximum number of channels is 256. One may set the
+    The default maximum number of channels is 256. One may set the
     environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure
     this.
 
@@ -369,8 +403,11 @@ def stream_stream(request_iterator: Iterator[RequestType],
     Returns:
       An iterator of responses.
     """
-    channel = ChannelCache.get().get_channel(target, options, channel_credentials, compression)
-    multicallable = channel.stream_stream(method, request_serializer, request_deserializer)
+    channel_credentials = channel_credentials or grpc.local_channel_credentials()
+    channel = ChannelCache.get().get_channel(target, options,
+                                             channel_credentials, compression)
+    multicallable = channel.stream_stream(method, request_serializer,
+                                          request_deserializer)
     return multicallable(request_iterator,
                          metadata=metadata,
                          wait_for_ready=wait_for_ready,

src/python/grpcio_tests/tests/unit/py3_only/_simple_stubs_test.py

@@ -32,6 +32,7 @@ from typing import Callable, Optional
 import test_common
 import grpc
 
+_REQUEST = b"0000"
 
 _CACHE_EPOCHS = 8
 _CACHE_TRIALS = 6
@@ -67,6 +68,7 @@ def _stream_stream_handler(request_iterator, context):
 
 
 class _GenericHandler(grpc.GenericRpcHandler):
+
     def service(self, handler_call_details):
         if handler_call_details.method == _UNARY_UNARY:
             return grpc.unary_unary_rpc_method_handler(_unary_unary_handler)
@@ -125,11 +127,12 @@ class SimpleStubsTest(unittest.TestCase):
                 runs.append(_time_invocation(lambda: to_check(text)))
             initial_runs.append(runs[0])
             cached_runs.extend(runs[1:])
-        average_cold = sum((run for run in initial_runs), datetime.timedelta()) / len(initial_runs)
-        average_warm = sum((run for run in cached_runs), datetime.timedelta()) / len(cached_runs)
+        average_cold = sum((run for run in initial_runs),
+                           datetime.timedelta()) / len(initial_runs)
+        average_warm = sum((run for run in cached_runs),
+                           datetime.timedelta()) / len(cached_runs)
         self.assertLess(average_warm, average_cold)
 
-
     def assert_eventually(self,
                           predicate: Callable[[], bool],
                           *,
@@ -148,106 +151,121 @@ class SimpleStubsTest(unittest.TestCase):
     def test_unary_unary_insecure(self):
         with _server(None) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            response = grpc.unary_unary(request, target, _UNARY_UNARY)
-            self.assertEqual(request, response)
+            response = grpc.unary_unary(
+                        _REQUEST,
+                        target,
+                        _UNARY_UNARY,
+                        channel_credentials=grpc.insecure_channel_credentials())
+            self.assertEqual(_REQUEST, response)
 
     def test_unary_unary_secure(self):
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            response = grpc.unary_unary(request,
-                                        target,
-                                        _UNARY_UNARY,
-                                        channel_credentials=grpc.local_channel_credentials())
-            self.assertEqual(request, response)
+            response = grpc.unary_unary(
+                _REQUEST,
+                target,
+                _UNARY_UNARY,
+                channel_credentials=grpc.local_channel_credentials())
+            self.assertEqual(_REQUEST, response)
+
+    def test_channel_credentials_default(self):
+        with _server(grpc.local_server_credentials()) as (_, port):
+            target = f'localhost:{port}'
+            response = grpc.unary_unary(
+                _REQUEST,
+                target,
+                _UNARY_UNARY)
+            self.assertEqual(_REQUEST, response)
 
     def test_channels_cached(self):
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
             test_name = inspect.stack()[0][3]
-            args = (request, target, _UNARY_UNARY)
+            args = (_REQUEST, target, _UNARY_UNARY)
             kwargs = {"channel_credentials": grpc.local_channel_credentials()}
+
             def _invoke(seed: str):
                 run_kwargs = dict(kwargs)
                 run_kwargs["options"] = ((test_name + seed, ""),)
                 grpc.unary_unary(*args, **run_kwargs)
+
             self.assert_cached(_invoke)
 
     def test_channels_evicted(self):
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            response = grpc.unary_unary(request,
-                                        target,
-                                        _UNARY_UNARY,
-                                        channel_credentials=grpc.local_channel_credentials())
+            response = grpc.unary_unary(
+                _REQUEST,
+                target,
+                _UNARY_UNARY,
+                channel_credentials=grpc.local_channel_credentials())
             self.assert_eventually(
-                lambda: grpc._simple_stubs.ChannelCache.get()._test_only_channel_count() == 0,
-                message=lambda: f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} remain")
+                lambda: grpc._simple_stubs.ChannelCache.get(
+                )._test_only_channel_count() == 0,
+                message=lambda:
+                f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} remain"
+            )
 
     def test_total_channels_enforced(self):
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
             for i in range(99):
                 # Ensure we get a new channel each time.
                 options = (("foo", str(i)),)
                 # Send messages at full blast.
-                grpc.unary_unary(request,
-                                 target,
-                                 _UNARY_UNARY,
-                                 options=options,
-                                 channel_credentials=grpc.local_channel_credentials())
+                grpc.unary_unary(
+                    _REQUEST,
+                    target,
+                    _UNARY_UNARY,
+                    options=options,
+                    channel_credentials=grpc.local_channel_credentials())
                 self.assert_eventually(
-                    lambda: grpc._simple_stubs.ChannelCache.get()._test_only_channel_count() <= _MAXIMUM_CHANNELS + 1,
-                    message=lambda: f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} channels remain")
+                    lambda: grpc._simple_stubs.ChannelCache.get(
+                    )._test_only_channel_count() <= _MAXIMUM_CHANNELS + 1,
+                    message=lambda:
+                    f"{grpc._simple_stubs.ChannelCache.get()._test_only_channel_count()} channels remain"
+                )
 
     def test_unary_stream(self):
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            for response in grpc.unary_stream(request,
-                                             target,
-                                             _UNARY_STREAM,
-                                             channel_credentials=grpc.local_channel_credentials()):
-                self.assertEqual(request, response)
+            for response in grpc.unary_stream(
+                    _REQUEST,
+                    target,
+                    _UNARY_STREAM,
+                    channel_credentials=grpc.local_channel_credentials()):
+                self.assertEqual(_REQUEST, response)
 
     def test_stream_unary(self):
+
         def request_iter():
             for _ in range(_CLIENT_REQUEST_COUNT):
-                yield request
+                yield _REQUEST
+
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            response = grpc.stream_unary(request_iter(),
-                                         target,
-                                         _STREAM_UNARY,
-                                         channel_credentials=grpc.local_channel_credentials())
-            self.assertEqual(request, response)
+            response = grpc.stream_unary(
+                request_iter(),
+                target,
+                _STREAM_UNARY,
+                channel_credentials=grpc.local_channel_credentials())
+            self.assertEqual(_REQUEST, response)
 
     def test_stream_stream(self):
+
         def request_iter():
             for _ in range(_CLIENT_REQUEST_COUNT):
-                yield request
+                yield _REQUEST
+
         with _server(grpc.local_server_credentials()) as (_, port):
             target = f'localhost:{port}'
-            request = b'0000'
-            for response in grpc.stream_stream(request_iter(),
-                                               target,
-                                               _STREAM_STREAM,
-                                               channel_credentials=grpc.local_channel_credentials()):
-                self.assertEqual(request, response)
-
-
-    # TODO: Test request_serializer
-    # TODO: Test request_deserializer
-    # TODO: Test channel_credentials
-    # TODO: Test call_credentials
-    # TODO: Test compression
-    # TODO: Test wait_for_ready
-    # TODO: Test metadata
+            for response in grpc.stream_stream(
+                    request_iter(),
+                    target,
+                    _STREAM_STREAM,
+                    channel_credentials=grpc.local_channel_credentials()):
+                self.assertEqual(_REQUEST, response)
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO)