Merge pull request #4923 from nathanielmanistaatgoogle/tests

Python test improvements

src/python/grpcio/grpc/framework/foundation/logging_pool.py

-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -29,7 +29,6 @@
 
 """A thread pool that logs exceptions raised by tasks executed within it."""
 
-import functools
 import logging
 
 from concurrent import futures
@@ -37,12 +36,12 @@ from concurrent import futures
 
 def _wrap(behavior):
   """Wraps an arbitrary callable behavior in exception-logging."""
-  @functools.wraps(behavior)
   def _wrapping(*args, **kwargs):
     try:
       return behavior(*args, **kwargs)
     except Exception as e:
-      logging.exception('Unexpected exception from task run in logging pool!')
+      logging.exception(
+          'Unexpected exception from %s executed in logging pool!', behavior)
       raise
   return _wrapping
 

src/python/grpcio/tests/unit/framework/foundation/_logging_pool_test.py

-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -29,6 +29,7 @@
 
 """Tests for grpc.framework.foundation.logging_pool."""
 
+import threading
 import unittest
 
 from grpc.framework.foundation import logging_pool
@@ -36,6 +37,21 @@ from grpc.framework.foundation import logging_pool
 _POOL_SIZE = 16
 
 
+class _CallableObject(object):
+
+  def __init__(self):
+    self._lock = threading.Lock()
+    self._passed_values = []
+
+  def __call__(self, value):
+    with self._lock:
+      self._passed_values.append(value)
+
+  def passed_values(self):
+    with self._lock:
+      return tuple(self._passed_values)
+
+
 class LoggingPoolTest(unittest.TestCase):
 
   def testUpAndDown(self):
@@ -59,6 +75,14 @@ class LoggingPoolTest(unittest.TestCase):
 
     self.assertIsNotNone(raised_exception)
 
+  def testCallableObjectExecuted(self):
+    callable_object = _CallableObject()
+    passed_object = object()
+    with logging_pool.pool(_POOL_SIZE) as pool:
+      future = pool.submit(callable_object, passed_object)
+    self.assertIsNone(future.result())
+    self.assertSequenceEqual((passed_object,), callable_object.passed_values())
+
 
 if __name__ == '__main__':
   unittest.main(verbosity=2)

src/python/grpcio/tests/unit/framework/interfaces/face/_blocking_invocation_inline_service.py

-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -30,9 +30,12 @@
 """Test code for the Face layer of RPC Framework."""
 
 import abc
+import itertools
 import unittest
+from concurrent import futures
 
 # test_interfaces is referenced from specification in this module.
+from grpc.framework.foundation import logging_pool
 from grpc.framework.interfaces.face import face
 from tests.unit.framework.common import test_constants
 from tests.unit.framework.common import test_control
@@ -139,13 +142,50 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         test_messages.verify(second_request, second_response, self)
 
-  @unittest.skip('Parallel invocations impossible with blocking control flow!')
   def testParallelInvocations(self):
-    raise NotImplementedError()
+    pool = logging_pool.pool(test_constants.PARALLELISM)
+    for (group, method), test_messages_sequence in (
+        self._digest.unary_unary_messages_sequences.iteritems()):
+      for test_messages in test_messages_sequence:
+        requests = []
+        response_futures = []
+        for _ in range(test_constants.PARALLELISM):
+          request = test_messages.request()
+          response_future = pool.submit(
+              self._invoker.blocking(group, method), request,
+              test_constants.LONG_TIMEOUT)
+          requests.append(request)
+          response_futures.append(response_future)
+
+        responses = [
+            response_future.result() for response_future in response_futures]
+
+        for request, response in zip(requests, responses):
+          test_messages.verify(request, response, self)
+    pool.shutdown(wait=True)
 
-  @unittest.skip('Parallel invocations impossible with blocking control flow!')
   def testWaitingForSomeButNotAllParallelInvocations(self):
-    raise NotImplementedError()
+    pool = logging_pool.pool(test_constants.PARALLELISM)
+    for (group, method), test_messages_sequence in (
+        self._digest.unary_unary_messages_sequences.iteritems()):
+      for test_messages in test_messages_sequence:
+        requests = []
+        response_futures_to_indices = {}
+        for index in range(test_constants.PARALLELISM):
+          request = test_messages.request()
+          response_future = pool.submit(
+              self._invoker.blocking(group, method), request,
+              test_constants.LONG_TIMEOUT)
+          requests.append(request)
+          response_futures_to_indices[response_future] = index
+
+        some_completed_response_futures_iterator = itertools.islice(
+            futures.as_completed(response_futures_to_indices),
+            test_constants.PARALLELISM / 2)
+        for response_future in some_completed_response_futures_iterator:
+          index = response_futures_to_indices[response_future]
+          test_messages.verify(requests[index], response_future.result(), self)
+    pool.shutdown(wait=True)
 
   @unittest.skip('Cancellation impossible with blocking control flow!')
   def testCancelledUnaryRequestUnaryResponse(self):

