Run Python formatting

13 files changed, 964 insertions, 972 deletions

diff --git a/src/python/grpcio_tests/tests/protoc_plugin/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/_python_plugin_test.py b/src/python/grpcio_tests/tests/protoc_plugin/_python_plugin_test.py
index 7ca2bcff38..ae5da2c3db 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/_python_plugin_test.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/_python_plugin_test.py
@@ -58,436 +58,440 @@ ADD_SERVICER_TO_SERVER_IDENTIFIER = 'add_TestServiceServicer_to_server'
 
 class _ServicerMethods(object):
 
-  def __init__(self):
-    self._condition = threading.Condition()
-    self._paused = False
-    self._fail = False
-
-  @contextlib.contextmanager
-  def pause(self):  # pylint: disable=invalid-name
-    with self._condition:
-      self._paused = True
-    yield
-    with self._condition:
-      self._paused = False
-      self._condition.notify_all()
-
-  @contextlib.contextmanager
-  def fail(self):  # pylint: disable=invalid-name
-    with self._condition:
-      self._fail = True
-    yield
-    with self._condition:
-      self._fail = False
-
-  def _control(self):  # pylint: disable=invalid-name
-    with self._condition:
-      if self._fail:
-        raise ValueError()
-      while self._paused:
-        self._condition.wait()
-
-  def UnaryCall(self, request, unused_rpc_context):
-    response = response_pb2.SimpleResponse()
-    response.payload.payload_type = payload_pb2.COMPRESSABLE
-    response.payload.payload_compressable = 'a' * request.response_size
-    self._control()
-    return response
-
-  def StreamingOutputCall(self, request, unused_rpc_context):
-    for parameter in request.response_parameters:
-      response = response_pb2.StreamingOutputCallResponse()
-      response.payload.payload_type = payload_pb2.COMPRESSABLE
-      response.payload.payload_compressable = 'a' * parameter.size
-      self._control()
-      yield response
-
-  def StreamingInputCall(self, request_iter, unused_rpc_context):
-    response = response_pb2.StreamingInputCallResponse()
-    aggregated_payload_size = 0
-    for request in request_iter:
-      aggregated_payload_size += len(request.payload.payload_compressable)
-    response.aggregated_payload_size = aggregated_payload_size
-    self._control()
-    return response
-
-  def FullDuplexCall(self, request_iter, unused_rpc_context):
-    for request in request_iter:
-      for parameter in request.response_parameters:
-        response = response_pb2.StreamingOutputCallResponse()
-        response.payload.payload_type = payload_pb2.COMPRESSABLE
-        response.payload.payload_compressable = 'a' * parameter.size
-        self._control()
-        yield response
+    def __init__(self):
+        self._condition = threading.Condition()
+        self._paused = False
+        self._fail = False
+
+    @contextlib.contextmanager
+    def pause(self):  # pylint: disable=invalid-name
+        with self._condition:
+            self._paused = True
+        yield
+        with self._condition:
+            self._paused = False
+            self._condition.notify_all()
 
-  def HalfDuplexCall(self, request_iter, unused_rpc_context):
-    responses = []
-    for request in request_iter:
-      for parameter in request.response_parameters:
-        response = response_pb2.StreamingOutputCallResponse()
+    @contextlib.contextmanager
+    def fail(self):  # pylint: disable=invalid-name
+        with self._condition:
+            self._fail = True
+        yield
+        with self._condition:
+            self._fail = False
+
+    def _control(self):  # pylint: disable=invalid-name
+        with self._condition:
+            if self._fail:
+                raise ValueError()
+            while self._paused:
+                self._condition.wait()
+
+    def UnaryCall(self, request, unused_rpc_context):
+        response = response_pb2.SimpleResponse()
         response.payload.payload_type = payload_pb2.COMPRESSABLE
-        response.payload.payload_compressable = 'a' * parameter.size
+        response.payload.payload_compressable = 'a' * request.response_size
+        self._control()
+        return response
+
+    def StreamingOutputCall(self, request, unused_rpc_context):
+        for parameter in request.response_parameters:
+            response = response_pb2.StreamingOutputCallResponse()
+            response.payload.payload_type = payload_pb2.COMPRESSABLE
+            response.payload.payload_compressable = 'a' * parameter.size
+            self._control()
+            yield response
+
+    def StreamingInputCall(self, request_iter, unused_rpc_context):
+        response = response_pb2.StreamingInputCallResponse()
+        aggregated_payload_size = 0
+        for request in request_iter:
+            aggregated_payload_size += len(request.payload.payload_compressable)
+        response.aggregated_payload_size = aggregated_payload_size
         self._control()
-        responses.append(response)
-    for response in responses:
-      yield response
+        return response
+
+    def FullDuplexCall(self, request_iter, unused_rpc_context):
+        for request in request_iter:
+            for parameter in request.response_parameters:
+                response = response_pb2.StreamingOutputCallResponse()
+                response.payload.payload_type = payload_pb2.COMPRESSABLE
+                response.payload.payload_compressable = 'a' * parameter.size
+                self._control()
+                yield response
+
+    def HalfDuplexCall(self, request_iter, unused_rpc_context):
+        responses = []
+        for request in request_iter:
+            for parameter in request.response_parameters:
+                response = response_pb2.StreamingOutputCallResponse()
+                response.payload.payload_type = payload_pb2.COMPRESSABLE
+                response.payload.payload_compressable = 'a' * parameter.size
+                self._control()
+                responses.append(response)
+        for response in responses:
+            yield response
 
 
 class _Service(
-    collections.namedtuple(
-      '_Service', ('servicer_methods', 'server', 'stub',))):
-  """A live and running service.
+        collections.namedtuple('_Service', (
+            'servicer_methods',
+            'server',
+            'stub',))):
+    """A live and running service.
 
   Attributes:
     servicer_methods: The _ServicerMethods servicing RPCs.
     server: The grpc.Server servicing RPCs.
     stub: A stub on which to invoke RPCs.
   """
-      
+
 
 def _CreateService():
-  """Provides a servicer backend and a stub.
+    """Provides a servicer backend and a stub.
 
   Returns:
     A _Service with which to test RPCs.
   """
-  servicer_methods = _ServicerMethods()
+    servicer_methods = _ServicerMethods()
 
-  class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
 
-    def UnaryCall(self, request, context):
-      return servicer_methods.UnaryCall(request, context)
+        def UnaryCall(self, request, context):
+            return servicer_methods.UnaryCall(request, context)
 
-    def StreamingOutputCall(self, request, context):
-      return servicer_methods.StreamingOutputCall(request, context)
+        def StreamingOutputCall(self, request, context):
+            return servicer_methods.StreamingOutputCall(request, context)
 
-    def StreamingInputCall(self, request_iter, context):
-      return servicer_methods.StreamingInputCall(request_iter, context)
+        def StreamingInputCall(self, request_iter, context):
+            return servicer_methods.StreamingInputCall(request_iter, context)
 
-    def FullDuplexCall(self, request_iter, context):
-      return servicer_methods.FullDuplexCall(request_iter, context)
+        def FullDuplexCall(self, request_iter, context):
+            return servicer_methods.FullDuplexCall(request_iter, context)
 
-    def HalfDuplexCall(self, request_iter, context):
-      return servicer_methods.HalfDuplexCall(request_iter, context)
+        def HalfDuplexCall(self, request_iter, context):
+            return servicer_methods.HalfDuplexCall(request_iter, context)
 
-  server = grpc.server(
-      futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-  getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
-  port = server.add_insecure_port('[::]:0')
-  server.start()
-  channel = grpc.insecure_channel('localhost:{}'.format(port))
-  stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
-  return _Service(servicer_methods, server, stub)
+    server = grpc.server(
+        futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
+    getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
+    port = server.add_insecure_port('[::]:0')
+    server.start()
+    channel = grpc.insecure_channel('localhost:{}'.format(port))
+    stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
+    return _Service(servicer_methods, server, stub)
 
 
 def _CreateIncompleteService():
-  """Provides a servicer backend that fails to implement methods and its stub.
+    """Provides a servicer backend that fails to implement methods and its stub.
 
   Returns:
     A _Service with which to test RPCs. The returned _Service's
       servicer_methods implements none of the methods required of it.
   """
 
-  class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
-    pass
+    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+        pass
 
-  server = grpc.server(
-      futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-  getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
-  port = server.add_insecure_port('[::]:0')
-  server.start()
-  channel = grpc.insecure_channel('localhost:{}'.format(port))
-  stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
-  return _Service(None, server, stub)
+    server = grpc.server(
+        futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
+    getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
+    port = server.add_insecure_port('[::]:0')
+    server.start()
+    channel = grpc.insecure_channel('localhost:{}'.format(port))
+    stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
+    return _Service(None, server, stub)
 
 
 def _streaming_input_request_iterator():
-  for _ in range(3):
-    request = request_pb2.StreamingInputCallRequest()
-    request.payload.payload_type = payload_pb2.COMPRESSABLE
-    request.payload.payload_compressable = 'a'
-    yield request
+    for _ in range(3):
+        request = request_pb2.StreamingInputCallRequest()
+        request.payload.payload_type = payload_pb2.COMPRESSABLE
+        request.payload.payload_compressable = 'a'
+        yield request
 
 
 def _streaming_output_request():
-  request = request_pb2.StreamingOutputCallRequest()
-  sizes = [1, 2, 3]
-  request.response_parameters.add(size=sizes[0], interval_us=0)
-  request.response_parameters.add(size=sizes[1], interval_us=0)
-  request.response_parameters.add(size=sizes[2], interval_us=0)
-  return request
+    request = request_pb2.StreamingOutputCallRequest()
+    sizes = [1, 2, 3]
+    request.response_parameters.add(size=sizes[0], interval_us=0)
+    request.response_parameters.add(size=sizes[1], interval_us=0)
+    request.response_parameters.add(size=sizes[2], interval_us=0)
+    return request
 
 
 def _full_duplex_request_iterator():
-  request = request_pb2.StreamingOutputCallRequest()
-  request.response_parameters.add(size=1, interval_us=0)
-  yield request
-  request = request_pb2.StreamingOutputCallRequest()
-  request.response_parameters.add(size=2, interval_us=0)
-  request.response_parameters.add(size=3, interval_us=0)
-  yield request
+    request = request_pb2.StreamingOutputCallRequest()
+    request.response_parameters.add(size=1, interval_us=0)
+    yield request
+    request = request_pb2.StreamingOutputCallRequest()
+    request.response_parameters.add(size=2, interval_us=0)
+    request.response_parameters.add(size=3, interval_us=0)
+    yield request
 
 
 class PythonPluginTest(unittest.TestCase):
-  """Test case for the gRPC Python protoc-plugin.
+    """Test case for the gRPC Python protoc-plugin.
 
   While reading these tests, remember that the futures API
   (`stub.method.future()`) only gives futures for the *response-unary*
   methods and does not exist for response-streaming methods.
   """
 
-  def testImportAttributes(self):
-    # check that we can access the generated module and its members.
-    self.assertIsNotNone(
-        getattr(service_pb2, STUB_IDENTIFIER, None))
-    self.assertIsNotNone(
-        getattr(service_pb2, SERVICER_IDENTIFIER, None))
-    self.assertIsNotNone(
-        getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER, None))
-
-  def testUpDown(self):
-    service = _CreateService()
-    self.assertIsNotNone(service.servicer_methods)
-    self.assertIsNotNone(service.server)
-    self.assertIsNotNone(service.stub)
-
-  def testIncompleteServicer(self):
-    service = _CreateIncompleteService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    with self.assertRaises(grpc.RpcError) as exception_context:
-      service.stub.UnaryCall(request)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.UNIMPLEMENTED)
-
-  def testUnaryCall(self):
-    service = _CreateService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    response = service.stub.UnaryCall(request)
-    expected_response = service.servicer_methods.UnaryCall(
-        request, 'not a real context!')
-    self.assertEqual(expected_response, response)
-
-  def testUnaryCallFuture(self):
-    service = _CreateService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    # Check that the call does not block waiting for the server to respond.
-    with service.servicer_methods.pause():
-      response_future = service.stub.UnaryCall.future(request)
-    response = response_future.result()
-    expected_response = service.servicer_methods.UnaryCall(
-        request, 'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testUnaryCallFutureExpired(self):
-    service = _CreateService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    with service.servicer_methods.pause():
-      response_future = service.stub.UnaryCall.future(
-          request, timeout=test_constants.SHORT_TIMEOUT)
-      with self.assertRaises(grpc.RpcError) as exception_context:
-        response_future.result()
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-    self.assertIs(response_future.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-
-  def testUnaryCallFutureCancelled(self):
-    service = _CreateService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    with service.servicer_methods.pause():
-      response_future = service.stub.UnaryCall.future(request)
-      response_future.cancel()
-    self.assertTrue(response_future.cancelled())
-    self.assertIs(response_future.code(), grpc.StatusCode.CANCELLED)
-
-  def testUnaryCallFutureFailed(self):
-    service = _CreateService()
-    request = request_pb2.SimpleRequest(response_size=13)
-    with service.servicer_methods.fail():
-      response_future = service.stub.UnaryCall.future(request)
-      self.assertIsNotNone(response_future.exception())
-    self.assertIs(response_future.code(), grpc.StatusCode.UNKNOWN)
-
-  def testStreamingOutputCall(self):
-    service = _CreateService()
-    request = _streaming_output_request()
-    responses = service.stub.StreamingOutputCall(request)
-    expected_responses = service.servicer_methods.StreamingOutputCall(
-        request, 'not a real RpcContext!')
-    for expected_response, response in moves.zip_longest(
-        expected_responses, responses):
-      self.assertEqual(expected_response, response)
-
-  def testStreamingOutputCallExpired(self):
-    service = _CreateService()
-    request = _streaming_output_request()
-    with service.servicer_methods.pause():
-      responses = service.stub.StreamingOutputCall(
-          request, timeout=test_constants.SHORT_TIMEOUT)
-      with self.assertRaises(grpc.RpcError) as exception_context:
-        list(responses)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-
-  def testStreamingOutputCallCancelled(self):
-    service = _CreateService()
-    request = _streaming_output_request()
-    responses = service.stub.StreamingOutputCall(request)
-    next(responses)
-    responses.cancel()
-    with self.assertRaises(grpc.RpcError) as exception_context:
-      next(responses)
-    self.assertIs(responses.code(), grpc.StatusCode.CANCELLED)
-
-  def testStreamingOutputCallFailed(self):
-    service = _CreateService()
-    request = _streaming_output_request()
-    with service.servicer_methods.fail():
-      responses = service.stub.StreamingOutputCall(request)
-      self.assertIsNotNone(responses)
-      with self.assertRaises(grpc.RpcError) as exception_context:
-        next(responses)
-    self.assertIs(exception_context.exception.code(), grpc.StatusCode.UNKNOWN)
-
-  def testStreamingInputCall(self):
-    service = _CreateService()
-    response = service.stub.StreamingInputCall(
-        _streaming_input_request_iterator())
-    expected_response = service.servicer_methods.StreamingInputCall(
-        _streaming_input_request_iterator(),
-        'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testStreamingInputCallFuture(self):
-    service = _CreateService()
-    with service.servicer_methods.pause():
-      response_future = service.stub.StreamingInputCall.future(
-          _streaming_input_request_iterator())
-    response = response_future.result()
-    expected_response = service.servicer_methods.StreamingInputCall(
-        _streaming_input_request_iterator(),
-        'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testStreamingInputCallFutureExpired(self):
-    service = _CreateService()
-    with service.servicer_methods.pause():
-      response_future = service.stub.StreamingInputCall.future(
-          _streaming_input_request_iterator(),
-          timeout=test_constants.SHORT_TIMEOUT)
-      with self.assertRaises(grpc.RpcError) as exception_context:
-        response_future.result()
-    self.assertIsInstance(response_future.exception(), grpc.RpcError)
-    self.assertIs(
-        response_future.exception().code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-
-  def testStreamingInputCallFutureCancelled(self):
-    service = _CreateService()
-    with service.servicer_methods.pause():
-      response_future = service.stub.StreamingInputCall.future(
-          _streaming_input_request_iterator())
-      response_future.cancel()
-    self.assertTrue(response_future.cancelled())
-    with self.assertRaises(grpc.FutureCancelledError):
-      response_future.result()
-
-  def testStreamingInputCallFutureFailed(self):
-    service = _CreateService()
-    with service.servicer_methods.fail():
-      response_future = service.stub.StreamingInputCall.future(
-          _streaming_input_request_iterator())
-      self.assertIsNotNone(response_future.exception())
-      self.assertIs(response_future.code(), grpc.StatusCode.UNKNOWN)
-
-  def testFullDuplexCall(self):
-    service = _CreateService()
-    responses = service.stub.FullDuplexCall(
-        _full_duplex_request_iterator())
-    expected_responses = service.servicer_methods.FullDuplexCall(
-        _full_duplex_request_iterator(),
-        'not a real RpcContext!')
-    for expected_response, response in moves.zip_longest(
-        expected_responses, responses):
-      self.assertEqual(expected_response, response)
-
-  def testFullDuplexCallExpired(self):
-    request_iterator = _full_duplex_request_iterator()
-    service = _CreateService()
-    with service.servicer_methods.pause():
-      responses = service.stub.FullDuplexCall(
-          request_iterator, timeout=test_constants.SHORT_TIMEOUT)
-      with self.assertRaises(grpc.RpcError) as exception_context:
-        list(responses)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
-
-  def testFullDuplexCallCancelled(self):
-    service = _CreateService()
-    request_iterator = _full_duplex_request_iterator()
-    responses = service.stub.FullDuplexCall(request_iterator)
-    next(responses)
-    responses.cancel()
-    with self.assertRaises(grpc.RpcError) as exception_context:
-      next(responses)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.CANCELLED)
-
-  def testFullDuplexCallFailed(self):
-    request_iterator = _full_duplex_request_iterator()
-    service = _CreateService()
-    with service.servicer_methods.fail():
-      responses = service.stub.FullDuplexCall(request_iterator)
-      with self.assertRaises(grpc.RpcError) as exception_context:
+    def testImportAttributes(self):
+        # check that we can access the generated module and its members.
+        self.assertIsNotNone(getattr(service_pb2, STUB_IDENTIFIER, None))
+        self.assertIsNotNone(getattr(service_pb2, SERVICER_IDENTIFIER, None))
+        self.assertIsNotNone(
+            getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER, None))
+
+    def testUpDown(self):
+        service = _CreateService()
+        self.assertIsNotNone(service.servicer_methods)
+        self.assertIsNotNone(service.server)
+        self.assertIsNotNone(service.stub)
+
+    def testIncompleteServicer(self):
+        service = _CreateIncompleteService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        with self.assertRaises(grpc.RpcError) as exception_context:
+            service.stub.UnaryCall(request)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.UNIMPLEMENTED)
+
+    def testUnaryCall(self):
+        service = _CreateService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        response = service.stub.UnaryCall(request)
+        expected_response = service.servicer_methods.UnaryCall(
+            request, 'not a real context!')
+        self.assertEqual(expected_response, response)
+
+    def testUnaryCallFuture(self):
+        service = _CreateService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        # Check that the call does not block waiting for the server to respond.
+        with service.servicer_methods.pause():
+            response_future = service.stub.UnaryCall.future(request)
+        response = response_future.result()
+        expected_response = service.servicer_methods.UnaryCall(
+            request, 'not a real RpcContext!')
+        self.assertEqual(expected_response, response)
+
+    def testUnaryCallFutureExpired(self):
+        service = _CreateService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        with service.servicer_methods.pause():
+            response_future = service.stub.UnaryCall.future(
+                request, timeout=test_constants.SHORT_TIMEOUT)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                response_future.result()
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
+        self.assertIs(response_future.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
+
+    def testUnaryCallFutureCancelled(self):
+        service = _CreateService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        with service.servicer_methods.pause():
+            response_future = service.stub.UnaryCall.future(request)
+            response_future.cancel()
+        self.assertTrue(response_future.cancelled())
+        self.assertIs(response_future.code(), grpc.StatusCode.CANCELLED)
+
+    def testUnaryCallFutureFailed(self):
+        service = _CreateService()
+        request = request_pb2.SimpleRequest(response_size=13)
+        with service.servicer_methods.fail():
+            response_future = service.stub.UnaryCall.future(request)
+            self.assertIsNotNone(response_future.exception())
+        self.assertIs(response_future.code(), grpc.StatusCode.UNKNOWN)
+
+    def testStreamingOutputCall(self):
+        service = _CreateService()
+        request = _streaming_output_request()
+        responses = service.stub.StreamingOutputCall(request)
+        expected_responses = service.servicer_methods.StreamingOutputCall(
+            request, 'not a real RpcContext!')
+        for expected_response, response in moves.zip_longest(expected_responses,
+                                                             responses):
+            self.assertEqual(expected_response, response)
+
+    def testStreamingOutputCallExpired(self):
+        service = _CreateService()
+        request = _streaming_output_request()
+        with service.servicer_methods.pause():
+            responses = service.stub.StreamingOutputCall(
+                request, timeout=test_constants.SHORT_TIMEOUT)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                list(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
+
+    def testStreamingOutputCallCancelled(self):
+        service = _CreateService()
+        request = _streaming_output_request()
+        responses = service.stub.StreamingOutputCall(request)
         next(responses)
-    self.assertIs(exception_context.exception.code(), grpc.StatusCode.UNKNOWN)
-
-  def testHalfDuplexCall(self):
-    service = _CreateService()
-    def half_duplex_request_iterator():
-      request = request_pb2.StreamingOutputCallRequest()
-      request.response_parameters.add(size=1, interval_us=0)
-      yield request
-      request = request_pb2.StreamingOutputCallRequest()
-      request.response_parameters.add(size=2, interval_us=0)
-      request.response_parameters.add(size=3, interval_us=0)
-      yield request
-    responses = service.stub.HalfDuplexCall(half_duplex_request_iterator())
-    expected_responses = service.servicer_methods.HalfDuplexCall(
-        half_duplex_request_iterator(), 'not a real RpcContext!')
-    for expected_response, response in moves.zip_longest(
-        expected_responses, responses):
-      self.assertEqual(expected_response, response)
-
-  def testHalfDuplexCallWedged(self):
-    condition = threading.Condition()
-    wait_cell = [False]
-    @contextlib.contextmanager
-    def wait():  # pylint: disable=invalid-name
-      # Where's Python 3's 'nonlocal' statement when you need it?
-      with condition:
-        wait_cell[0] = True
-      yield
-      with condition:
-        wait_cell[0] = False
-        condition.notify_all()
-    def half_duplex_request_iterator():
-      request = request_pb2.StreamingOutputCallRequest()
-      request.response_parameters.add(size=1, interval_us=0)
-      yield request
-      with condition:
-        while wait_cell[0]:
-          condition.wait()
-    service = _CreateService()
-    with wait():
-      responses = service.stub.HalfDuplexCall(
-          half_duplex_request_iterator(), timeout=test_constants.SHORT_TIMEOUT)
-      # half-duplex waits for the client to send all info
-      with self.assertRaises(grpc.RpcError) as exception_context:
+        responses.cancel()
+        with self.assertRaises(grpc.RpcError) as exception_context:
+            next(responses)
+        self.assertIs(responses.code(), grpc.StatusCode.CANCELLED)
+
+    def testStreamingOutputCallFailed(self):
+        service = _CreateService()
+        request = _streaming_output_request()
+        with service.servicer_methods.fail():
+            responses = service.stub.StreamingOutputCall(request)
+            self.assertIsNotNone(responses)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                next(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.UNKNOWN)
+
+    def testStreamingInputCall(self):
+        service = _CreateService()
+        response = service.stub.StreamingInputCall(
+            _streaming_input_request_iterator())
+        expected_response = service.servicer_methods.StreamingInputCall(
+            _streaming_input_request_iterator(), 'not a real RpcContext!')
+        self.assertEqual(expected_response, response)
+
+    def testStreamingInputCallFuture(self):
+        service = _CreateService()
+        with service.servicer_methods.pause():
+            response_future = service.stub.StreamingInputCall.future(
+                _streaming_input_request_iterator())
+        response = response_future.result()
+        expected_response = service.servicer_methods.StreamingInputCall(
+            _streaming_input_request_iterator(), 'not a real RpcContext!')
+        self.assertEqual(expected_response, response)
+
+    def testStreamingInputCallFutureExpired(self):
+        service = _CreateService()
+        with service.servicer_methods.pause():
+            response_future = service.stub.StreamingInputCall.future(
+                _streaming_input_request_iterator(),
+                timeout=test_constants.SHORT_TIMEOUT)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                response_future.result()
+        self.assertIsInstance(response_future.exception(), grpc.RpcError)
+        self.assertIs(response_future.exception().code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
+
+    def testStreamingInputCallFutureCancelled(self):
+        service = _CreateService()
+        with service.servicer_methods.pause():
+            response_future = service.stub.StreamingInputCall.future(
+                _streaming_input_request_iterator())
+            response_future.cancel()
+        self.assertTrue(response_future.cancelled())
+        with self.assertRaises(grpc.FutureCancelledError):
+            response_future.result()
+
+    def testStreamingInputCallFutureFailed(self):
+        service = _CreateService()
+        with service.servicer_methods.fail():
+            response_future = service.stub.StreamingInputCall.future(
+                _streaming_input_request_iterator())
+            self.assertIsNotNone(response_future.exception())
+            self.assertIs(response_future.code(), grpc.StatusCode.UNKNOWN)
+
+    def testFullDuplexCall(self):
+        service = _CreateService()
+        responses = service.stub.FullDuplexCall(_full_duplex_request_iterator())
+        expected_responses = service.servicer_methods.FullDuplexCall(
+            _full_duplex_request_iterator(), 'not a real RpcContext!')
+        for expected_response, response in moves.zip_longest(expected_responses,
+                                                             responses):
+            self.assertEqual(expected_response, response)
+
+    def testFullDuplexCallExpired(self):
+        request_iterator = _full_duplex_request_iterator()
+        service = _CreateService()
+        with service.servicer_methods.pause():
+            responses = service.stub.FullDuplexCall(
+                request_iterator, timeout=test_constants.SHORT_TIMEOUT)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                list(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
+
+    def testFullDuplexCallCancelled(self):
+        service = _CreateService()
+        request_iterator = _full_duplex_request_iterator()
+        responses = service.stub.FullDuplexCall(request_iterator)
         next(responses)
-    self.assertIs(
-        exception_context.exception.code(), grpc.StatusCode.DEADLINE_EXCEEDED)
+        responses.cancel()
+        with self.assertRaises(grpc.RpcError) as exception_context:
+            next(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.CANCELLED)
+
+    def testFullDuplexCallFailed(self):
+        request_iterator = _full_duplex_request_iterator()
+        service = _CreateService()
+        with service.servicer_methods.fail():
+            responses = service.stub.FullDuplexCall(request_iterator)
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                next(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.UNKNOWN)
+
+    def testHalfDuplexCall(self):
+        service = _CreateService()
+
+        def half_duplex_request_iterator():
+            request = request_pb2.StreamingOutputCallRequest()
+            request.response_parameters.add(size=1, interval_us=0)
+            yield request
+            request = request_pb2.StreamingOutputCallRequest()
+            request.response_parameters.add(size=2, interval_us=0)
+            request.response_parameters.add(size=3, interval_us=0)
+            yield request
+
+        responses = service.stub.HalfDuplexCall(half_duplex_request_iterator())
+        expected_responses = service.servicer_methods.HalfDuplexCall(
+            half_duplex_request_iterator(), 'not a real RpcContext!')
+        for expected_response, response in moves.zip_longest(expected_responses,
+                                                             responses):
+            self.assertEqual(expected_response, response)
+
+    def testHalfDuplexCallWedged(self):
+        condition = threading.Condition()
+        wait_cell = [False]
+
+        @contextlib.contextmanager
+        def wait():  # pylint: disable=invalid-name
+            # Where's Python 3's 'nonlocal' statement when you need it?
+            with condition:
+                wait_cell[0] = True
+            yield
+            with condition:
+                wait_cell[0] = False
+                condition.notify_all()
+
+        def half_duplex_request_iterator():
+            request = request_pb2.StreamingOutputCallRequest()
+            request.response_parameters.add(size=1, interval_us=0)
+            yield request
+            with condition:
+                while wait_cell[0]:
+                    condition.wait()
+
+        service = _CreateService()
+        with wait():
+            responses = service.stub.HalfDuplexCall(
+                half_duplex_request_iterator(),
+                timeout=test_constants.SHORT_TIMEOUT)
+            # half-duplex waits for the client to send all info
+            with self.assertRaises(grpc.RpcError) as exception_context:
+                next(responses)
+        self.assertIs(exception_context.exception.code(),
+                      grpc.StatusCode.DEADLINE_EXCEEDED)
 
 
 if __name__ == '__main__':
-  unittest.main(verbosity=2)
+    unittest.main(verbosity=2)
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py b/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py
index f8ae05bb7a..bcc01f3978 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py
@@ -49,256 +49,264 @@ from tests.unit.framework.common import test_constants
 
 _MESSAGES_IMPORT = b'import "messages.proto";'
 
+
 @contextlib.contextmanager
 def _system_path(path):
-  old_system_path = sys.path[:]
-  sys.path = sys.path[0:1] + path + sys.path[1:]
-  yield
-  sys.path = old_system_path
+    old_system_path = sys.path[:]
+    sys.path = sys.path[0:1] + path + sys.path[1:]
+    yield
+    sys.path = old_system_path
 
 
 class DummySplitServicer(object):
 
-  def __init__(self, request_class, response_class):
-    self.request_class = request_class
-    self.response_class = response_class
+    def __init__(self, request_class, response_class):
+        self.request_class = request_class
+        self.response_class = response_class
 
-  def Call(self, request, context):
-    return self.response_class()
+    def Call(self, request, context):
+        return self.response_class()
 
 
 class SeparateTestMixin(object):
 
-  def testImportAttributes(self):
-    with _system_path([self.python_out_directory]):
-      pb2 = importlib.import_module(self.pb2_import)
-    pb2.Request
-    pb2.Response
-    if self.should_find_services_in_pb2:
-      pb2.TestServiceServicer
-    else:
-      with self.assertRaises(AttributeError):
-        pb2.TestServiceServicer
-
-    with _system_path([self.grpc_python_out_directory]):
-      pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-    pb2_grpc.TestServiceServicer
-    with self.assertRaises(AttributeError):
-      pb2_grpc.Request
-    with self.assertRaises(AttributeError):
-      pb2_grpc.Response
-
-  def testCall(self):
-    with _system_path([self.python_out_directory]):
-      pb2 = importlib.import_module(self.pb2_import)
-    with _system_path([self.grpc_python_out_directory]):
-      pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-    server = grpc.server(
-        futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-    pb2_grpc.add_TestServiceServicer_to_server(
-        DummySplitServicer(
-            pb2.Request, pb2.Response), server)
-    port = server.add_insecure_port('[::]:0')
-    server.start()
-    channel = grpc.insecure_channel('localhost:{}'.format(port))
-    stub = pb2_grpc.TestServiceStub(channel)
-    request = pb2.Request()
-    expected_response = pb2.Response()
-    response = stub.Call(request)
-    self.assertEqual(expected_response, response)
+    def testImportAttributes(self):
+        with _system_path([self.python_out_directory]):
+            pb2 = importlib.import_module(self.pb2_import)
+        pb2.Request
+        pb2.Response
+        if self.should_find_services_in_pb2:
+            pb2.TestServiceServicer
+        else:
+            with self.assertRaises(AttributeError):
+                pb2.TestServiceServicer
+
+        with _system_path([self.grpc_python_out_directory]):
+            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
+        pb2_grpc.TestServiceServicer
+        with self.assertRaises(AttributeError):
+            pb2_grpc.Request
+        with self.assertRaises(AttributeError):
+            pb2_grpc.Response
+
+    def testCall(self):
+        with _system_path([self.python_out_directory]):
+            pb2 = importlib.import_module(self.pb2_import)
+        with _system_path([self.grpc_python_out_directory]):
+            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
+        server = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
+        pb2_grpc.add_TestServiceServicer_to_server(
+            DummySplitServicer(pb2.Request, pb2.Response), server)
+        port = server.add_insecure_port('[::]:0')
+        server.start()
+        channel = grpc.insecure_channel('localhost:{}'.format(port))
+        stub = pb2_grpc.TestServiceStub(channel)
+        request = pb2.Request()
+        expected_response = pb2.Response()
+        response = stub.Call(request)
+        self.assertEqual(expected_response, response)
 
 
 class CommonTestMixin(object):
 
-  def testImportAttributes(self):
-    with _system_path([self.python_out_directory]):
-      pb2 = importlib.import_module(self.pb2_import)
-    pb2.Request
-    pb2.Response
-    if self.should_find_services_in_pb2:
-      pb2.TestServiceServicer
-    else:
-      with self.assertRaises(AttributeError):
-        pb2.TestServiceServicer
-
-    with _system_path([self.grpc_python_out_directory]):
-      pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-    pb2_grpc.TestServiceServicer
-    with self.assertRaises(AttributeError):
-      pb2_grpc.Request
-    with self.assertRaises(AttributeError):
-      pb2_grpc.Response
-
-  def testCall(self):
-    with _system_path([self.python_out_directory]):
-      pb2 = importlib.import_module(self.pb2_import)
-    with _system_path([self.grpc_python_out_directory]):
-      pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-    server = grpc.server(
-        futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-    pb2_grpc.add_TestServiceServicer_to_server(
-        DummySplitServicer(
-            pb2.Request, pb2.Response), server)
-    port = server.add_insecure_port('[::]:0')
-    server.start()
-    channel = grpc.insecure_channel('localhost:{}'.format(port))
-    stub = pb2_grpc.TestServiceStub(channel)
-    request = pb2.Request()
-    expected_response = pb2.Response()
-    response = stub.Call(request)
-    self.assertEqual(expected_response, response)
+    def testImportAttributes(self):
+        with _system_path([self.python_out_directory]):
+            pb2 = importlib.import_module(self.pb2_import)
+        pb2.Request
+        pb2.Response
+        if self.should_find_services_in_pb2:
+            pb2.TestServiceServicer
+        else:
+            with self.assertRaises(AttributeError):
+                pb2.TestServiceServicer
+
+        with _system_path([self.grpc_python_out_directory]):
+            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
+        pb2_grpc.TestServiceServicer
+        with self.assertRaises(AttributeError):
+            pb2_grpc.Request
+        with self.assertRaises(AttributeError):
+            pb2_grpc.Response
+
+    def testCall(self):
+        with _system_path([self.python_out_directory]):
+            pb2 = importlib.import_module(self.pb2_import)
+        with _system_path([self.grpc_python_out_directory]):
+            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
+        server = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
+        pb2_grpc.add_TestServiceServicer_to_server(
+            DummySplitServicer(pb2.Request, pb2.Response), server)
+        port = server.add_insecure_port('[::]:0')
+        server.start()
+        channel = grpc.insecure_channel('localhost:{}'.format(port))
+        stub = pb2_grpc.TestServiceStub(channel)
+        request = pb2.Request()
+        expected_response = pb2.Response()
+        response = stub.Call(request)
+        self.assertEqual(expected_response, response)
 
 
 class SameSeparateTest(unittest.TestCase, SeparateTestMixin):
 
-  def setUp(self):
-    same_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
-    self.directory = tempfile.mkdtemp(suffix='same_separate', dir='.')
-    self.proto_directory = os.path.join(self.directory, 'proto_path')
-    self.python_out_directory = os.path.join(self.directory, 'python_out')
-    self.grpc_python_out_directory = os.path.join(self.directory, 'grpc_python_out')
-    os.makedirs(self.proto_directory)
-    os.makedirs(self.python_out_directory)
-    os.makedirs(self.grpc_python_out_directory)
-    same_proto_file = os.path.join(self.proto_directory, 'same_separate.proto')
-    open(same_proto_file, 'wb').write(same_proto_contents)
-    protoc_result = protoc.main([
-        '',
-        '--proto_path={}'.format(self.proto_directory),
-        '--python_out={}'.format(self.python_out_directory),
-        '--grpc_python_out=grpc_2_0:{}'.format(self.grpc_python_out_directory),
-        same_proto_file,
-    ])
-    if protoc_result != 0:
-      raise Exception("unexpected protoc error")
-    open(os.path.join(self.grpc_python_out_directory, '__init__.py'), 'w').write('')
-    open(os.path.join(self.python_out_directory, '__init__.py'), 'w').write('')
-    self.pb2_import = 'same_separate_pb2'
-    self.pb2_grpc_import = 'same_separate_pb2_grpc'
-    self.should_find_services_in_pb2 = False
-
-  def tearDown(self):
-    shutil.rmtree(self.directory)
+    def setUp(self):
+        same_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
+        self.directory = tempfile.mkdtemp(suffix='same_separate', dir='.')
+        self.proto_directory = os.path.join(self.directory, 'proto_path')
+        self.python_out_directory = os.path.join(self.directory, 'python_out')
+        self.grpc_python_out_directory = os.path.join(self.directory,
+                                                      'grpc_python_out')
+        os.makedirs(self.proto_directory)
+        os.makedirs(self.python_out_directory)
+        os.makedirs(self.grpc_python_out_directory)
+        same_proto_file = os.path.join(self.proto_directory,
+                                       'same_separate.proto')
+        open(same_proto_file, 'wb').write(same_proto_contents)
+        protoc_result = protoc.main([
+            '',
+            '--proto_path={}'.format(self.proto_directory),
+            '--python_out={}'.format(self.python_out_directory),
+            '--grpc_python_out=grpc_2_0:{}'.format(
+                self.grpc_python_out_directory),
+            same_proto_file,
+        ])
+        if protoc_result != 0:
+            raise Exception("unexpected protoc error")
+        open(os.path.join(self.grpc_python_out_directory, '__init__.py'),
+             'w').write('')
+        open(os.path.join(self.python_out_directory, '__init__.py'),
+             'w').write('')
+        self.pb2_import = 'same_separate_pb2'
+        self.pb2_grpc_import = 'same_separate_pb2_grpc'
+        self.should_find_services_in_pb2 = False
+
+    def tearDown(self):
+        shutil.rmtree(self.directory)
 
 
 class SameCommonTest(unittest.TestCase, CommonTestMixin):
 
-  def setUp(self):
-    same_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
-    self.directory = tempfile.mkdtemp(suffix='same_common', dir='.')
-    self.proto_directory = os.path.join(self.directory, 'proto_path')
-    self.python_out_directory = os.path.join(self.directory, 'python_out')
-    self.grpc_python_out_directory = self.python_out_directory
-    os.makedirs(self.proto_directory)
-    os.makedirs(self.python_out_directory)
-    same_proto_file = os.path.join(self.proto_directory, 'same_common.proto')
-    open(same_proto_file, 'wb').write(same_proto_contents)
-    protoc_result = protoc.main([
-        '',
-        '--proto_path={}'.format(self.proto_directory),
-        '--python_out={}'.format(self.python_out_directory),
-        '--grpc_python_out={}'.format(self.grpc_python_out_directory),
-        same_proto_file,
-    ])
-    if protoc_result != 0:
-      raise Exception("unexpected protoc error")
-    open(os.path.join(self.python_out_directory, '__init__.py'), 'w').write('')
-    self.pb2_import = 'same_common_pb2'
-    self.pb2_grpc_import = 'same_common_pb2_grpc'
-    self.should_find_services_in_pb2 = True
-
-  def tearDown(self):
-    shutil.rmtree(self.directory)
+    def setUp(self):
+        same_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
+        self.directory = tempfile.mkdtemp(suffix='same_common', dir='.')
+        self.proto_directory = os.path.join(self.directory, 'proto_path')
+        self.python_out_directory = os.path.join(self.directory, 'python_out')
+        self.grpc_python_out_directory = self.python_out_directory
+        os.makedirs(self.proto_directory)
+        os.makedirs(self.python_out_directory)
+        same_proto_file = os.path.join(self.proto_directory,
+                                       'same_common.proto')
+        open(same_proto_file, 'wb').write(same_proto_contents)
+        protoc_result = protoc.main([
+            '',
+            '--proto_path={}'.format(self.proto_directory),
+            '--python_out={}'.format(self.python_out_directory),
+            '--grpc_python_out={}'.format(self.grpc_python_out_directory),
+            same_proto_file,
+        ])
+        if protoc_result != 0:
+            raise Exception("unexpected protoc error")
+        open(os.path.join(self.python_out_directory, '__init__.py'),
+             'w').write('')
+        self.pb2_import = 'same_common_pb2'
+        self.pb2_grpc_import = 'same_common_pb2_grpc'
+        self.should_find_services_in_pb2 = True
+
+    def tearDown(self):
+        shutil.rmtree(self.directory)
 
 
 class SplitCommonTest(unittest.TestCase, CommonTestMixin):
 
-  def setUp(self):
-    services_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing.split_services',
-        'services.proto')
-    messages_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing.split_messages',
-        'messages.proto')
-    self.directory = tempfile.mkdtemp(suffix='split_common', dir='.')
-    self.proto_directory = os.path.join(self.directory, 'proto_path')
-    self.python_out_directory = os.path.join(self.directory, 'python_out')
-    self.grpc_python_out_directory = self.python_out_directory
-    os.makedirs(self.proto_directory)
-    os.makedirs(self.python_out_directory)
-    services_proto_file = os.path.join(self.proto_directory,
-                                       'split_common_services.proto')
-    messages_proto_file = os.path.join(self.proto_directory,
-                                       'split_common_messages.proto')
-    open(services_proto_file, 'wb').write(services_proto_contents.replace(
-        _MESSAGES_IMPORT,
-        b'import "split_common_messages.proto";'
-    ))
-    open(messages_proto_file, 'wb').write(messages_proto_contents)
-    protoc_result = protoc.main([
-        '',
-        '--proto_path={}'.format(self.proto_directory),
-        '--python_out={}'.format(self.python_out_directory),
-        '--grpc_python_out={}'.format(self.grpc_python_out_directory),
-        services_proto_file,
-        messages_proto_file,
-    ])
-    if protoc_result != 0:
-      raise Exception("unexpected protoc error")
-    open(os.path.join(self.python_out_directory, '__init__.py'), 'w').write('')
-    self.pb2_import = 'split_common_messages_pb2'
-    self.pb2_grpc_import = 'split_common_services_pb2_grpc'
-    self.should_find_services_in_pb2 = False
-
-  def tearDown(self):
-    shutil.rmtree(self.directory)
+    def setUp(self):
+        services_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing.split_services',
+            'services.proto')
+        messages_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing.split_messages',
+            'messages.proto')
+        self.directory = tempfile.mkdtemp(suffix='split_common', dir='.')
+        self.proto_directory = os.path.join(self.directory, 'proto_path')
+        self.python_out_directory = os.path.join(self.directory, 'python_out')
+        self.grpc_python_out_directory = self.python_out_directory
+        os.makedirs(self.proto_directory)
+        os.makedirs(self.python_out_directory)
+        services_proto_file = os.path.join(self.proto_directory,
+                                           'split_common_services.proto')
+        messages_proto_file = os.path.join(self.proto_directory,
+                                           'split_common_messages.proto')
+        open(services_proto_file, 'wb').write(
+            services_proto_contents.replace(
+                _MESSAGES_IMPORT, b'import "split_common_messages.proto";'))
+        open(messages_proto_file, 'wb').write(messages_proto_contents)
+        protoc_result = protoc.main([
+            '',
+            '--proto_path={}'.format(self.proto_directory),
+            '--python_out={}'.format(self.python_out_directory),
+            '--grpc_python_out={}'.format(self.grpc_python_out_directory),
+            services_proto_file,
+            messages_proto_file,
+        ])
+        if protoc_result != 0:
+            raise Exception("unexpected protoc error")
+        open(os.path.join(self.python_out_directory, '__init__.py'),
+             'w').write('')
+        self.pb2_import = 'split_common_messages_pb2'
+        self.pb2_grpc_import = 'split_common_services_pb2_grpc'
+        self.should_find_services_in_pb2 = False
+
+    def tearDown(self):
+        shutil.rmtree(self.directory)
 
 
 class SplitSeparateTest(unittest.TestCase, SeparateTestMixin):
 
-  def setUp(self):
-    services_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing.split_services',
-        'services.proto')
-    messages_proto_contents = pkgutil.get_data(
-        'tests.protoc_plugin.protos.invocation_testing.split_messages',
-        'messages.proto')
-    self.directory = tempfile.mkdtemp(suffix='split_separate', dir='.')
-    self.proto_directory = os.path.join(self.directory, 'proto_path')
-    self.python_out_directory = os.path.join(self.directory, 'python_out')
-    self.grpc_python_out_directory = os.path.join(self.directory, 'grpc_python_out')
-    os.makedirs(self.proto_directory)
-    os.makedirs(self.python_out_directory)
-    os.makedirs(self.grpc_python_out_directory)
-    services_proto_file = os.path.join(self.proto_directory,
-                                       'split_separate_services.proto')
-    messages_proto_file = os.path.join(self.proto_directory,
-                                       'split_separate_messages.proto')
-    open(services_proto_file, 'wb').write(services_proto_contents.replace(
-        _MESSAGES_IMPORT,
-        b'import "split_separate_messages.proto";'
-    ))
-    open(messages_proto_file, 'wb').write(messages_proto_contents)
-    protoc_result = protoc.main([
-        '',
-        '--proto_path={}'.format(self.proto_directory),
-        '--python_out={}'.format(self.python_out_directory),
-        '--grpc_python_out=grpc_2_0:{}'.format(self.grpc_python_out_directory),
-        services_proto_file,
-        messages_proto_file,
-    ])
-    if protoc_result != 0:
-      raise Exception("unexpected protoc error")
-    open(os.path.join(self.python_out_directory, '__init__.py'), 'w').write('')
-    self.pb2_import = 'split_separate_messages_pb2'
-    self.pb2_grpc_import = 'split_separate_services_pb2_grpc'
-    self.should_find_services_in_pb2 = False
-
-  def tearDown(self):
-    shutil.rmtree(self.directory)
+    def setUp(self):
+        services_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing.split_services',
+            'services.proto')
+        messages_proto_contents = pkgutil.get_data(
+            'tests.protoc_plugin.protos.invocation_testing.split_messages',
+            'messages.proto')
+        self.directory = tempfile.mkdtemp(suffix='split_separate', dir='.')
+        self.proto_directory = os.path.join(self.directory, 'proto_path')
+        self.python_out_directory = os.path.join(self.directory, 'python_out')
+        self.grpc_python_out_directory = os.path.join(self.directory,
+                                                      'grpc_python_out')
+        os.makedirs(self.proto_directory)
+        os.makedirs(self.python_out_directory)
+        os.makedirs(self.grpc_python_out_directory)
+        services_proto_file = os.path.join(self.proto_directory,
+                                           'split_separate_services.proto')
+        messages_proto_file = os.path.join(self.proto_directory,
+                                           'split_separate_messages.proto')
+        open(services_proto_file, 'wb').write(
+            services_proto_contents.replace(
+                _MESSAGES_IMPORT, b'import "split_separate_messages.proto";'))
+        open(messages_proto_file, 'wb').write(messages_proto_contents)
+        protoc_result = protoc.main([
+            '',
+            '--proto_path={}'.format(self.proto_directory),
+            '--python_out={}'.format(self.python_out_directory),
+            '--grpc_python_out=grpc_2_0:{}'.format(
+                self.grpc_python_out_directory),
+            services_proto_file,
+            messages_proto_file,
+        ])
+        if protoc_result != 0:
+            raise Exception("unexpected protoc error")
+        open(os.path.join(self.python_out_directory, '__init__.py'),
+             'w').write('')
+        self.pb2_import = 'split_separate_messages_pb2'
+        self.pb2_grpc_import = 'split_separate_services_pb2_grpc'
+        self.should_find_services_in_pb2 = False
+
+    def tearDown(self):
+        shutil.rmtree(self.directory)
 
 
 if __name__ == '__main__':
-  unittest.main(verbosity=2)
+    unittest.main(verbosity=2)
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py b/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
index 1eba9c9354..f64f4e962b 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/beta_python_plugin_test.py
@@ -64,84 +64,84 @@ STUB_FACTORY_IDENTIFIER = 'beta_create_TestService_stub'
 
 class _ServicerMethods(object):
 
-  def __init__(self):
-    self._condition = threading.Condition()
-    self._paused = False
-    self._fail = False
-
-  @contextlib.contextmanager
-  def pause(self):  # pylint: disable=invalid-name
-    with self._condition:
-      self._paused = True
-    yield
-    with self._condition:
-      self._paused = False
-      self._condition.notify_all()
-
-  @contextlib.contextmanager
-  def fail(self):  # pylint: disable=invalid-name
-    with self._condition:
-      self._fail = True
-    yield
-    with self._condition:
-      self._fail = False
-
-  def _control(self):  # pylint: disable=invalid-name
-    with self._condition:
-      if self._fail:
-        raise ValueError()
-      while self._paused:
-        self._condition.wait()
-
-  def UnaryCall(self, request, unused_rpc_context):
-    response = response_pb2.SimpleResponse()
-    response.payload.payload_type = payload_pb2.COMPRESSABLE
-    response.payload.payload_compressable = 'a' * request.response_size
-    self._control()
-    return response
-
-  def StreamingOutputCall(self, request, unused_rpc_context):
-    for parameter in request.response_parameters:
-      response = response_pb2.StreamingOutputCallResponse()
-      response.payload.payload_type = payload_pb2.COMPRESSABLE
-      response.payload.payload_compressable = 'a' * parameter.size
-      self._control()
-      yield response
-
-  def StreamingInputCall(self, request_iter, unused_rpc_context):
-    response = response_pb2.StreamingInputCallResponse()
-    aggregated_payload_size = 0
-    for request in request_iter:
-      aggregated_payload_size += len(request.payload.payload_compressable)
-    response.aggregated_payload_size = aggregated_payload_size
-    self._control()
-    return response
-
-  def FullDuplexCall(self, request_iter, unused_rpc_context):
-    for request in request_iter:
-      for parameter in request.response_parameters:
-        response = response_pb2.StreamingOutputCallResponse()
-        response.payload.payload_type = payload_pb2.COMPRESSABLE
-        response.payload.payload_compressable = 'a' * parameter.size
-        self._control()
-        yield response
+    def __init__(self):
+        self._condition = threading.Condition()
+        self._paused = False
+        self._fail = False
+
+    @contextlib.contextmanager
+    def pause(self):  # pylint: disable=invalid-name
+        with self._condition:
+            self._paused = True
+        yield
+        with self._condition:
+            self._paused = False
+            self._condition.notify_all()
 
-  def HalfDuplexCall(self, request_iter, unused_rpc_context):
-    responses = []
-    for request in request_iter:
-      for parameter in request.response_parameters:
-        response = response_pb2.StreamingOutputCallResponse()
+    @contextlib.contextmanager
+    def fail(self):  # pylint: disable=invalid-name
+        with self._condition:
+            self._fail = True
+        yield
+        with self._condition:
+            self._fail = False
+
+    def _control(self):  # pylint: disable=invalid-name
+        with self._condition:
+            if self._fail:
+                raise ValueError()
+            while self._paused:
+                self._condition.wait()
+
+    def UnaryCall(self, request, unused_rpc_context):
+        response = response_pb2.SimpleResponse()
         response.payload.payload_type = payload_pb2.COMPRESSABLE
-        response.payload.payload_compressable = 'a' * parameter.size
+        response.payload.payload_compressable = 'a' * request.response_size
+        self._control()
+        return response
+
+    def StreamingOutputCall(self, request, unused_rpc_context):
+        for parameter in request.response_parameters:
+            response = response_pb2.StreamingOutputCallResponse()
+            response.payload.payload_type = payload_pb2.COMPRESSABLE
+            response.payload.payload_compressable = 'a' * parameter.size
+            self._control()
+            yield response
+
+    def StreamingInputCall(self, request_iter, unused_rpc_context):
+        response = response_pb2.StreamingInputCallResponse()
+        aggregated_payload_size = 0
+        for request in request_iter:
+            aggregated_payload_size += len(request.payload.payload_compressable)
+        response.aggregated_payload_size = aggregated_payload_size
         self._control()
-        responses.append(response)
-    for response in responses:
-      yield response
+        return response
+
+    def FullDuplexCall(self, request_iter, unused_rpc_context):
+        for request in request_iter:
+            for parameter in request.response_parameters:
+                response = response_pb2.StreamingOutputCallResponse()
+                response.payload.payload_type = payload_pb2.COMPRESSABLE
+                response.payload.payload_compressable = 'a' * parameter.size
+                self._control()
+                yield response
+
+    def HalfDuplexCall(self, request_iter, unused_rpc_context):
+        responses = []
+        for request in request_iter:
+            for parameter in request.response_parameters:
+                response = response_pb2.StreamingOutputCallResponse()
+                response.payload.payload_type = payload_pb2.COMPRESSABLE
+                response.payload.payload_compressable = 'a' * parameter.size
+                self._control()
+                responses.append(response)
+        for response in responses:
+            yield response
 
 
 @contextlib.contextmanager
 def _CreateService():
-  """Provides a servicer backend and a stub.
+    """Provides a servicer backend and a 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.
@@ -151,38 +151,38 @@ def _CreateService():
       the service bound to the stub and and stub is the stub on which to invoke
       RPCs.
   """
-  servicer_methods = _ServicerMethods()
+    servicer_methods = _ServicerMethods()
 
-  class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
 
-    def UnaryCall(self, request, context):
-      return servicer_methods.UnaryCall(request, context)
+        def UnaryCall(self, request, context):
+            return servicer_methods.UnaryCall(request, context)
 
-    def StreamingOutputCall(self, request, context):
-      return servicer_methods.StreamingOutputCall(request, context)
+        def StreamingOutputCall(self, request, context):
+            return servicer_methods.StreamingOutputCall(request, context)
 
-    def StreamingInputCall(self, request_iter, context):
-      return servicer_methods.StreamingInputCall(request_iter, context)
+        def StreamingInputCall(self, request_iter, context):
+            return servicer_methods.StreamingInputCall(request_iter, context)
 
-    def FullDuplexCall(self, request_iter, context):
-      return servicer_methods.FullDuplexCall(request_iter, context)
+        def FullDuplexCall(self, request_iter, context):
+            return servicer_methods.FullDuplexCall(request_iter, context)
 
-    def HalfDuplexCall(self, request_iter, context):
-      return servicer_methods.HalfDuplexCall(request_iter, context)
+        def HalfDuplexCall(self, request_iter, context):
+            return servicer_methods.HalfDuplexCall(request_iter, context)
 
-  servicer = Servicer()
-  server = getattr(service_pb2, SERVER_FACTORY_IDENTIFIER)(servicer)
-  port = server.add_insecure_port('[::]:0')
-  server.start()
-  channel = implementations.insecure_channel('localhost', port)
-  stub = getattr(service_pb2, STUB_FACTORY_IDENTIFIER)(channel)
-  yield (servicer_methods, stub)
-  server.stop(0)
+    servicer = Servicer()
+    server = getattr(service_pb2, SERVER_FACTORY_IDENTIFIER)(servicer)
+    port = server.add_insecure_port('[::]:0')
+    server.start()
+    channel = implementations.insecure_channel('localhost', port)
+    stub = getattr(service_pb2, STUB_FACTORY_IDENTIFIER)(channel)
+    yield (servicer_methods, stub)
+    server.stop(0)
 
 
 @contextlib.contextmanager
 def _CreateIncompleteService():
-  """Provides a servicer backend that fails to implement methods and its stub.
+    """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.
@@ -194,297 +194,297 @@ def _CreateIncompleteService():
       RPCs.
   """
 
-  class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
-    pass
+    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+        pass
 
-  servicer = Servicer()
-  server = getattr(service_pb2, SERVER_FACTORY_IDENTIFIER)(servicer)
-  port = server.add_insecure_port('[::]:0')
-  server.start()
-  channel = implementations.insecure_channel('localhost', port)
-  stub = getattr(service_pb2, STUB_FACTORY_IDENTIFIER)(channel)
-  yield None, stub
-  server.stop(0)
+    servicer = Servicer()
+    server = getattr(service_pb2, SERVER_FACTORY_IDENTIFIER)(servicer)
+    port = server.add_insecure_port('[::]:0')
+    server.start()
+    channel = implementations.insecure_channel('localhost', port)
+    stub = getattr(service_pb2, STUB_FACTORY_IDENTIFIER)(channel)
+    yield None, stub
+    server.stop(0)
 
 
 def _streaming_input_request_iterator():
-  for _ in range(3):
-    request = request_pb2.StreamingInputCallRequest()
-    request.payload.payload_type = payload_pb2.COMPRESSABLE
-    request.payload.payload_compressable = 'a'
-    yield request
+    for _ in range(3):
+        request = request_pb2.StreamingInputCallRequest()
+        request.payload.payload_type = payload_pb2.COMPRESSABLE
+        request.payload.payload_compressable = 'a'
+        yield request
 
 
 def _streaming_output_request():
-  request = request_pb2.StreamingOutputCallRequest()
-  sizes = [1, 2, 3]
-  request.response_parameters.add(size=sizes[0], interval_us=0)
-  request.response_parameters.add(size=sizes[1], interval_us=0)
-  request.response_parameters.add(size=sizes[2], interval_us=0)
-  return request
+    request = request_pb2.StreamingOutputCallRequest()
+    sizes = [1, 2, 3]
+    request.response_parameters.add(size=sizes[0], interval_us=0)
+    request.response_parameters.add(size=sizes[1], interval_us=0)
+    request.response_parameters.add(size=sizes[2], interval_us=0)
+    return request
 
 
 def _full_duplex_request_iterator():
-  request = request_pb2.StreamingOutputCallRequest()
-  request.response_parameters.add(size=1, interval_us=0)
-  yield request
-  request = request_pb2.StreamingOutputCallRequest()
-  request.response_parameters.add(size=2, interval_us=0)
-  request.response_parameters.add(size=3, interval_us=0)
-  yield request
+    request = request_pb2.StreamingOutputCallRequest()
+    request.response_parameters.add(size=1, interval_us=0)
+    yield request
+    request = request_pb2.StreamingOutputCallRequest()
+    request.response_parameters.add(size=2, interval_us=0)
+    request.response_parameters.add(size=3, interval_us=0)
+    yield request
 
 
 class PythonPluginTest(unittest.TestCase):
-  """Test case for the gRPC Python protoc-plugin.
+    """Test case for the gRPC Python protoc-plugin.
 
   While reading these tests, remember that the futures API
   (`stub.method.future()`) only gives futures for the *response-unary*
   methods and does not exist for response-streaming methods.
   """
 
-  def testImportAttributes(self):
-    # check that we can access the generated module and its members.
-    self.assertIsNotNone(
-        getattr(service_pb2, SERVICER_IDENTIFIER, None))
-    self.assertIsNotNone(
-        getattr(service_pb2, STUB_IDENTIFIER, None))
-    self.assertIsNotNone(
-        getattr(service_pb2, SERVER_FACTORY_IDENTIFIER, None))
-    self.assertIsNotNone(
-        getattr(service_pb2, STUB_FACTORY_IDENTIFIER, None))
-
-  def testUpDown(self):
-    with _CreateService():
-      request_pb2.SimpleRequest(response_size=13)
-
-  def testIncompleteServicer(self):
-    with _CreateIncompleteService() as (_, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      try:
-        stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
-      except face.AbortionError as error:
-        self.assertEqual(interfaces.StatusCode.UNIMPLEMENTED, error.code)
-
-  def testUnaryCall(self):
-    with _CreateService() as (methods, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      response = stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
-    expected_response = methods.UnaryCall(request, 'not a real context!')
-    self.assertEqual(expected_response, response)
-
-  def testUnaryCallFuture(self):
-    with _CreateService() as (methods, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      # Check that the call does not block waiting for the server to respond.
-      with methods.pause():
-        response_future = stub.UnaryCall.future(
-            request, test_constants.LONG_TIMEOUT)
-      response = response_future.result()
-    expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testUnaryCallFutureExpired(self):
-    with _CreateService() as (methods, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      with methods.pause():
-        response_future = stub.UnaryCall.future(
-            request, test_constants.SHORT_TIMEOUT)
-        with self.assertRaises(face.ExpirationError):
-          response_future.result()
-
-  def testUnaryCallFutureCancelled(self):
-    with _CreateService() as (methods, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      with methods.pause():
-        response_future = stub.UnaryCall.future(request, 1)
-        response_future.cancel()
-        self.assertTrue(response_future.cancelled())
-
-  def testUnaryCallFutureFailed(self):
-    with _CreateService() as (methods, stub):
-      request = request_pb2.SimpleRequest(response_size=13)
-      with methods.fail():
-        response_future = stub.UnaryCall.future(
-            request, test_constants.LONG_TIMEOUT)
-        self.assertIsNotNone(response_future.exception())
-
-  def testStreamingOutputCall(self):
-    with _CreateService() as (methods, stub):
-      request = _streaming_output_request()
-      responses = stub.StreamingOutputCall(
-          request, test_constants.LONG_TIMEOUT)
-      expected_responses = methods.StreamingOutputCall(
-          request, 'not a real RpcContext!')
-      for expected_response, response in moves.zip_longest(
-          expected_responses, responses):
+    def testImportAttributes(self):
+        # check that we can access the generated module and its members.
+        self.assertIsNotNone(getattr(service_pb2, SERVICER_IDENTIFIER, None))
+        self.assertIsNotNone(getattr(service_pb2, STUB_IDENTIFIER, None))
+        self.assertIsNotNone(
+            getattr(service_pb2, SERVER_FACTORY_IDENTIFIER, None))
+        self.assertIsNotNone(
+            getattr(service_pb2, STUB_FACTORY_IDENTIFIER, None))
+
+    def testUpDown(self):
+        with _CreateService():
+            request_pb2.SimpleRequest(response_size=13)
+
+    def testIncompleteServicer(self):
+        with _CreateIncompleteService() as (_, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            try:
+                stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
+            except face.AbortionError as error:
+                self.assertEqual(interfaces.StatusCode.UNIMPLEMENTED,
+                                 error.code)
+
+    def testUnaryCall(self):
+        with _CreateService() as (methods, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            response = stub.UnaryCall(request, test_constants.LONG_TIMEOUT)
+        expected_response = methods.UnaryCall(request, 'not a real context!')
         self.assertEqual(expected_response, response)
 
-  def testStreamingOutputCallExpired(self):
-    with _CreateService() as (methods, stub):
-      request = _streaming_output_request()
-      with methods.pause():
-        responses = stub.StreamingOutputCall(
-            request, test_constants.SHORT_TIMEOUT)
-        with self.assertRaises(face.ExpirationError):
-          list(responses)
-
-  def testStreamingOutputCallCancelled(self):
-    with _CreateService() as (methods, stub):
-      request = _streaming_output_request()
-      responses = stub.StreamingOutputCall(
-          request, test_constants.LONG_TIMEOUT)
-      next(responses)
-      responses.cancel()
-      with self.assertRaises(face.CancellationError):
-        next(responses)
-
-  def testStreamingOutputCallFailed(self):
-    with _CreateService() as (methods, stub):
-      request = _streaming_output_request()
-      with methods.fail():
-        responses = stub.StreamingOutputCall(request, 1)
-        self.assertIsNotNone(responses)
-        with self.assertRaises(face.RemoteError):
-          next(responses)
-
-  def testStreamingInputCall(self):
-    with _CreateService() as (methods, stub):
-      response = stub.StreamingInputCall(
-          _streaming_input_request_iterator(),
-          test_constants.LONG_TIMEOUT)
-    expected_response = methods.StreamingInputCall(
-        _streaming_input_request_iterator(),
-        'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testStreamingInputCallFuture(self):
-    with _CreateService() as (methods, stub):
-      with methods.pause():
-        response_future = stub.StreamingInputCall.future(
-            _streaming_input_request_iterator(),
-            test_constants.LONG_TIMEOUT)
-      response = response_future.result()
-    expected_response = methods.StreamingInputCall(
-        _streaming_input_request_iterator(),
-        'not a real RpcContext!')
-    self.assertEqual(expected_response, response)
-
-  def testStreamingInputCallFutureExpired(self):
-    with _CreateService() as (methods, stub):
-      with methods.pause():
-        response_future = stub.StreamingInputCall.future(
-            _streaming_input_request_iterator(),
-            test_constants.SHORT_TIMEOUT)
-        with self.assertRaises(face.ExpirationError):
-          response_future.result()
-        self.assertIsInstance(
-            response_future.exception(), face.ExpirationError)
-
-  def testStreamingInputCallFutureCancelled(self):
-    with _CreateService() as (methods, stub):
-      with methods.pause():
-        response_future = stub.StreamingInputCall.future(
-            _streaming_input_request_iterator(),
-            test_constants.LONG_TIMEOUT)
-        response_future.cancel()
-        self.assertTrue(response_future.cancelled())
-      with self.assertRaises(future.CancelledError):
-        response_future.result()
-
-  def testStreamingInputCallFutureFailed(self):
-    with _CreateService() as (methods, stub):
-      with methods.fail():
-        response_future = stub.StreamingInputCall.future(
-            _streaming_input_request_iterator(),
-            test_constants.LONG_TIMEOUT)
-        self.assertIsNotNone(response_future.exception())
-
-  def testFullDuplexCall(self):
-    with _CreateService() as (methods, stub):
-      responses = stub.FullDuplexCall(
-          _full_duplex_request_iterator(),
-          test_constants.LONG_TIMEOUT)
-      expected_responses = methods.FullDuplexCall(
-          _full_duplex_request_iterator(),
-          'not a real RpcContext!')
-      for expected_response, response in moves.zip_longest(
-          expected_responses, responses):
+    def testUnaryCallFuture(self):
+        with _CreateService() as (methods, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            # Check that the call does not block waiting for the server to respond.
+            with methods.pause():
+                response_future = stub.UnaryCall.future(
+                    request, test_constants.LONG_TIMEOUT)
+            response = response_future.result()
+        expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
         self.assertEqual(expected_response, response)
 
-  def testFullDuplexCallExpired(self):
-    request_iterator = _full_duplex_request_iterator()
-    with _CreateService() as (methods, stub):
-      with methods.pause():
-        responses = stub.FullDuplexCall(
-            request_iterator, test_constants.SHORT_TIMEOUT)
-        with self.assertRaises(face.ExpirationError):
-          list(responses)
-
-  def testFullDuplexCallCancelled(self):
-    with _CreateService() as (methods, stub):
-      request_iterator = _full_duplex_request_iterator()
-      responses = stub.FullDuplexCall(
-          request_iterator, test_constants.LONG_TIMEOUT)
-      next(responses)
-      responses.cancel()
-      with self.assertRaises(face.CancellationError):
-        next(responses)
-
-  def testFullDuplexCallFailed(self):
-    request_iterator = _full_duplex_request_iterator()
-    with _CreateService() as (methods, stub):
-      with methods.fail():
-        responses = stub.FullDuplexCall(
-            request_iterator, test_constants.LONG_TIMEOUT)
-        self.assertIsNotNone(responses)
-        with self.assertRaises(face.RemoteError):
-          next(responses)
-
-  def testHalfDuplexCall(self):
-    with _CreateService() as (methods, stub):
-      def half_duplex_request_iterator():
-        request = request_pb2.StreamingOutputCallRequest()
-        request.response_parameters.add(size=1, interval_us=0)
-        yield request
-        request = request_pb2.StreamingOutputCallRequest()
-        request.response_parameters.add(size=2, interval_us=0)
-        request.response_parameters.add(size=3, interval_us=0)
-        yield request
-      responses = stub.HalfDuplexCall(
-          half_duplex_request_iterator(), test_constants.LONG_TIMEOUT)
-      expected_responses = methods.HalfDuplexCall(
-          half_duplex_request_iterator(), 'not a real RpcContext!')
-      for check in moves.zip_longest(expected_responses, responses):
-        expected_response, response = check
+    def testUnaryCallFutureExpired(self):
+        with _CreateService() as (methods, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            with methods.pause():
+                response_future = stub.UnaryCall.future(
+                    request, test_constants.SHORT_TIMEOUT)
+                with self.assertRaises(face.ExpirationError):
+                    response_future.result()
+
+    def testUnaryCallFutureCancelled(self):
+        with _CreateService() as (methods, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            with methods.pause():
+                response_future = stub.UnaryCall.future(request, 1)
+                response_future.cancel()
+                self.assertTrue(response_future.cancelled())
+
+    def testUnaryCallFutureFailed(self):
+        with _CreateService() as (methods, stub):
+            request = request_pb2.SimpleRequest(response_size=13)
+            with methods.fail():
+                response_future = stub.UnaryCall.future(
+                    request, test_constants.LONG_TIMEOUT)
+                self.assertIsNotNone(response_future.exception())
+
+    def testStreamingOutputCall(self):
+        with _CreateService() as (methods, stub):
+            request = _streaming_output_request()
+            responses = stub.StreamingOutputCall(request,
+                                                 test_constants.LONG_TIMEOUT)
+            expected_responses = methods.StreamingOutputCall(
+                request, 'not a real RpcContext!')
+            for expected_response, response in moves.zip_longest(
+                    expected_responses, responses):
+                self.assertEqual(expected_response, response)
+
+    def testStreamingOutputCallExpired(self):
+        with _CreateService() as (methods, stub):
+            request = _streaming_output_request()
+            with methods.pause():
+                responses = stub.StreamingOutputCall(
+                    request, test_constants.SHORT_TIMEOUT)
+                with self.assertRaises(face.ExpirationError):
+                    list(responses)
+
+    def testStreamingOutputCallCancelled(self):
+        with _CreateService() as (methods, stub):
+            request = _streaming_output_request()
+            responses = stub.StreamingOutputCall(request,
+                                                 test_constants.LONG_TIMEOUT)
+            next(responses)
+            responses.cancel()
+            with self.assertRaises(face.CancellationError):
+                next(responses)
+
+    def testStreamingOutputCallFailed(self):
+        with _CreateService() as (methods, stub):
+            request = _streaming_output_request()
+            with methods.fail():
+                responses = stub.StreamingOutputCall(request, 1)
+                self.assertIsNotNone(responses)
+                with self.assertRaises(face.RemoteError):
+                    next(responses)
+
+    def testStreamingInputCall(self):
+        with _CreateService() as (methods, stub):
+            response = stub.StreamingInputCall(
+                _streaming_input_request_iterator(),
+                test_constants.LONG_TIMEOUT)
+        expected_response = methods.StreamingInputCall(
+            _streaming_input_request_iterator(), 'not a real RpcContext!')
         self.assertEqual(expected_response, response)
 
-  def testHalfDuplexCallWedged(self):
-    condition = threading.Condition()
-    wait_cell = [False]
-    @contextlib.contextmanager
-    def wait():  # pylint: disable=invalid-name
-      # Where's Python 3's 'nonlocal' statement when you need it?
-      with condition:
-        wait_cell[0] = True
-      yield
-      with condition:
-        wait_cell[0] = False
-        condition.notify_all()
-    def half_duplex_request_iterator():
-      request = request_pb2.StreamingOutputCallRequest()
-      request.response_parameters.add(size=1, interval_us=0)
-      yield request
-      with condition:
-        while wait_cell[0]:
-          condition.wait()
-    with _CreateService() as (methods, stub):
-      with wait():
-        responses = stub.HalfDuplexCall(
-            half_duplex_request_iterator(), test_constants.SHORT_TIMEOUT)
-        # half-duplex waits for the client to send all info
-        with self.assertRaises(face.ExpirationError):
-          next(responses)
+    def testStreamingInputCallFuture(self):
+        with _CreateService() as (methods, stub):
+            with methods.pause():
+                response_future = stub.StreamingInputCall.future(
+                    _streaming_input_request_iterator(),
+                    test_constants.LONG_TIMEOUT)
+            response = response_future.result()
+        expected_response = methods.StreamingInputCall(
+            _streaming_input_request_iterator(), 'not a real RpcContext!')
+        self.assertEqual(expected_response, response)
+
+    def testStreamingInputCallFutureExpired(self):
+        with _CreateService() as (methods, stub):
+            with methods.pause():
+                response_future = stub.StreamingInputCall.future(
+                    _streaming_input_request_iterator(),
+                    test_constants.SHORT_TIMEOUT)
+                with self.assertRaises(face.ExpirationError):
+                    response_future.result()
+                self.assertIsInstance(response_future.exception(),
+                                      face.ExpirationError)
+
+    def testStreamingInputCallFutureCancelled(self):
+        with _CreateService() as (methods, stub):
+            with methods.pause():
+                response_future = stub.StreamingInputCall.future(
+                    _streaming_input_request_iterator(),
+                    test_constants.LONG_TIMEOUT)
+                response_future.cancel()
+                self.assertTrue(response_future.cancelled())
+            with self.assertRaises(future.CancelledError):
+                response_future.result()
+
+    def testStreamingInputCallFutureFailed(self):
+        with _CreateService() as (methods, stub):
+            with methods.fail():
+                response_future = stub.StreamingInputCall.future(
+                    _streaming_input_request_iterator(),
+                    test_constants.LONG_TIMEOUT)
+                self.assertIsNotNone(response_future.exception())
+
+    def testFullDuplexCall(self):
+        with _CreateService() as (methods, stub):
+            responses = stub.FullDuplexCall(_full_duplex_request_iterator(),
+                                            test_constants.LONG_TIMEOUT)
+            expected_responses = methods.FullDuplexCall(
+                _full_duplex_request_iterator(), 'not a real RpcContext!')
+            for expected_response, response in moves.zip_longest(
+                    expected_responses, responses):
+                self.assertEqual(expected_response, response)
+
+    def testFullDuplexCallExpired(self):
+        request_iterator = _full_duplex_request_iterator()
+        with _CreateService() as (methods, stub):
+            with methods.pause():
+                responses = stub.FullDuplexCall(request_iterator,
+                                                test_constants.SHORT_TIMEOUT)
+                with self.assertRaises(face.ExpirationError):
+                    list(responses)
+
+    def testFullDuplexCallCancelled(self):
+        with _CreateService() as (methods, stub):
+            request_iterator = _full_duplex_request_iterator()
+            responses = stub.FullDuplexCall(request_iterator,
+                                            test_constants.LONG_TIMEOUT)
+            next(responses)
+            responses.cancel()
+            with self.assertRaises(face.CancellationError):
+                next(responses)
+
+    def testFullDuplexCallFailed(self):
+        request_iterator = _full_duplex_request_iterator()
+        with _CreateService() as (methods, stub):
+            with methods.fail():
+                responses = stub.FullDuplexCall(request_iterator,
+                                                test_constants.LONG_TIMEOUT)
+                self.assertIsNotNone(responses)
+                with self.assertRaises(face.RemoteError):
+                    next(responses)
+
+    def testHalfDuplexCall(self):
+        with _CreateService() as (methods, stub):
+
+            def half_duplex_request_iterator():
+                request = request_pb2.StreamingOutputCallRequest()
+                request.response_parameters.add(size=1, interval_us=0)
+                yield request
+                request = request_pb2.StreamingOutputCallRequest()
+                request.response_parameters.add(size=2, interval_us=0)
+                request.response_parameters.add(size=3, interval_us=0)
+                yield request
+
+            responses = stub.HalfDuplexCall(half_duplex_request_iterator(),
+                                            test_constants.LONG_TIMEOUT)
+            expected_responses = methods.HalfDuplexCall(
+                half_duplex_request_iterator(), 'not a real RpcContext!')
+            for check in moves.zip_longest(expected_responses, responses):
+                expected_response, response = check
+                self.assertEqual(expected_response, response)
+
+    def testHalfDuplexCallWedged(self):
+        condition = threading.Condition()
+        wait_cell = [False]
+
+        @contextlib.contextmanager
+        def wait():  # pylint: disable=invalid-name
+            # Where's Python 3's 'nonlocal' statement when you need it?
+            with condition:
+                wait_cell[0] = True
+            yield
+            with condition:
+                wait_cell[0] = False
+                condition.notify_all()
+
+        def half_duplex_request_iterator():
+            request = request_pb2.StreamingOutputCallRequest()
+            request.response_parameters.add(size=1, interval_us=0)
+            yield request
+            with condition:
+                while wait_cell[0]:
+                    condition.wait()
+
+        with _CreateService() as (methods, stub):
+            with wait():
+                responses = stub.HalfDuplexCall(half_duplex_request_iterator(),
+                                                test_constants.SHORT_TIMEOUT)
+                # half-duplex waits for the client to send all info
+                with self.assertRaises(face.ExpirationError):
+                    next(responses)
 
 
 if __name__ == '__main__':
-  unittest.main(verbosity=2)
+    unittest.main(verbosity=2)
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/__init__.py
index 2f88fa0412..100a624dc9 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_messages/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_messages/__init__.py
index 2f88fa0412..100a624dc9 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_messages/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_messages/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_services/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_services/__init__.py
index 2f88fa0412..100a624dc9 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_services/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/invocation_testing/split_services/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/payload/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/payload/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/payload/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/payload/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/r/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/r/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/r/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/requests/r/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/responses/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/responses/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/responses/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/responses/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-
diff --git a/src/python/grpcio_tests/tests/protoc_plugin/protos/service/__init__.py b/src/python/grpcio_tests/tests/protoc_plugin/protos/service/__init__.py
index 7086519106..b89398809f 100644
--- a/src/python/grpcio_tests/tests/protoc_plugin/protos/service/__init__.py
+++ b/src/python/grpcio_tests/tests/protoc_plugin/protos/service/__init__.py
@@ -26,5 +26,3 @@
 # 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.
-
-