Allow additive changes to protos w/o forcing user implementation

1 files changed, 41 insertions, 0 deletions

diff --git a/src/python/grpcio/tests/protoc_plugin/beta_python_plugin_test.py b/src/python/grpcio/tests/protoc_plugin/beta_python_plugin_test.py
index 6fba3d4271..3dc3042e38 100644
--- a/src/python/grpcio/tests/protoc_plugin/beta_python_plugin_test.py
+++ b/src/python/grpcio/tests/protoc_plugin/beta_python_plugin_test.py
@@ -45,6 +45,7 @@ import unittest
 from six import moves
 
 from grpc.beta import implementations
+from grpc.beta import interfaces
 from grpc.framework.foundation import future
 from grpc.framework.interfaces.face import face
 from tests.unit.framework.common import test_constants
@@ -178,6 +179,36 @@ def _CreateService(test_pb2):
   server.stop(0)
 
 
+@contextlib.contextmanager
+def _CreateIncompleteService(test_pb2):
+  """Provides a servicer backend that fails to implement methods and its stub.
+
+  The servicer is just the implementation of the actual servicer passed to the
+  face player of the python RPC implementation; the two are detached.
+
+  Args:
+    test_pb2: The test_pb2 module generated by this test.
+
+  Yields:
+    A (servicer_methods, stub) pair where servicer_methods is the back-end of
+      the service bound to the stub and and stub is the stub on which to invoke
+      RPCs.
+  """
+  servicer_methods = _ServicerMethods(test_pb2)
+
+  class Servicer(getattr(test_pb2, SERVICER_IDENTIFIER)):
+    pass
+
+  servicer = Servicer()
+  server = getattr(test_pb2, SERVER_FACTORY_IDENTIFIER)(servicer)
+  port = server.add_insecure_port('[::]:0')
+  server.start()
+  channel = implementations.insecure_channel('localhost', port)
+  stub = getattr(test_pb2, STUB_FACTORY_IDENTIFIER)(channel)
+  yield servicer_methods, stub
+  server.stop(0)
+
+
 def _streaming_input_request_iterator(test_pb2):
   for _ in range(3):
     request = test_pb2.StreamingInputCallRequest()
@@ -264,6 +295,16 @@ class PythonPluginTest(unittest.TestCase):
     with _CreateService(test_pb2) as (servicer, stub):
       request = test_pb2.SimpleRequest(response_size=13)
 
+  def testIncompleteServicer(self):
+    import protoc_plugin_test_pb2 as test_pb2
+    moves.reload_module(test_pb2)
+    with _CreateIncompleteService(test_pb2) as (servicer, stub):
+      request = test_pb2.SimpleRequest(response_size=13)
+      try:
+        response = stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
+      except face.AbortionError as error:
+        self.assertEqual(interfaces.StatusCode.UNIMPLEMENTED, error.code)
+
   def testUnaryCall(self):
     import protoc_plugin_test_pb2 as test_pb2  # pylint: disable=g-import-not-at-top
     moves.reload_module(test_pb2)