Add test and fake LRU cache interface

.gitignore

@@ -115,11 +115,7 @@ Podfile.lock
 .idea/
 
 # Bazel files
-bazel-bin
-bazel-genfiles
-bazel-grpc
-bazel-out
-bazel-testlogs
+bazel-*
 bazel_format_virtual_environment/
 tools/bazel-*
 

src/python/grpcio/grpc/BUILD.bazel

@@ -15,6 +15,7 @@ py_library(
         ":interceptor",
         ":server",
         ":compression",
+        ":_simple_stubs",
         "//src/python/grpcio/grpc/_cython:cygrpc",
         "//src/python/grpcio/grpc/experimental",
         "//src/python/grpcio/grpc/framework",
@@ -85,3 +86,8 @@ py_library(
         ":common",
     ],
 )
+
+py_library(
+    name = "_simple_stubs",
+    srcs = ["_simple_stubs.py"],
+)

src/python/grpcio/grpc/__init__.py

@@ -1949,6 +1949,7 @@ 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
@@ -2031,8 +2032,13 @@ __all__ = (
     'secure_channel',
     'intercept_channel',
     'server',
+    'unary_unary',
 )
 
+if sys.version_info[0] > 2:
+    from grpc._simple_stubs import unary_unary
+    __all__ = __all__ + (unary_unary,)
+
 ############################### Extension Shims ################################
 
 # Here to maintain backwards compatibility; avoid using these in new code!

src/python/grpcio/grpc/_simple_stubs.py

@@ -0,0 +1,46 @@
+# TODO: Flowerbox.
+
+import grpc
+from typing import Any, Callable, Optional, Sequence, Text, Tuple, Union
+
+
+def _get_cached_channel(target: Text,
+                        options: Sequence[Tuple[Text, Text]],
+                        channel_credentials: Optional[grpc.ChannelCredentials],
+                        compression: Optional[grpc.Compression]) -> grpc.Channel:
+    # TODO: Actually cache.
+    if channel_credentials is None:
+        return grpc.insecure_channel(target,
+                                     options=options,
+                                     compression=compression)
+    else:
+        return grpc.secure_channel(target,
+                                   credentials=channel_credentials,
+                                   options=options,
+                                   compression=compression)
+
+def unary_unary(request: Any,
+                target: Text,
+                method: Text,
+                request_serializer: Optional[Callable[[Any], bytes]] = None,
+                request_deserializer: Optional[Callable[[bytes], Any]] = None,
+                options: Sequence[Tuple[Text, Text]] = (),
+                # 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,
+                metadata: Optional[Sequence[Tuple[Text, Union[Text, bytes]]]] = None) -> Any:
+    """Invokes a unary RPC without an explicitly specified channel.
+
+    This is backed by an LRU cache of channels evicted by a background thread
+    on a periodic basis.
+
+    TODO: Document the parameters and return value.
+    """
+    channel = _get_cached_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,
+                         credentials=call_credentials)

src/python/grpcio_tests/tests/tests.json

@@ -70,6 +70,7 @@
   "unit._server_test.ServerTest",
   "unit._server_wait_for_termination_test.ServerWaitForTerminationTest",
   "unit._session_cache_test.SSLSessionCacheTest",
+  "unit._simple_stubs_test.SimpleStubsTest",
   "unit._signal_handling_test.SignalHandlingTest",
   "unit._version_test.VersionTest",
   "unit.beta._beta_features_test.BetaFeaturesTest",

src/python/grpcio_tests/tests/unit/BUILD.bazel

@@ -32,6 +32,7 @@ GRPCIO_TESTS_UNIT = [
     "_resource_exhausted_test.py",
     "_rpc_test.py",
     "_signal_handling_test.py",
+    "_simple_stubs_test.py",
     # TODO(ghostwriternr): To be added later.
     # "_server_ssl_cert_config_test.py",
     "_server_test.py",

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

@@ -0,0 +1,65 @@
+# Copyright 2020 The gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for Simple Stubs."""
+
+import unittest
+import sys
+
+import logging
+
+import grpc
+import test_common
+
+
+_UNARY_UNARY = "/test/UnaryUnary"
+
+
+def _unary_unary_handler(request, context):
+    return request
+
+
+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)
+        else:
+            raise NotImplementedError()
+
+
+@unittest.skipIf(sys.version_info[0] < 3, "Unsupported on Python 2.")
+class SimpleStubsTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super(SimpleStubsTest, cls).setUpClass()
+        cls._server = test_common.test_server()
+        cls._port = cls._server.add_insecure_port('[::]:0')
+        cls._server.add_generic_rpc_handlers((_GenericHandler(),))
+        cls._server.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._server.stop(None)
+        super(SimpleStubsTest, cls).tearDownClass()
+
+    def test_unary_unary(self):
+        target = f'localhost:{self._port}'
+        request = b'0000'
+        response = grpc.unary_unary(request, target, _UNARY_UNARY)
+        self.assertEqual(request, response)
+
+if __name__ == "__main__":
+    logging.basicConfig()
+    unittest.main(verbosity=2)
+
+