|
|
|
|
@ -23,6 +23,7 @@ import grpc |
|
|
|
|
from grpc.framework.foundation import logging_pool |
|
|
|
|
|
|
|
|
|
from tests.unit import test_common |
|
|
|
|
from tests.unit import thread_pool |
|
|
|
|
from tests.unit.framework.common import test_constants |
|
|
|
|
from tests.unit.framework.common import test_control |
|
|
|
|
|
|
|
|
|
@ -33,8 +34,10 @@ _DESERIALIZE_RESPONSE = lambda bytestring: bytestring[:len(bytestring) // 3] |
|
|
|
|
|
|
|
|
|
_UNARY_UNARY = '/test/UnaryUnary' |
|
|
|
|
_UNARY_STREAM = '/test/UnaryStream' |
|
|
|
|
_UNARY_STREAM_NON_BLOCKING = '/test/UnaryStreamNonBlocking' |
|
|
|
|
_STREAM_UNARY = '/test/StreamUnary' |
|
|
|
|
_STREAM_STREAM = '/test/StreamStream' |
|
|
|
|
_STREAM_STREAM_NON_BLOCKING = '/test/StreamStreamNonBlocking' |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class _Callback(object): |
|
|
|
|
@ -59,8 +62,14 @@ class _Callback(object): |
|
|
|
|
|
|
|
|
|
class _Handler(object): |
|
|
|
|
|
|
|
|
|
def __init__(self, control): |
|
|
|
|
def __init__(self, control, thread_pool): |
|
|
|
|
self._control = control |
|
|
|
|
self._thread_pool = thread_pool |
|
|
|
|
non_blocking_functions = (self.handle_unary_stream_non_blocking, |
|
|
|
|
self.handle_stream_stream_non_blocking) |
|
|
|
|
for non_blocking_function in non_blocking_functions: |
|
|
|
|
non_blocking_function.__func__.experimental_non_blocking = True |
|
|
|
|
non_blocking_function.__func__.experimental_thread_pool = self._thread_pool |
|
|
|
|
|
|
|
|
|
def handle_unary_unary(self, request, servicer_context): |
|
|
|
|
self._control.control() |
|
|
|
|
@ -87,6 +96,19 @@ class _Handler(object): |
|
|
|
|
'testvalue', |
|
|
|
|
),)) |
|
|
|
|
|
|
|
|
|
def handle_unary_stream_non_blocking(self, request, servicer_context, |
|
|
|
|
on_next): |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH): |
|
|
|
|
self._control.control() |
|
|
|
|
on_next(request) |
|
|
|
|
self._control.control() |
|
|
|
|
if servicer_context is not None: |
|
|
|
|
servicer_context.set_trailing_metadata((( |
|
|
|
|
'testkey', |
|
|
|
|
'testvalue', |
|
|
|
|
),)) |
|
|
|
|
on_next(None) |
|
|
|
|
|
|
|
|
|
def handle_stream_unary(self, request_iterator, servicer_context): |
|
|
|
|
if servicer_context is not None: |
|
|
|
|
servicer_context.invocation_metadata() |
|
|
|
|
@ -115,6 +137,20 @@ class _Handler(object): |
|
|
|
|
yield request |
|
|
|
|
self._control.control() |
|
|
|
|
|
|
|
|
|
def handle_stream_stream_non_blocking(self, request_iterator, |
|
|
|
|
servicer_context, on_next): |
|
|
|
|
self._control.control() |
|
|
|
|
if servicer_context is not None: |
|
|
|
|
servicer_context.set_trailing_metadata((( |
|
|
|
|
'testkey', |
|
|
|
|
'testvalue', |
|
|
|
|
),)) |
|
|
|
|
for request in request_iterator: |
|
|
|
|
self._control.control() |
|
|
|
|
on_next(request) |
|
|
|
|
self._control.control() |
|
|
|
|
on_next(None) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class _MethodHandler(grpc.RpcMethodHandler): |
|
|
|
|
|
|
|
|
|
@ -145,6 +181,10 @@ class _GenericHandler(grpc.GenericRpcHandler): |
|
|
|
|
return _MethodHandler(False, True, _DESERIALIZE_REQUEST, |
|
|
|
|
_SERIALIZE_RESPONSE, None, |
|
|
|
|
self._handler.handle_unary_stream, None, None) |
|
|
|
|
elif handler_call_details.method == _UNARY_STREAM_NON_BLOCKING: |
|
|
|
|
return _MethodHandler( |
|
|
|
|
False, True, _DESERIALIZE_REQUEST, _SERIALIZE_RESPONSE, None, |
|
|
|
|
self._handler.handle_unary_stream_non_blocking, None, None) |
|
|
|
|
elif handler_call_details.method == _STREAM_UNARY: |
|
|
|
|
return _MethodHandler(True, False, _DESERIALIZE_REQUEST, |
|
|
|
|
_SERIALIZE_RESPONSE, None, None, |
|
|
|
|
@ -152,6 +192,10 @@ class _GenericHandler(grpc.GenericRpcHandler): |
|
|
|
|
elif handler_call_details.method == _STREAM_STREAM: |
|
|
|
|
return _MethodHandler(True, True, None, None, None, None, None, |
|
|
|
|
self._handler.handle_stream_stream) |
|
|
|
|
elif handler_call_details.method == _STREAM_STREAM_NON_BLOCKING: |
|
|
|
|
return _MethodHandler( |
|
|
|
|
True, True, None, None, None, None, None, |
|
|
|
|
self._handler.handle_stream_stream_non_blocking) |
|
|
|
|
else: |
|
|
|
|
return None |
|
|
|
|
|
|
|
|
|
@ -167,6 +211,13 @@ def _unary_stream_multi_callable(channel): |
|
|
|
|
response_deserializer=_DESERIALIZE_RESPONSE) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _unary_stream_non_blocking_multi_callable(channel): |
|
|
|
|
return channel.unary_stream( |
|
|
|
|
_UNARY_STREAM_NON_BLOCKING, |
|
|
|
|
request_serializer=_SERIALIZE_REQUEST, |
|
|
|
|
response_deserializer=_DESERIALIZE_RESPONSE) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _stream_unary_multi_callable(channel): |
|
|
|
|
return channel.stream_unary( |
|
|
|
|
_STREAM_UNARY, |
|
|
|
|
@ -178,11 +229,16 @@ def _stream_stream_multi_callable(channel): |
|
|
|
|
return channel.stream_stream(_STREAM_STREAM) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _stream_stream_non_blocking_multi_callable(channel): |
|
|
|
|
return channel.stream_stream(_STREAM_STREAM_NON_BLOCKING) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class RPCTest(unittest.TestCase): |
|
|
|
|
|
|
|
|
|
def setUp(self): |
|
|
|
|
self._control = test_control.PauseFailControl() |
|
|
|
|
self._handler = _Handler(self._control) |
|
|
|
|
self._thread_pool = thread_pool.RecordingThreadPool(max_workers=None) |
|
|
|
|
self._handler = _Handler(self._control, self._thread_pool) |
|
|
|
|
|
|
|
|
|
self._server = test_common.test_server() |
|
|
|
|
port = self._server.add_insecure_port('[::]:0') |
|
|
|
|
@ -195,6 +251,16 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self._server.stop(None) |
|
|
|
|
self._channel.close() |
|
|
|
|
|
|
|
|
|
def testDefaultThreadPoolIsUsed(self): |
|
|
|
|
self._consume_one_stream_response_unary_request( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
self.assertFalse(self._thread_pool.was_used()) |
|
|
|
|
|
|
|
|
|
def testExperimentalThreadPoolIsUsed(self): |
|
|
|
|
self._consume_one_stream_response_unary_request( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
self.assertTrue(self._thread_pool.was_used()) |
|
|
|
|
|
|
|
|
|
def testUnrecognizedMethod(self): |
|
|
|
|
request = b'abc' |
|
|
|
|
|
|
|
|
|
@ -227,7 +293,7 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
|
|
|
|
|
self.assertEqual(expected_response, response) |
|
|
|
|
self.assertIs(grpc.StatusCode.OK, call.code()) |
|
|
|
|
self.assertEqual("", call.debug_error_string()) |
|
|
|
|
self.assertEqual('', call.debug_error_string()) |
|
|
|
|
|
|
|
|
|
def testSuccessfulUnaryRequestFutureUnaryResponse(self): |
|
|
|
|
request = b'\x07\x08' |
|
|
|
|
@ -310,6 +376,7 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
def testSuccessfulStreamRequestStreamResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x77\x58' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
|
|
|
|
|
expected_responses = tuple( |
|
|
|
|
self._handler.handle_stream_stream(iter(requests), None)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
@ -425,58 +492,36 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
test_is_running_cell[0] = False |
|
|
|
|
|
|
|
|
|
def testConsumingOneStreamResponseUnaryRequest(self): |
|
|
|
|
request = b'\x57\x38' |
|
|
|
|
self._consume_one_stream_response_unary_request( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'ConsumingOneStreamResponseUnaryRequest'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
def testConsumingOneStreamResponseUnaryRequestNonBlocking(self): |
|
|
|
|
self._consume_one_stream_response_unary_request( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testConsumingSomeButNotAllStreamResponsesUnaryRequest(self): |
|
|
|
|
request = b'\x57\x38' |
|
|
|
|
self._consume_some_but_not_all_stream_responses_unary_request( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingSomeButNotAllStreamResponsesUnaryRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH // 2): |
|
|
|
|
next(response_iterator) |
|
|
|
|
def testConsumingSomeButNotAllStreamResponsesUnaryRequestNonBlocking(self): |
|
|
|
|
self._consume_some_but_not_all_stream_responses_unary_request( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testConsumingSomeButNotAllStreamResponsesStreamRequest(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
self._consume_some_but_not_all_stream_responses_stream_request( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingSomeButNotAllStreamResponsesStreamRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH // 2): |
|
|
|
|
next(response_iterator) |
|
|
|
|
def testConsumingSomeButNotAllStreamResponsesStreamRequestNonBlocking(self): |
|
|
|
|
self._consume_some_but_not_all_stream_responses_stream_request( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testConsumingTooManyStreamResponsesStreamRequest(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
self._consume_too_many_stream_responses_stream_request( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingTooManyStreamResponsesStreamRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH): |
|
|
|
|
next(response_iterator) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH): |
|
|
|
|
with self.assertRaises(StopIteration): |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.OK, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
def testConsumingTooManyStreamResponsesStreamRequestNonBlocking(self): |
|
|
|
|
self._consume_too_many_stream_responses_stream_request( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testCancelledUnaryRequestUnaryResponse(self): |
|
|
|
|
request = b'\x07\x17' |
|
|
|
|
@ -498,24 +543,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, response_future.code()) |
|
|
|
|
|
|
|
|
|
def testCancelledUnaryRequestStreamResponse(self): |
|
|
|
|
request = b'\x07\x19' |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
with self._control.pause(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'CancelledUnaryRequestStreamResponse'),)) |
|
|
|
|
self._control.block_until_paused() |
|
|
|
|
response_iterator.cancel() |
|
|
|
|
self._cancelled_unary_request_stream_response( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
next(response_iterator) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
def testCancelledUnaryRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._cancelled_unary_request_stream_response( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testCancelledStreamRequestUnaryResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
@ -543,23 +576,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self.assertIsNotNone(response_future.trailing_metadata()) |
|
|
|
|
|
|
|
|
|
def testCancelledStreamRequestStreamResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
self._cancelled_stream_request_stream_response( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
with self._control.pause(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', 'CancelledStreamRequestStreamResponse'),)) |
|
|
|
|
response_iterator.cancel() |
|
|
|
|
|
|
|
|
|
with self.assertRaises(grpc.RpcError): |
|
|
|
|
next(response_iterator) |
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
def testCancelledStreamRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._cancelled_stream_request_stream_response( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testExpiredUnaryRequestBlockingUnaryResponse(self): |
|
|
|
|
request = b'\x07\x17' |
|
|
|
|
@ -608,21 +630,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
response_future.exception().code()) |
|
|
|
|
|
|
|
|
|
def testExpiredUnaryRequestStreamResponse(self): |
|
|
|
|
request = b'\x07\x19' |
|
|
|
|
self._expired_unary_request_stream_response( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
with self._control.pause(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
timeout=test_constants.SHORT_TIMEOUT, |
|
|
|
|
metadata=(('test', 'ExpiredUnaryRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
response_iterator.code()) |
|
|
|
|
def testExpiredUnaryRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._expired_unary_request_stream_response( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testExpiredStreamRequestBlockingUnaryResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
@ -678,23 +691,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self.assertIsNotNone(response_future.trailing_metadata()) |
|
|
|
|
|
|
|
|
|
def testExpiredStreamRequestStreamResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x18' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
with self._control.pause(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
timeout=test_constants.SHORT_TIMEOUT, |
|
|
|
|
metadata=(('test', 'ExpiredStreamRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
self._expired_stream_request_stream_response( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
response_iterator.code()) |
|
|
|
|
def testExpiredStreamRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._expired_stream_request_stream_response( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testFailedUnaryRequestBlockingUnaryResponse(self): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
@ -712,10 +714,10 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
# sanity checks on to make sure returned string contains default members |
|
|
|
|
# of the error |
|
|
|
|
debug_error_string = exception_context.exception.debug_error_string() |
|
|
|
|
self.assertIn("created", debug_error_string) |
|
|
|
|
self.assertIn("description", debug_error_string) |
|
|
|
|
self.assertIn("file", debug_error_string) |
|
|
|
|
self.assertIn("file_line", debug_error_string) |
|
|
|
|
self.assertIn('created', debug_error_string) |
|
|
|
|
self.assertIn('description', debug_error_string) |
|
|
|
|
self.assertIn('file', debug_error_string) |
|
|
|
|
self.assertIn('file_line', debug_error_string) |
|
|
|
|
|
|
|
|
|
def testFailedUnaryRequestFutureUnaryResponse(self): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
@ -742,18 +744,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self.assertIs(response_future, value_passed_to_callback) |
|
|
|
|
|
|
|
|
|
def testFailedUnaryRequestStreamResponse(self): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
self._failed_unary_request_stream_response( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
with self._control.fail(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'FailedUnaryRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
def testFailedUnaryRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._failed_unary_request_stream_response( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testFailedStreamRequestBlockingUnaryResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
@ -795,21 +791,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
self.assertIs(response_future, value_passed_to_callback) |
|
|
|
|
|
|
|
|
|
def testFailedStreamRequestStreamResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
self._failed_stream_request_stream_response( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
with self._control.fail(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', 'FailedStreamRequestStreamResponse'),)) |
|
|
|
|
tuple(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, response_iterator.code()) |
|
|
|
|
def testFailedStreamRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._failed_stream_request_stream_response( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testIgnoredUnaryRequestFutureUnaryResponse(self): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
@ -820,11 +807,12 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
metadata=(('test', 'IgnoredUnaryRequestFutureUnaryResponse'),)) |
|
|
|
|
|
|
|
|
|
def testIgnoredUnaryRequestStreamResponse(self): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
self._ignored_unary_stream_request_future_unary_response( |
|
|
|
|
_unary_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
multi_callable = _unary_stream_multi_callable(self._channel) |
|
|
|
|
multi_callable( |
|
|
|
|
request, metadata=(('test', 'IgnoredUnaryRequestStreamResponse'),)) |
|
|
|
|
def testIgnoredUnaryRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._ignored_unary_stream_request_future_unary_response( |
|
|
|
|
_unary_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testIgnoredStreamRequestFutureUnaryResponse(self): |
|
|
|
|
requests = tuple( |
|
|
|
|
@ -837,11 +825,177 @@ class RPCTest(unittest.TestCase): |
|
|
|
|
metadata=(('test', 'IgnoredStreamRequestFutureUnaryResponse'),)) |
|
|
|
|
|
|
|
|
|
def testIgnoredStreamRequestStreamResponse(self): |
|
|
|
|
self._ignored_stream_request_stream_response( |
|
|
|
|
_stream_stream_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def testIgnoredStreamRequestStreamResponseNonBlocking(self): |
|
|
|
|
self._ignored_stream_request_stream_response( |
|
|
|
|
_stream_stream_non_blocking_multi_callable(self._channel)) |
|
|
|
|
|
|
|
|
|
def _consume_one_stream_response_unary_request(self, multi_callable): |
|
|
|
|
request = b'\x57\x38' |
|
|
|
|
|
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'ConsumingOneStreamResponseUnaryRequest'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
def _consume_some_but_not_all_stream_responses_unary_request( |
|
|
|
|
self, multi_callable): |
|
|
|
|
request = b'\x57\x38' |
|
|
|
|
|
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingSomeButNotAllStreamResponsesUnaryRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH // 2): |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
def _consume_some_but_not_all_stream_responses_stream_request( |
|
|
|
|
self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingSomeButNotAllStreamResponsesStreamRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH // 2): |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
def _consume_too_many_stream_responses_stream_request(self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', |
|
|
|
|
'ConsumingTooManyStreamResponsesStreamRequest'),)) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH): |
|
|
|
|
next(response_iterator) |
|
|
|
|
for _ in range(test_constants.STREAM_LENGTH): |
|
|
|
|
with self.assertRaises(StopIteration): |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.OK, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
|
|
|
|
|
def _cancelled_unary_request_stream_response(self, multi_callable): |
|
|
|
|
request = b'\x07\x19' |
|
|
|
|
|
|
|
|
|
with self._control.pause(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'CancelledUnaryRequestStreamResponse'),)) |
|
|
|
|
self._control.block_until_paused() |
|
|
|
|
response_iterator.cancel() |
|
|
|
|
|
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
next(response_iterator) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
|
|
|
|
|
def _cancelled_stream_request_stream_response(self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
with self._control.pause(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', 'CancelledStreamRequestStreamResponse'),)) |
|
|
|
|
response_iterator.cancel() |
|
|
|
|
|
|
|
|
|
with self.assertRaises(grpc.RpcError): |
|
|
|
|
next(response_iterator) |
|
|
|
|
self.assertIsNotNone(response_iterator.initial_metadata()) |
|
|
|
|
self.assertIs(grpc.StatusCode.CANCELLED, response_iterator.code()) |
|
|
|
|
self.assertIsNotNone(response_iterator.details()) |
|
|
|
|
self.assertIsNotNone(response_iterator.trailing_metadata()) |
|
|
|
|
|
|
|
|
|
def _expired_unary_request_stream_response(self, multi_callable): |
|
|
|
|
request = b'\x07\x19' |
|
|
|
|
|
|
|
|
|
with self._control.pause(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
timeout=test_constants.SHORT_TIMEOUT, |
|
|
|
|
metadata=(('test', 'ExpiredUnaryRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
response_iterator.code()) |
|
|
|
|
|
|
|
|
|
def _expired_stream_request_stream_response(self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x18' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
with self._control.pause(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
timeout=test_constants.SHORT_TIMEOUT, |
|
|
|
|
metadata=(('test', 'ExpiredStreamRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, |
|
|
|
|
response_iterator.code()) |
|
|
|
|
|
|
|
|
|
def _failed_unary_request_stream_response(self, multi_callable): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
|
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
with self._control.fail(): |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request, |
|
|
|
|
metadata=(('test', 'FailedUnaryRequestStreamResponse'),)) |
|
|
|
|
next(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
|
|
|
|
|
def _failed_stream_request_stream_response(self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
with self._control.fail(): |
|
|
|
|
with self.assertRaises(grpc.RpcError) as exception_context: |
|
|
|
|
response_iterator = multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', 'FailedStreamRequestStreamResponse'),)) |
|
|
|
|
tuple(response_iterator) |
|
|
|
|
|
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, |
|
|
|
|
exception_context.exception.code()) |
|
|
|
|
self.assertIs(grpc.StatusCode.UNKNOWN, response_iterator.code()) |
|
|
|
|
|
|
|
|
|
def _ignored_unary_stream_request_future_unary_response( |
|
|
|
|
self, multi_callable): |
|
|
|
|
request = b'\x37\x17' |
|
|
|
|
|
|
|
|
|
multi_callable( |
|
|
|
|
request, metadata=(('test', 'IgnoredUnaryRequestStreamResponse'),)) |
|
|
|
|
|
|
|
|
|
def _ignored_stream_request_stream_response(self, multi_callable): |
|
|
|
|
requests = tuple( |
|
|
|
|
b'\x67\x88' for _ in range(test_constants.STREAM_LENGTH)) |
|
|
|
|
request_iterator = iter(requests) |
|
|
|
|
|
|
|
|
|
multi_callable = _stream_stream_multi_callable(self._channel) |
|
|
|
|
multi_callable( |
|
|
|
|
request_iterator, |
|
|
|
|
metadata=(('test', 'IgnoredStreamRequestStreamResponse'),)) |
|
|
|
|
|
Lidi Zheng
6 years ago
committed by
GitHub