src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py

@@ -31,8 +31,10 @@
 
 import abc
 import contextlib
+import itertools
 import threading
 import unittest
+from concurrent import futures
 
 # test_interfaces is referenced from specification in this module.
 from grpc.framework.foundation import logging_pool
@@ -219,6 +221,23 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         test_messages.verify(second_request, second_response, self)
 
+  def testParallelInvocations(self):
+    for (group, method), test_messages_sequence in (
+        self._digest.unary_unary_messages_sequences.iteritems()):
+      for test_messages in test_messages_sequence:
+        first_request = test_messages.request()
+        second_request = test_messages.request()
+
+        first_response_future = self._invoker.future(group, method)(
+            first_request, test_constants.LONG_TIMEOUT)
+        second_response_future = self._invoker.future(group, method)(
+            second_request, test_constants.LONG_TIMEOUT)
+        first_response = first_response_future.result()
+        second_response = second_response_future.result()
+
+        test_messages.verify(first_request, first_response, self)
+        test_messages.verify(second_request, second_response, self)
+
     for (group, method), test_messages_sequence in (
         self._digest.unary_unary_messages_sequences.iteritems()):
       for test_messages in test_messages_sequence:
@@ -237,26 +256,28 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
         for request, response in zip(requests, responses):
           test_messages.verify(request, response, self)
 
-  def testParallelInvocations(self):
+  def testWaitingForSomeButNotAllParallelInvocations(self):
+    pool = logging_pool.pool(test_constants.PARALLELISM)
     for (group, method), test_messages_sequence in (
         self._digest.unary_unary_messages_sequences.iteritems()):
       for test_messages in test_messages_sequence:
-        first_request = test_messages.request()
-        second_request = test_messages.request()
-
-        first_response_future = self._invoker.future(group, method)(
-            first_request, test_constants.LONG_TIMEOUT)
-        second_response_future = self._invoker.future(group, method)(
-            second_request, test_constants.LONG_TIMEOUT)
-        first_response = first_response_future.result()
-        second_response = second_response_future.result()
-
-        test_messages.verify(first_request, first_response, self)
-        test_messages.verify(second_request, second_response, self)
-
-  @unittest.skip('TODO(nathaniel): implement.')
-  def testWaitingForSomeButNotAllParallelInvocations(self):
-    raise NotImplementedError()
+        requests = []
+        response_futures_to_indices = {}
+        for index in range(test_constants.PARALLELISM):
+          request = test_messages.request()
+          inner_response_future = self._invoker.future(group, method)(
+              request, test_constants.LONG_TIMEOUT)
+          outer_response_future = pool.submit(inner_response_future.result)
+          requests.append(request)
+          response_futures_to_indices[outer_response_future] = index
+
+        some_completed_response_futures_iterator = itertools.islice(
+            futures.as_completed(response_futures_to_indices),
+            test_constants.PARALLELISM / 2)
+        for response_future in some_completed_response_futures_iterator:
+          index = response_futures_to_indices[response_future]
+          test_messages.verify(requests[index], response_future.result(), self)
+    pool.shutdown(wait=True)
 
   def testCancelledUnaryRequestUnaryResponse(self):
     for (group, method), test_messages_sequence in (

src/python/grpcio/tests/unit/framework/interfaces/face/_receiver.py

-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -76,7 +76,7 @@ class Receiver(face.ResponseReceiver):
   def unary_response(self):
     with self._condition:
       if self._abortion is not None:
-        raise AssertionError('Aborted with abortion "%s"!' % self._abortion)
+        raise AssertionError('Aborted: "{}"!'.format(self._abortion))
       elif len(self._responses) != 1:
         raise AssertionError(
             '%d responses received, not exactly one!', len(self._responses))
@@ -88,7 +88,7 @@ class Receiver(face.ResponseReceiver):
       if self._abortion is None:
         return list(self._responses)
       else:
-        raise AssertionError('Aborted with abortion "%s"!' % self._abortion)
+        raise AssertionError('Aborted: "{}"!'.format(self._abortion))
 
   def abortion(self):
     with self._condition: