| # Copyright 2015, Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # * Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # * Redistributions in binary form must reproduce the above |
| # copyright notice, this list of conditions and the following disclaimer |
| # in the documentation and/or other materials provided with the |
| # distribution. |
| # * Neither the name of Google Inc. nor the names of its |
| # contributors may be used to endorse or promote products derived from |
| # this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| import argparse |
| import contextlib |
| import distutils.spawn |
| import errno |
| import itertools |
| import os |
| import pkg_resources |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| import threading |
| import time |
| import unittest |
| |
| from grpc.framework.alpha import exceptions |
| from grpc.framework.foundation import future |
| |
| # Identifiers of entities we expect to find in the generated module. |
| SERVICER_IDENTIFIER = 'EarlyAdopterTestServiceServicer' |
| SERVER_IDENTIFIER = 'EarlyAdopterTestServiceServer' |
| STUB_IDENTIFIER = 'EarlyAdopterTestServiceStub' |
| SERVER_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_server' |
| STUB_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_stub' |
| |
| # The timeout used in tests of RPCs that are supposed to expire. |
| SHORT_TIMEOUT = 2 |
| # The timeout used in tests of RPCs that are not supposed to expire. The |
| # absurdly large value doesn't matter since no passing execution of this test |
| # module will ever wait the duration. |
| LONG_TIMEOUT = 600 |
| NO_DELAY = 0 |
| |
| |
| class _ServicerMethods(object): |
| |
| def __init__(self, test_pb2, delay): |
| self._condition = threading.Condition() |
| self._delay = delay |
| self._paused = False |
| self._fail = False |
| self._test_pb2 = test_pb2 |
| |
| @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() |
| time.sleep(self._delay) |
| |
| def UnaryCall(self, request, unused_rpc_context): |
| response = self._test_pb2.SimpleResponse() |
| response.payload.payload_type = self._test_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 = self._test_pb2.StreamingOutputCallResponse() |
| response.payload.payload_type = self._test_pb2.COMPRESSABLE |
| response.payload.payload_compressable = 'a' * parameter.size |
| self._control() |
| yield response |
| |
| def StreamingInputCall(self, request_iter, unused_rpc_context): |
| response = self._test_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 = self._test_pb2.StreamingOutputCallResponse() |
| response.payload.payload_type = self._test_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 = self._test_pb2.StreamingOutputCallResponse() |
| response.payload.payload_type = self._test_pb2.COMPRESSABLE |
| response.payload.payload_compressable = 'a' * parameter.size |
| self._control() |
| responses.append(response) |
| for response in responses: |
| yield response |
| |
| |
| @contextlib.contextmanager |
| def _CreateService(test_pb2, delay): |
| """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. |
| |
| Non-zero delay puts a delay on each call to the servicer, representative of |
| communication latency. Timeout is the default timeout for the stub while |
| waiting for the service. |
| |
| Args: |
| test_pb2: The test_pb2 module generated by this test. |
| delay: Delay in seconds per response from the servicer. |
| |
| Yields: |
| A (servicer_methods, servicer, stub) three-tuple where servicer_methods is |
| the back-end of the service bound to the stub and the server and stub |
| are both activated and ready for use. |
| """ |
| servicer_methods = _ServicerMethods(test_pb2, delay) |
| |
| class Servicer(getattr(test_pb2, SERVICER_IDENTIFIER)): |
| |
| def UnaryCall(self, request, context): |
| return servicer_methods.UnaryCall(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 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) |
| |
| servicer = Servicer() |
| server = getattr( |
| test_pb2, SERVER_FACTORY_IDENTIFIER)(servicer, 0) |
| with server: |
| port = server.port() |
| stub = getattr(test_pb2, STUB_FACTORY_IDENTIFIER)('localhost', port) |
| with stub: |
| yield servicer_methods, stub, server |
| |
| |
| def _streaming_input_request_iterator(test_pb2): |
| for _ in range(3): |
| request = test_pb2.StreamingInputCallRequest() |
| request.payload.payload_type = test_pb2.COMPRESSABLE |
| request.payload.payload_compressable = 'a' |
| yield request |
| |
| |
| def _streaming_output_request(test_pb2): |
| request = test_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(test_pb2): |
| request = test_pb2.StreamingOutputCallRequest() |
| request.response_parameters.add(size=1, interval_us=0) |
| yield request |
| request = test_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. |
| |
| While reading these tests, remember that the futures API |
| (`stub.method.async()`) only gives futures for the *non-streaming* responses, |
| else it behaves like its blocking cousin. |
| """ |
| |
| def setUp(self): |
| # Assume that the appropriate protoc and grpc_python_plugins are on the |
| # path. |
| protoc_command = 'protoc' |
| protoc_plugin_filename = distutils.spawn.find_executable( |
| 'grpc_python_plugin') |
| test_proto_filename = pkg_resources.resource_filename( |
| 'grpc_protoc_plugin', 'test.proto') |
| if not os.path.isfile(protoc_command): |
| # Assume that if we haven't built protoc that it's on the system. |
| protoc_command = 'protoc' |
| |
| # Ensure that the output directory exists. |
| self.outdir = tempfile.mkdtemp() |
| |
| # Invoke protoc with the plugin. |
| cmd = [ |
| protoc_command, |
| '--plugin=protoc-gen-python-grpc=%s' % protoc_plugin_filename, |
| '-I .', |
| '--python_out=%s' % self.outdir, |
| '--python-grpc_out=%s' % self.outdir, |
| os.path.basename(test_proto_filename), |
| ] |
| subprocess.check_call(' '.join(cmd), shell=True, env=os.environ, |
| cwd=os.path.dirname(test_proto_filename)) |
| sys.path.append(self.outdir) |
| |
| def tearDown(self): |
| try: |
| shutil.rmtree(self.outdir) |
| except OSError as exc: |
| if exc.errno != errno.ENOENT: |
| raise |
| |
| # TODO(atash): Figure out which of these tests is hanging flakily with small |
| # probability. |
| |
| def testImportAttributes(self): |
| # check that we can access the generated module and its members. |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| self.assertIsNotNone(getattr(test_pb2, SERVICER_IDENTIFIER, None)) |
| self.assertIsNotNone(getattr(test_pb2, SERVER_IDENTIFIER, None)) |
| self.assertIsNotNone(getattr(test_pb2, STUB_IDENTIFIER, None)) |
| self.assertIsNotNone(getattr(test_pb2, SERVER_FACTORY_IDENTIFIER, None)) |
| self.assertIsNotNone(getattr(test_pb2, STUB_FACTORY_IDENTIFIER, None)) |
| |
| def testUpDown(self): |
| import test_pb2 |
| with _CreateService( |
| test_pb2, NO_DELAY) as (servicer, stub, unused_server): |
| request = test_pb2.SimpleRequest(response_size=13) |
| |
| def testUnaryCall(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| timeout = 6 # TODO(issue 2039): LONG_TIMEOUT like the other methods. |
| request = test_pb2.SimpleRequest(response_size=13) |
| response = stub.UnaryCall(request, timeout) |
| expected_response = methods.UnaryCall(request, 'not a real RpcContext!') |
| self.assertEqual(expected_response, response) |
| |
| def testUnaryCallAsync(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = test_pb2.SimpleRequest(response_size=13) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| # Check that the call does not block waiting for the server to respond. |
| with methods.pause(): |
| response_future = stub.UnaryCall.async(request, LONG_TIMEOUT) |
| response = response_future.result() |
| expected_response = methods.UnaryCall(request, 'not a real RpcContext!') |
| self.assertEqual(expected_response, response) |
| |
| def testUnaryCallAsyncExpired(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| request = test_pb2.SimpleRequest(response_size=13) |
| with methods.pause(): |
| response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT) |
| with self.assertRaises(exceptions.ExpirationError): |
| response_future.result() |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testUnaryCallAsyncCancelled(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = test_pb2.SimpleRequest(response_size=13) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| response_future = stub.UnaryCall.async(request, 1) |
| response_future.cancel() |
| self.assertTrue(response_future.cancelled()) |
| |
| def testUnaryCallAsyncFailed(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = test_pb2.SimpleRequest(response_size=13) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.fail(): |
| response_future = stub.UnaryCall.async(request, LONG_TIMEOUT) |
| self.assertIsNotNone(response_future.exception()) |
| |
| def testStreamingOutputCall(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = _streaming_output_request(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| responses = stub.StreamingOutputCall(request, LONG_TIMEOUT) |
| expected_responses = methods.StreamingOutputCall( |
| request, 'not a real RpcContext!') |
| for expected_response, response in itertools.izip_longest( |
| expected_responses, responses): |
| self.assertEqual(expected_response, response) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testStreamingOutputCallExpired(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = _streaming_output_request(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT) |
| with self.assertRaises(exceptions.ExpirationError): |
| list(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testStreamingOutputCallCancelled(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = _streaming_output_request(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| unused_methods, stub, unused_server): |
| responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT) |
| next(responses) |
| responses.cancel() |
| with self.assertRaises(future.CancelledError): |
| next(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this times out ' |
| 'instead of raising the proper error.') |
| def testStreamingOutputCallFailed(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request = _streaming_output_request(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.fail(): |
| responses = stub.StreamingOutputCall(request, 1) |
| self.assertIsNotNone(responses) |
| with self.assertRaises(exceptions.ServicerError): |
| next(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testStreamingInputCall(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| response = stub.StreamingInputCall( |
| _streaming_input_request_iterator(test_pb2), LONG_TIMEOUT) |
| expected_response = methods.StreamingInputCall( |
| _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!') |
| self.assertEqual(expected_response, response) |
| |
| def testStreamingInputCallAsync(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| response_future = stub.StreamingInputCall.async( |
| _streaming_input_request_iterator(test_pb2), LONG_TIMEOUT) |
| response = response_future.result() |
| expected_response = methods.StreamingInputCall( |
| _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!') |
| self.assertEqual(expected_response, response) |
| |
| def testStreamingInputCallAsyncExpired(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| response_future = stub.StreamingInputCall.async( |
| _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT) |
| with self.assertRaises(exceptions.ExpirationError): |
| response_future.result() |
| self.assertIsInstance( |
| response_future.exception(), exceptions.ExpirationError) |
| |
| def testStreamingInputCallAsyncCancelled(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| timeout = 6 # TODO(issue 2039): LONG_TIMEOUT like the other methods. |
| response_future = stub.StreamingInputCall.async( |
| _streaming_input_request_iterator(test_pb2), timeout) |
| response_future.cancel() |
| self.assertTrue(response_future.cancelled()) |
| with self.assertRaises(future.CancelledError): |
| response_future.result() |
| |
| def testStreamingInputCallAsyncFailed(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.fail(): |
| response_future = stub.StreamingInputCall.async( |
| _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT) |
| self.assertIsNotNone(response_future.exception()) |
| |
| def testFullDuplexCall(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| responses = stub.FullDuplexCall( |
| _full_duplex_request_iterator(test_pb2), LONG_TIMEOUT) |
| expected_responses = methods.FullDuplexCall( |
| _full_duplex_request_iterator(test_pb2), 'not a real RpcContext!') |
| for expected_response, response in itertools.izip_longest( |
| expected_responses, responses): |
| self.assertEqual(expected_response, response) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testFullDuplexCallExpired(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request_iterator = _full_duplex_request_iterator(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.pause(): |
| responses = stub.FullDuplexCall(request_iterator, SHORT_TIMEOUT) |
| with self.assertRaises(exceptions.ExpirationError): |
| list(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testFullDuplexCallCancelled(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| request_iterator = _full_duplex_request_iterator(test_pb2) |
| responses = stub.FullDuplexCall(request_iterator, LONG_TIMEOUT) |
| next(responses) |
| responses.cancel() |
| with self.assertRaises(future.CancelledError): |
| next(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this hangs forever ' |
| 'and fix.') |
| def testFullDuplexCallFailed(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| request_iterator = _full_duplex_request_iterator(test_pb2) |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| with methods.fail(): |
| responses = stub.FullDuplexCall(request_iterator, LONG_TIMEOUT) |
| self.assertIsNotNone(responses) |
| with self.assertRaises(exceptions.ServicerError): |
| next(responses) |
| |
| @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' |
| 'forever and fix.') |
| def testHalfDuplexCall(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| with _CreateService(test_pb2, NO_DELAY) as ( |
| methods, stub, unused_server): |
| def half_duplex_request_iterator(): |
| request = test_pb2.StreamingOutputCallRequest() |
| request.response_parameters.add(size=1, interval_us=0) |
| yield request |
| request = test_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(), LONG_TIMEOUT) |
| expected_responses = methods.HalfDuplexCall( |
| half_duplex_request_iterator(), 'not a real RpcContext!') |
| for check in itertools.izip_longest(expected_responses, responses): |
| expected_response, response = check |
| self.assertEqual(expected_response, response) |
| |
| def testHalfDuplexCallWedged(self): |
| import test_pb2 # pylint: disable=g-import-not-at-top |
| 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 = test_pb2.StreamingOutputCallRequest() |
| request.response_parameters.add(size=1, interval_us=0) |
| yield request |
| with condition: |
| while wait_cell[0]: |
| condition.wait() |
| with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): |
| with wait(): |
| responses = stub.HalfDuplexCall( |
| half_duplex_request_iterator(), SHORT_TIMEOUT) |
| # half-duplex waits for the client to send all info |
| with self.assertRaises(exceptions.ExpirationError): |
| next(responses) |
| |
| |
| if __name__ == '__main__': |
| os.chdir(os.path.dirname(sys.argv[0])) |
| unittest.main(verbosity=2) |