Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Cosmetic tweaks to python_plugin_test.py

Browse Source 
No behavioral changes. Changes to comments are error correction, not
changes of specification.
pull/2024/head
Nathaniel Manista 10 years ago
parent a561ea66ae
commit 26a5bfad2f
1 changed files with 93 additions and 89 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 182
      test/compiler/python_plugin_test.py

182
test/compiler/python_plugin_test.py
Unescape View File

@ -66,8 +66,8 @@ class _ServicerMethods(object):
def __init__(self, test_pb2, delay): def __init__(self, test_pb2, delay):
self._paused = False self._paused = False
self._failed = False self._failed = False
self.test_pb2 = test_pb2 self._test_pb2 = test_pb2
self.delay = delay self._delay = delay
@contextlib.contextmanager @contextlib.contextmanager
def pause(self): # pylint: disable=invalid-name def pause(self): # pylint: disable=invalid-name
@ -84,27 +84,27 @@ class _ServicerMethods(object):
def _control(self): # pylint: disable=invalid-name def _control(self): # pylint: disable=invalid-name
if self._failed: if self._failed:
raise ValueError() raise ValueError()
time.sleep(self.delay) time.sleep(self._delay)
while self._paused: while self._paused:
time.sleep(0) time.sleep(0)
def UnaryCall(self, request, unused_context): def UnaryCall(self, request, unused_rpc_context):
response = self.test_pb2.SimpleResponse() response = self._test_pb2.SimpleResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE response.payload.payload_type = self._test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * request.response_size response.payload.payload_compressable = 'a' * request.response_size
self._control() self._control()
return response return response
def StreamingOutputCall(self, request, unused_context): def StreamingOutputCall(self, request, unused_rpc_context):
for parameter in request.response_parameters: for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse() response = self._test_pb2.StreamingOutputCallResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE response.payload.payload_type = self._test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size response.payload.payload_compressable = 'a' * parameter.size
self._control() self._control()
yield response yield response
def StreamingInputCall(self, request_iter, unused_context): def StreamingInputCall(self, request_iter, unused_rpc_context):
response = self.test_pb2.StreamingInputCallResponse() response = self._test_pb2.StreamingInputCallResponse()
aggregated_payload_size = 0 aggregated_payload_size = 0
for request in request_iter: for request in request_iter:
aggregated_payload_size += len(request.payload.payload_compressable) aggregated_payload_size += len(request.payload.payload_compressable)
@ -112,21 +112,21 @@ class _ServicerMethods(object):
self._control() self._control()
return response return response
def FullDuplexCall(self, request_iter, unused_context): def FullDuplexCall(self, request_iter, unused_rpc_context):
for request in request_iter: for request in request_iter:
for parameter in request.response_parameters: for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse() response = self._test_pb2.StreamingOutputCallResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE response.payload.payload_type = self._test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size response.payload.payload_compressable = 'a' * parameter.size
self._control() self._control()
yield response yield response
def HalfDuplexCall(self, request_iter, unused_context): def HalfDuplexCall(self, request_iter, unused_rpc_context):
responses = [] responses = []
for request in request_iter: for request in request_iter:
for parameter in request.response_parameters: for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse() response = self._test_pb2.StreamingOutputCallResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE response.payload.payload_type = self._test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * parameter.size response.payload.payload_compressable = 'a' * parameter.size
self._control() self._control()
responses.append(response) responses.append(response)
@ -152,7 +152,7 @@ def _CreateService(test_pb2, delay):
timeout: how long the stub will wait for the servicer by default. timeout: how long the stub will wait for the servicer by default.
Yields: 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 the back-end of the service bound to the stub and the server and stub
are both activated and ready for use. are both activated and ready for use.
""" """
@ -185,7 +185,7 @@ def _CreateService(test_pb2, delay):
yield servicer_methods, stub, server yield servicer_methods, stub, server
def StreamingInputRequest(test_pb2): def _streaming_input_request_iterator(test_pb2):
for _ in range(3): for _ in range(3):
request = test_pb2.StreamingInputCallRequest() request = test_pb2.StreamingInputCallRequest()
request.payload.payload_type = test_pb2.COMPRESSABLE request.payload.payload_type = test_pb2.COMPRESSABLE
@ -193,7 +193,7 @@ def StreamingInputRequest(test_pb2):
yield request yield request
def StreamingOutputRequest(test_pb2): def _streaming_output_request(test_pb2):
request = test_pb2.StreamingOutputCallRequest() request = test_pb2.StreamingOutputCallRequest()
sizes = [1, 2, 3] sizes = [1, 2, 3]
request.response_parameters.add(size=sizes[0], interval_us=0) request.response_parameters.add(size=sizes[0], interval_us=0)
@ -202,7 +202,7 @@ def StreamingOutputRequest(test_pb2):
return request return request
def FullDuplexRequest(test_pb2): def _full_duplex_request_iterator(test_pb2):
request = test_pb2.StreamingOutputCallRequest() request = test_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0) request.response_parameters.add(size=1, interval_us=0)
yield request yield request
@ -270,32 +270,32 @@ class PythonPluginTest(unittest.TestCase):
def testUnaryCall(self): def testUnaryCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top 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):
request = test_pb2.SimpleRequest(response_size=13) request = test_pb2.SimpleRequest(response_size=13)
response = stub.UnaryCall(request, NORMAL_TIMEOUT) response = stub.UnaryCall(request, NORMAL_TIMEOUT)
expected_response = servicer.UnaryCall(request, None) expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
def testUnaryCallAsync(self): def testUnaryCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = test_pb2.SimpleRequest(response_size=13) request = test_pb2.SimpleRequest(response_size=13)
with _CreateService(test_pb2, LONG_DELAY) as ( with _CreateService(test_pb2, LONG_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
start_time = time.clock() start_time = time.clock()
response_future = stub.UnaryCall.async(request, LONG_TIMEOUT) response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
# Check that we didn't block on the asynchronous call. # Check that we didn't block on the asynchronous call.
self.assertGreater(LONG_DELAY, time.clock() - start_time) self.assertGreater(LONG_DELAY, time.clock() - start_time)
response = response_future.result() response = response_future.result()
expected_response = servicer.UnaryCall(request, None) expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
def testUnaryCallAsyncExpired(self): def testUnaryCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low... # set the timeout super low...
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
request = test_pb2.SimpleRequest(response_size=13) request = test_pb2.SimpleRequest(response_size=13)
with servicer.pause(): with methods.pause():
response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT) response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
response_future.result() response_future.result()
@ -306,8 +306,8 @@ class PythonPluginTest(unittest.TestCase):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = test_pb2.SimpleRequest(response_size=13) request = test_pb2.SimpleRequest(response_size=13)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.pause(): with methods.pause():
response_future = stub.UnaryCall.async(request, 1) response_future = stub.UnaryCall.async(request, 1)
response_future.cancel() response_future.cancel()
self.assertTrue(response_future.cancelled()) self.assertTrue(response_future.cancelled())
@ -316,29 +316,30 @@ class PythonPluginTest(unittest.TestCase):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = test_pb2.SimpleRequest(response_size=13) request = test_pb2.SimpleRequest(response_size=13)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.fail(): with methods.fail():
response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT) response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT)
self.assertIsNotNone(response_future.exception()) self.assertIsNotNone(response_future.exception())
def testStreamingOutputCall(self): def testStreamingOutputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2) request = _streaming_output_request(test_pb2)
with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT) responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT)
expected_responses = servicer.StreamingOutputCall(request, None) expected_responses = methods.StreamingOutputCall(
for check in itertools.izip_longest(expected_responses, responses): request, 'not a real RpcContext!')
expected_response, response = check for expected_response, response in itertools.izip_longest(
expected_responses, responses):
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
@unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
'forever and fix.') 'forever and fix.')
def testStreamingOutputCallExpired(self): def testStreamingOutputCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2) request = _streaming_output_request(test_pb2)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.pause(): with methods.pause():
responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT) responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
list(responses) list(responses)
@ -347,9 +348,9 @@ class PythonPluginTest(unittest.TestCase):
'forever and fix.') 'forever and fix.')
def testStreamingOutputCallCancelled(self): def testStreamingOutputCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2) request = _streaming_output_request(test_pb2)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
unused_servicer, stub, unused_server): unused_methods, stub, unused_server):
responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT) responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
next(responses) next(responses)
responses.cancel() responses.cancel()
@ -360,10 +361,10 @@ class PythonPluginTest(unittest.TestCase):
'instead of raising the proper error.') 'instead of raising the proper error.')
def testStreamingOutputCallFailed(self): def testStreamingOutputCallFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = StreamingOutputRequest(test_pb2) request = _streaming_output_request(test_pb2)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.fail(): with methods.fail():
responses = stub.StreamingOutputCall(request, 1) responses = stub.StreamingOutputCall(request, 1)
self.assertIsNotNone(responses) self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError): with self.assertRaises(exceptions.ServicerError):
@ -373,34 +374,34 @@ class PythonPluginTest(unittest.TestCase):
'forever and fix.') 'forever and fix.')
def testStreamingInputCall(self): def testStreamingInputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top 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):
response = stub.StreamingInputCall(StreamingInputRequest(test_pb2), response = stub.StreamingInputCall(StreamingInputRequest(test_pb2),
NORMAL_TIMEOUT) NORMAL_TIMEOUT)
expected_response = servicer.StreamingInputCall( expected_response = methods.StreamingInputCall(
StreamingInputRequest(test_pb2), None) _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!')
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
def testStreamingInputCallAsync(self): def testStreamingInputCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, LONG_DELAY) as ( with _CreateService(test_pb2, LONG_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
start_time = time.clock() start_time = time.clock()
response_future = stub.StreamingInputCall.async( response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), LONG_TIMEOUT) _streaming_input_request_iterator(test_pb2), LONG_TIMEOUT)
self.assertGreater(LONG_DELAY, time.clock() - start_time) self.assertGreater(LONG_DELAY, time.clock() - start_time)
response = response_future.result() response = response_future.result()
expected_response = servicer.StreamingInputCall( expected_response = methods.StreamingInputCall(
StreamingInputRequest(test_pb2), None) _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!')
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
def testStreamingInputCallAsyncExpired(self): def testStreamingInputCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low... # set the timeout super low...
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.pause(): with methods.pause():
response_future = stub.StreamingInputCall.async( response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), SHORT_TIMEOUT) _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
response_future.result() response_future.result()
self.assertIsInstance( self.assertIsInstance(
@ -409,10 +410,10 @@ class PythonPluginTest(unittest.TestCase):
def testStreamingInputCallAsyncCancelled(self): def testStreamingInputCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.pause(): with methods.pause():
response_future = stub.StreamingInputCall.async( response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), NORMAL_TIMEOUT) _streaming_input_request_iterator(test_pb2), NORMAL_TIMEOUT)
response_future.cancel() response_future.cancel()
self.assertTrue(response_future.cancelled()) self.assertTrue(response_future.cancelled())
with self.assertRaises(future.CancelledError): with self.assertRaises(future.CancelledError):
@ -421,32 +422,32 @@ class PythonPluginTest(unittest.TestCase):
def testStreamingInputCallAsyncFailed(self): def testStreamingInputCallAsyncFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.fail(): with methods.fail():
response_future = stub.StreamingInputCall.async( response_future = stub.StreamingInputCall.async(
StreamingInputRequest(test_pb2), SHORT_TIMEOUT) _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT)
self.assertIsNotNone(response_future.exception()) self.assertIsNotNone(response_future.exception())
def testFullDuplexCall(self): def testFullDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top 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):
responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2), responses = stub.FullDuplexCall(
NORMAL_TIMEOUT) _full_duplex_request_iterator(test_pb2), NORMAL_TIMEOUT)
expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2), expected_responses = methods.FullDuplexCall(
None) _full_duplex_request_iterator(test_pb2), 'not a real RpcContext!')
for check in itertools.izip_longest(expected_responses, responses): for expected_response, response in itertools.izip_longest(
expected_response, response = check expected_responses, responses):
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
@unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs ' @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
'forever and fix.') 'forever and fix.')
def testFullDuplexCallExpired(self): def testFullDuplexCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = FullDuplexRequest(test_pb2) request_iterator = _full_duplex_request_iterator(test_pb2)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.pause(): with methods.pause():
responses = stub.FullDuplexCall(request, SHORT_TIMEOUT) responses = stub.FullDuplexCall(request_iterator, SHORT_TIMEOUT)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
list(responses) list(responses)
@ -454,9 +455,9 @@ class PythonPluginTest(unittest.TestCase):
'forever and fix.') 'forever and fix.')
def testFullDuplexCallCancelled(self): def testFullDuplexCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top 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):
request = FullDuplexRequest(test_pb2) request_iterator = _full_duplex_request_iterator(test_pb2)
responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT) responses = stub.FullDuplexCall(request_iterator, NORMAL_TIMEOUT)
next(responses) next(responses)
responses.cancel() responses.cancel()
with self.assertRaises(future.CancelledError): with self.assertRaises(future.CancelledError):
@ -466,11 +467,11 @@ class PythonPluginTest(unittest.TestCase):
'and fix.') 'and fix.')
def testFullDuplexCallFailed(self): def testFullDuplexCallFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
request = FullDuplexRequest(test_pb2) request_iterator = _full_duplex_request_iterator(test_pb2)
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
with servicer.fail(): with methods.fail():
responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT) responses = stub.FullDuplexCall(request_iterator, NORMAL_TIMEOUT)
self.assertIsNotNone(responses) self.assertIsNotNone(responses)
with self.assertRaises(exceptions.ServicerError): with self.assertRaises(exceptions.ServicerError):
next(responses) next(responses)
@ -480,8 +481,8 @@ class PythonPluginTest(unittest.TestCase):
def testHalfDuplexCall(self): def testHalfDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as ( with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
servicer, stub, unused_server): methods, stub, unused_server):
def HalfDuplexRequest(): def half_duplex_request_iterator():
request = test_pb2.StreamingOutputCallRequest() request = test_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0) request.response_parameters.add(size=1, interval_us=0)
yield request yield request
@ -489,30 +490,33 @@ class PythonPluginTest(unittest.TestCase):
request.response_parameters.add(size=2, interval_us=0) request.response_parameters.add(size=2, interval_us=0)
request.response_parameters.add(size=3, interval_us=0) request.response_parameters.add(size=3, interval_us=0)
yield request yield request
responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT) responses = stub.HalfDuplexCall(
expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest(), None) half_duplex_request_iterator(), NORMAL_TIMEOUT)
expected_responses = methods.HalfDuplexCall(
HalfDuplexRequest(), 'not a real RpcContext!')
for check in itertools.izip_longest(expected_responses, responses): for check in itertools.izip_longest(expected_responses, responses):
expected_response, response = check expected_response, response = check
self.assertEqual(expected_response, response) self.assertEqual(expected_response, response)
def testHalfDuplexCallWedged(self): def testHalfDuplexCallWedged(self):
import test_pb2 # pylint: disable=g-import-not-at-top import test_pb2 # pylint: disable=g-import-not-at-top
wait_flag = [False] wait_cell = [False]
@contextlib.contextmanager @contextlib.contextmanager
def wait(): # pylint: disable=invalid-name def wait(): # pylint: disable=invalid-name
# Where's Python 3's 'nonlocal' statement when you need it? # Where's Python 3's 'nonlocal' statement when you need it?
wait_flag[0] = True wait_cell[0] = True
yield yield
wait_flag[0] = False wait_cell[0] = False
def HalfDuplexRequest(): def half_duplex_request_iterator():
request = test_pb2.StreamingOutputCallRequest() request = test_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0) request.response_parameters.add(size=1, interval_us=0)
yield request yield request
while wait_flag[0]: while wait_cell[0]:
time.sleep(0.1) time.sleep(0.1)
with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server): with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
with wait(): with wait():
responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT) responses = stub.HalfDuplexCall(
half_duplex_request_iterator(), NORMAL_TIMEOUT)
# half-duplex waits for the client to send all info # half-duplex waits for the client to send all info
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
next(responses) next(responses)

Write Preview
Loading…
Cancel
Save