Merge pull request #8686 from thunderboltsid/issue-8231-re

"Handle" non-iterator objects in consume_request_iterator by failing the RPC.

7 files changed, 265 insertions, 7 deletions

diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py
index 1298cfbb6f..f07142aad3 100644
--- a/src/python/grpcio/grpc/_channel.py
+++ b/src/python/grpcio/grpc/_channel.py
@@ -32,6 +32,7 @@
 import sys
 import threading
 import time
+import logging
 
 import grpc
 from grpc import _common
@@ -197,7 +198,16 @@ def _consume_request_iterator(
   event_handler = _event_handler(state, call, None)
 
   def consume_request_iterator():
-    for request in request_iterator:
+    while True:
+      try:
+        request = next(request_iterator)
+      except StopIteration:
+        break
+      except Exception as e:
+        logging.exception("Exception iterating requests!")
+        call.cancel()
+        _abort(state, grpc.StatusCode.UNKNOWN, "Exception iterating requests!")
+        return
       serialized_request = _common.serialize(request, request_serializer)
       with state.condition:
         if state.code is None and not state.cancelled:
diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json
index 2ac51ac542..d47631cf75 100644
--- a/src/python/grpcio_tests/tests/tests.json
+++ b/src/python/grpcio_tests/tests/tests.json
@@ -26,6 +26,7 @@
   "_implementations_test.ChannelCredentialsTest",
   "_insecure_interop_test.InsecureInteropTest",
   "_invalid_metadata_test.InvalidMetadataTest",
+  "_invocation_defects_test.InvocationDefectsTest",
   "_logging_pool_test.LoggingPoolTest",
   "_metadata_code_details_test.MetadataCodeDetailsTest",
   "_metadata_test.MetadataTest",
diff --git a/src/python/grpcio_tests/tests/unit/_compression_test.py b/src/python/grpcio_tests/tests/unit/_compression_test.py
index 83b9109466..4d3f02e917 100644
--- a/src/python/grpcio_tests/tests/unit/_compression_test.py
+++ b/src/python/grpcio_tests/tests/unit/_compression_test.py
@@ -125,7 +125,7 @@ class CompressionTest(unittest.TestCase):
     compressed_channel = grpc.insecure_channel('localhost:%d' % self._port,
         options=[('grpc.default_compression_algorithm', 1)])
     multi_callable = compressed_channel.stream_stream(_STREAM_STREAM)
-    call = multi_callable([request] * test_constants.STREAM_LENGTH)
+    call = multi_callable(iter([request] * test_constants.STREAM_LENGTH))
     for response in call:
       self.assertEqual(request, response)
 
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 131f6e9452..69f4689279 100644
--- a/src/python/grpcio_tests/tests/unit/_empty_message_test.py
+++ b/src/python/grpcio_tests/tests/unit/_empty_message_test.py
@@ -123,12 +123,12 @@ class EmptyMessageTest(unittest.TestCase):
 
   def testStreamUnary(self):
     response = self._channel.stream_unary(_STREAM_UNARY)(
-        [_REQUEST] * test_constants.STREAM_LENGTH)
+        iter([_REQUEST] * test_constants.STREAM_LENGTH))
     self.assertEqual(_RESPONSE, response)
 
   def testStreamStream(self):
     response_iterator = self._channel.stream_stream(_STREAM_STREAM)(
-        [_REQUEST] * test_constants.STREAM_LENGTH)
+        iter([_REQUEST] * test_constants.STREAM_LENGTH))
     self.assertSequenceEqual(
         [_RESPONSE] * test_constants.STREAM_LENGTH, list(response_iterator))
 
diff --git a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py
index b33802bf57..777527137f 100644
--- a/src/python/grpcio_tests/tests/unit/_exit_scenarios.py
+++ b/src/python/grpcio_tests/tests/unit/_exit_scenarios.py
@@ -240,7 +240,7 @@ if __name__ == '__main__':
       multi_callable = channel.stream_unary(method)
       future = multi_callable.future(infinite_request_iterator())
       result, call = multi_callable.with_call(
-          [REQUEST] * test_constants.STREAM_LENGTH)
+          iter([REQUEST] * test_constants.STREAM_LENGTH))
     elif (args.scenario == IN_FLIGHT_STREAM_STREAM_CALL or
           args.scenario == IN_FLIGHT_PARTIAL_STREAM_STREAM_CALL):
       multi_callable = channel.stream_stream(method)
