diff options
Diffstat (limited to 'test/compiler/python_plugin_test.py')
| -rw-r--r-- | test/compiler/python_plugin_test.py | 292 |
1 files changed, 153 insertions, 139 deletions
diff --git a/test/compiler/python_plugin_test.py b/test/compiler/python_plugin_test.py index 367effdb1a..0e58d912b9 100644 --- a/test/compiler/python_plugin_test.py +++ b/test/compiler/python_plugin_test.py @@ -36,6 +36,7 @@ import shutil import subprocess import sys import tempfile +import threading import time import unittest @@ -49,13 +50,13 @@ STUB_IDENTIFIER = 'EarlyAdopterTestServiceStub' SERVER_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_server' STUB_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_stub' -# Timeouts and delays. -SHORT_TIMEOUT = 0.1 -NORMAL_TIMEOUT = 1 -LONG_TIMEOUT = 2 -DOES_NOT_MATTER_DELAY = 0 +# 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 -LONG_DELAY = 1 # Build mode environment variable set by tools/run_tests/run_tests.py. _build_mode = os.environ['CONFIG'] @@ -64,47 +65,54 @@ _build_mode = os.environ['CONFIG'] class _ServicerMethods(object): def __init__(self, test_pb2, delay): + self._condition = threading.Condition() + self._delay = delay self._paused = False - self._failed = False - self.test_pb2 = test_pb2 - self.delay = delay + self._fail = False + self._test_pb2 = test_pb2 @contextlib.contextmanager def pause(self): # pylint: disable=invalid-name - self._paused = True + with self._condition: + self._paused = True yield - self._paused = False + with self._condition: + self._paused = False + self._condition.notify_all() @contextlib.contextmanager def fail(self): # pylint: disable=invalid-name - self._failed = True + with self._condition: + self._fail = True yield - self._failed = False + with self._condition: + self._fail = False def _control(self): # pylint: disable=invalid-name - if self._failed: - raise ValueError() - time.sleep(self.delay) - while self._paused: - time.sleep(0) - - def UnaryCall(self, request, unused_context): - response = self.test_pb2.SimpleResponse() - response.payload.payload_type = self.test_pb2.COMPRESSABLE + 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_context): + 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 = 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_context): - response = self.test_pb2.StreamingInputCallResponse() + 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) @@ -112,21 +120,21 @@ class _ServicerMethods(object): self._control() return response - def FullDuplexCall(self, request_iter, unused_context): + 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 = 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_context): + 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 = self._test_pb2.StreamingOutputCallResponse() + response.payload.payload_type = self._test_pb2.COMPRESSABLE response.payload.payload_compressable = 'a' * parameter.size self._control() responses.append(response) @@ -147,12 +155,11 @@ def _CreateService(test_pb2, delay): waiting for the service. Args: - test_pb2: the test_pb2 module generated by this test - delay: delay in seconds per response from the servicer - timeout: how long the stub will wait for the servicer by default. + test_pb2: The test_pb2 module generated by this test. + delay: Delay in seconds per response from the servicer. Yields: - A three-tuple (servicer_methods, servicer, stub), where the servicer is + 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. """ @@ -185,7 +192,7 @@ def _CreateService(test_pb2, delay): yield servicer_methods, stub, server -def StreamingInputRequest(test_pb2): +def _streaming_input_request_iterator(test_pb2): for _ in range(3): request = test_pb2.StreamingInputCallRequest() request.payload.payload_type = test_pb2.COMPRESSABLE @@ -193,7 +200,7 @@ def StreamingInputRequest(test_pb2): yield request -def StreamingOutputRequest(test_pb2): +def _streaming_output_request(test_pb2): request = test_pb2.StreamingOutputCallRequest() sizes = [1, 2, 3] request.response_parameters.add(size=sizes[0], interval_us=0) @@ -202,7 +209,7 @@ def StreamingOutputRequest(test_pb2): return request -def FullDuplexRequest(test_pb2): +def _full_duplex_request_iterator(test_pb2): request = test_pb2.StreamingOutputCallRequest() request.response_parameters.add(size=1, interval_us=0) yield request @@ -250,7 +257,7 @@ class PythonPluginTest(unittest.TestCase): if exc.errno != errno.ENOENT: raise - # TODO(atash): Figure out which of theses tests is hanging flakily with small + # TODO(atash): Figure out which of these tests is hanging flakily with small # probability. def testImportAttributes(self): @@ -265,37 +272,36 @@ class PythonPluginTest(unittest.TestCase): def testUpDown(self): import test_pb2 with _CreateService( - test_pb2, DOES_NOT_MATTER_DELAY) as (servicer, stub, unused_server): + 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 (servicer, stub, unused_server): + 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, NORMAL_TIMEOUT) - expected_response = servicer.UnaryCall(request, None) + 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, LONG_DELAY) as ( - servicer, stub, unused_server): - start_time = time.clock() - response_future = stub.UnaryCall.async(request, LONG_TIMEOUT) - # Check that we didn't block on the asynchronous call. - self.assertGreater(LONG_DELAY, time.clock() - start_time) + 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 = servicer.UnaryCall(request, None) + 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 - # set the timeout super low... - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): + with _CreateService(test_pb2, NO_DELAY) as ( + methods, stub, unused_server): request = test_pb2.SimpleRequest(response_size=13) - with servicer.pause(): + with methods.pause(): response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT) with self.assertRaises(exceptions.ExpirationError): response_future.result() @@ -305,9 +311,9 @@ class PythonPluginTest(unittest.TestCase): def testUnaryCallAsyncCancelled(self): import test_pb2 # pylint: disable=g-import-not-at-top request = test_pb2.SimpleRequest(response_size=13) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.pause(): + 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()) @@ -315,30 +321,31 @@ class PythonPluginTest(unittest.TestCase): def testUnaryCallAsyncFailed(self): import test_pb2 # pylint: disable=g-import-not-at-top request = test_pb2.SimpleRequest(response_size=13) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.fail(): - response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT) + 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 = StreamingOutputRequest(test_pb2) - with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): - responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT) - expected_responses = servicer.StreamingOutputCall(request, None) - for check in itertools.izip_longest(expected_responses, responses): - expected_response, response = check + 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 = StreamingOutputRequest(test_pb2) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.pause(): + 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) @@ -347,9 +354,9 @@ class PythonPluginTest(unittest.TestCase): 'forever and fix.') def testStreamingOutputCallCancelled(self): import test_pb2 # pylint: disable=g-import-not-at-top - request = StreamingOutputRequest(test_pb2) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - unused_servicer, stub, unused_server): + 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() @@ -360,10 +367,10 @@ class PythonPluginTest(unittest.TestCase): 'instead of raising the proper error.') def testStreamingOutputCallFailed(self): import test_pb2 # pylint: disable=g-import-not-at-top - request = StreamingOutputRequest(test_pb2) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.fail(): + 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): @@ -373,34 +380,32 @@ class PythonPluginTest(unittest.TestCase): 'forever and fix.') def testStreamingInputCall(self): import test_pb2 # pylint: disable=g-import-not-at-top - with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): - response = stub.StreamingInputCall(StreamingInputRequest(test_pb2), - NORMAL_TIMEOUT) - expected_response = servicer.StreamingInputCall( - StreamingInputRequest(test_pb2), None) + 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, LONG_DELAY) as ( - servicer, stub, unused_server): - start_time = time.clock() - response_future = stub.StreamingInputCall.async( - StreamingInputRequest(test_pb2), LONG_TIMEOUT) - self.assertGreater(LONG_DELAY, time.clock() - start_time) + 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 = servicer.StreamingInputCall( - StreamingInputRequest(test_pb2), None) + 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 - # set the timeout super low... - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.pause(): + with _CreateService(test_pb2, NO_DELAY) as ( + methods, stub, unused_server): + with methods.pause(): response_future = stub.StreamingInputCall.async( - StreamingInputRequest(test_pb2), SHORT_TIMEOUT) + _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT) with self.assertRaises(exceptions.ExpirationError): response_future.result() self.assertIsInstance( @@ -408,11 +413,12 @@ class PythonPluginTest(unittest.TestCase): def testStreamingInputCallAsyncCancelled(self): import test_pb2 # pylint: disable=g-import-not-at-top - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.pause(): + 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( - StreamingInputRequest(test_pb2), NORMAL_TIMEOUT) + _streaming_input_request_iterator(test_pb2), timeout) response_future.cancel() self.assertTrue(response_future.cancelled()) with self.assertRaises(future.CancelledError): @@ -420,33 +426,33 @@ class PythonPluginTest(unittest.TestCase): def testStreamingInputCallAsyncFailed(self): import test_pb2 # pylint: disable=g-import-not-at-top - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.fail(): + with _CreateService(test_pb2, NO_DELAY) as ( + methods, stub, unused_server): + with methods.fail(): response_future = stub.StreamingInputCall.async( - StreamingInputRequest(test_pb2), SHORT_TIMEOUT) + _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 (servicer, stub, unused_server): - responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2), - NORMAL_TIMEOUT) - expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2), - None) - for check in itertools.izip_longest(expected_responses, responses): - expected_response, response = check + 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 = FullDuplexRequest(test_pb2) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.pause(): - responses = stub.FullDuplexCall(request, SHORT_TIMEOUT) + 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) @@ -454,9 +460,9 @@ class PythonPluginTest(unittest.TestCase): 'forever and fix.') def testFullDuplexCallCancelled(self): import test_pb2 # pylint: disable=g-import-not-at-top - with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): - request = FullDuplexRequest(test_pb2) - responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT) + 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): @@ -466,11 +472,11 @@ class PythonPluginTest(unittest.TestCase): 'and fix.') def testFullDuplexCallFailed(self): import test_pb2 # pylint: disable=g-import-not-at-top - request = FullDuplexRequest(test_pb2) - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - with servicer.fail(): - responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT) + 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) @@ -479,9 +485,9 @@ class PythonPluginTest(unittest.TestCase): 'forever and fix.') def testHalfDuplexCall(self): import test_pb2 # pylint: disable=g-import-not-at-top - with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( - servicer, stub, unused_server): - def HalfDuplexRequest(): + 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 @@ -489,30 +495,38 @@ class PythonPluginTest(unittest.TestCase): request.response_parameters.add(size=2, interval_us=0) request.response_parameters.add(size=3, interval_us=0) yield request - responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT) - expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest(), None) + 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 - wait_flag = [False] + 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? - wait_flag[0] = True + with condition: + wait_cell[0] = True yield - wait_flag[0] = False - def HalfDuplexRequest(): + 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 - while wait_flag[0]: - time.sleep(0.1) - with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): + with condition: + while wait_cell[0]: + condition.wait() + with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server): with wait(): - responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT) + 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) |