diff --git a/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py b/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py
new file mode 100644
index 0000000000..4312679bb9
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_invocation_defects_test.py
@@ -0,0 +1,247 @@
+# Copyright 2016, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import itertools
+import threading
+import unittest
+from concurrent import futures
+
+import grpc
+from grpc.framework.foundation import logging_pool
+
+from tests.unit.framework.common import test_constants
+from tests.unit.framework.common import test_control
+
+_SERIALIZE_REQUEST = lambda bytestring: bytestring * 2
+_DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:]
+_SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3
+_DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3]
+
+_UNARY_UNARY = '/test/UnaryUnary'
+_UNARY_STREAM = '/test/UnaryStream'
+_STREAM_UNARY = '/test/StreamUnary'
+_STREAM_STREAM = '/test/StreamStream'
+
+
+class _Callback(object):
+  def __init__(self):
+    self._condition = threading.Condition()
+    self._value = None
+    self._called = False
+
+  def __call__(self, value):
+    with self._condition:
+      self._value = value
+      self._called = True
+      self._condition.notify_all()
+
+  def value(self):
+    with self._condition:
+      while not self._called:
+        self._condition.wait()
+      return self._value
+
+
+class _Handler(object):
+  def __init__(self, control):
+    self._control = control
+
+  def handle_unary_unary(self, request, servicer_context):
+    self._control.control()
+    if servicer_context is not None:
+      servicer_context.set_trailing_metadata((('testkey', 'testvalue',),))
+    return request
+
+  def handle_unary_stream(self, request, servicer_context):
+    for _ in range(test_constants.STREAM_LENGTH):
+      self._control.control()
+      yield request
+    self._control.control()
+    if servicer_context is not None:
+      servicer_context.set_trailing_metadata((('testkey', 'testvalue',),))
+
+  def handle_stream_unary(self, request_iterator, servicer_context):
+    if servicer_context is not None:
+      servicer_context.invocation_metadata()
+    self._control.control()
+    response_elements = []
+    for request in request_iterator:
+      self._control.control()
+      response_elements.append(request)
+    self._control.control()
+    if servicer_context is not None:
+      servicer_context.set_trailing_metadata((('testkey', 'testvalue',),))
+    return b''.join(response_elements)
+
+  def handle_stream_stream(self, request_iterator, servicer_context):
+    self._control.control()
+    if servicer_context is not None:
+      servicer_context.set_trailing_metadata((('testkey', 'testvalue',),))
+    for request in request_iterator:
+      self._control.control()
+      yield request
+    self._control.control()
+
+
+class _MethodHandler(grpc.RpcMethodHandler):
+  def __init__(
+    self, request_streaming, response_streaming, request_deserializer,
+    response_serializer, unary_unary, unary_stream, stream_unary,
+    stream_stream):
+    self.request_streaming = request_streaming
+    self.response_streaming = response_streaming
+    self.request_deserializer = request_deserializer
+    self.response_serializer = response_serializer
+    self.unary_unary = unary_unary
+    self.unary_stream = unary_stream
+    self.stream_unary = stream_unary
+    self.stream_stream = stream_stream
+
+
+class _GenericHandler(grpc.GenericRpcHandler):
+  def __init__(self, handler):
+    self._handler = handler
+
+  def service(self, handler_call_details):
+    if handler_call_details.method == _UNARY_UNARY:
+      return _MethodHandler(
+        False, False, None, None, self._handler.handle_unary_unary, None,
+        None, None)
+    elif handler_call_details.method == _UNARY_STREAM:
+      return _MethodHandler(
+        False, True, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None,
+        self._handler.handle_unary_stream, None, None)
+    elif handler_call_details.method == _STREAM_UNARY:
+      return _MethodHandler(
+        True, False, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, None,
+        self._handler.handle_stream_unary, None)
+    elif handler_call_details.method == _STREAM_STREAM:
+      return _MethodHandler(
+        True, True, None, None, None, None, None,
+        self._handler.handle_stream_stream)
+    else:
+      return None
+
+
+class FailAfterFewIterationsCounter(object):
+    def __init__(self, high, bytestring):
+        self._current = 0
+        self._high = high
+        self._bytestring = bytestring
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._current >= self._high:
+            raise Exception("This is a deliberate failure in a unit test.")
+        else:
+            self._current += 1
+            return self._bytestring
+
+
+def _unary_unary_multi_callable(channel):
+  return channel.unary_unary(_UNARY_UNARY)
+
+
+def _unary_stream_multi_callable(channel):
+  return channel.unary_stream(
+    _UNARY_STREAM,
+    request_serializer=_SERIALIZE_REQUEST,
+    response_deserializer=_DESERIALIZE_RESPONSE)
+
+
+def _stream_unary_multi_callable(channel):
+  return channel.stream_unary(
+    _STREAM_UNARY,
+    request_serializer=_SERIALIZE_REQUEST,
+    response_deserializer=_DESERIALIZE_RESPONSE)
+
+
+def _stream_stream_multi_callable(channel):
+  return channel.stream_stream(_STREAM_STREAM)
+
+
+class InvocationDefectsTest(unittest.TestCase):
+  def setUp(self):
+    self._control = test_control.PauseFailControl()
+    self._handler = _Handler(self._control)
+    self._server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
+
+    self._server = grpc.server(self._server_pool)
+    port = self._server.add_insecure_port('[::]:0')
+    self._server.add_generic_rpc_handlers((_GenericHandler(self._handler),))
+    self._server.start()
+
+    self._channel = grpc.insecure_channel('localhost:%d' % port)
+
+  def tearDown(self):
+    self._server.stop(0)
+
+  def testIterableStreamRequestBlockingUnaryResponse(self):
+    requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)]
+    multi_callable = _stream_unary_multi_callable(self._channel)
+
+    with self.assertRaises(grpc.RpcError):
+      response = multi_callable(
+        requests,
+        metadata=(('test', 'IterableStreamRequestBlockingUnaryResponse'),))
+
+  def testIterableStreamRequestFutureUnaryResponse(self):
+    requests = [b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)]
+    multi_callable = _stream_unary_multi_callable(self._channel)
+    response_future = multi_callable.future(
+      requests,
+      metadata=(
+        ('test', 'IterableStreamRequestFutureUnaryResponse'),))
+
+    with self.assertRaises(grpc.RpcError):
+      response = response_future.result()
+
+  def testIterableStreamRequestStreamResponse(self):
+    requests = [b'\x77\x58' for _ in range(test_constants.STREAM_LENGTH)]
+    multi_callable = _stream_stream_multi_callable(self._channel)
+    response_iterator = multi_callable(
+      requests,
+      metadata=(('test', 'IterableStreamRequestStreamResponse'),))
+
+    with self.assertRaises(grpc.RpcError):
+      next(response_iterator)
+
+  def testIteratorStreamRequestStreamResponse(self):
+    requests_iterator = FailAfterFewIterationsCounter(
+      test_constants.STREAM_LENGTH // 2, b'\x07\x08')
+    multi_callable = _stream_stream_multi_callable(self._channel)
+    response_iterator = multi_callable(
+      requests_iterator,
+      metadata=(('test', 'IteratorStreamRequestStreamResponse'),))
+
+    with self.assertRaises(grpc.RpcError):
+      for _ in range(test_constants.STREAM_LENGTH // 2 + 1):
+        next(response_iterator)
diff --git a/src/python/grpcio_tests/tests/unit/_metadata_test.py b/src/python/grpcio_tests/tests/unit/_metadata_test.py
index da73476929..caba53ffcc 100644
--- a/src/python/grpcio_tests/tests/unit/_metadata_test.py
+++ b/src/python/grpcio_tests/tests/unit/_metadata_test.py
@@ -193,7 +193,7 @@ class MetadataTest(unittest.TestCase):
   def testStreamUnary(self):
     multi_callable = self._channel.stream_unary(_STREAM_UNARY)
     unused_response, call = multi_callable.with_call(
-        [_REQUEST] * test_constants.STREAM_LENGTH,
+        iter([_REQUEST] * test_constants.STREAM_LENGTH),
         metadata=_CLIENT_METADATA)
     self.assertTrue(test_common.metadata_transmitted(
         _SERVER_INITIAL_METADATA, call.initial_metadata()))
@@ -202,7 +202,7 @@ class MetadataTest(unittest.TestCase):
 
   def testStreamStream(self):
     multi_callable = self._channel.stream_stream(_STREAM_STREAM)
-    call = multi_callable([_REQUEST] * test_constants.STREAM_LENGTH,
+    call = multi_callable(iter([_REQUEST] * test_constants.STREAM_LENGTH),
                           metadata=_CLIENT_METADATA)
     self.assertTrue(test_common.metadata_transmitted(
         _SERVER_INITIAL_METADATA, call.initial_metadata()))