# Copyright 2016 gRPC authors. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Test of RPCs made against gRPC Python's application-layer API.""" import itertools import threading import unittest import logging from concurrent import futures 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 _SERIALIZE_REQUEST = lambda bytestring: bytestring * 2 _DESERIALIZE_REQUEST = lambda bytestring: bytestring[len(bytestring) // 2:] _SERIALIZE_RESPONSE = lambda bytestring: bytestring * 3 _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): def __init__(self): self._condition = threading.Condition() self._value = None self._called = False def __call__(self, value): with self._condition: self._value = value self._called = True self._condition.notify_all() def value(self): with self._condition: while not self._called: self._condition.wait() return self._value class _Handler(object): 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() if servicer_context is not None: servicer_context.set_trailing_metadata((( 'testkey', 'testvalue', ),)) # TODO(https://github.com/grpc/grpc/issues/8483): test the values # returned by these methods rather than only "smoke" testing that # the return after having been called. servicer_context.is_active() servicer_context.time_remaining() return request def handle_unary_stream(self, request, servicer_context): for _ in range(test_constants.STREAM_LENGTH): self._control.control() yield request self._control.control() if servicer_context is not None: servicer_context.set_trailing_metadata((( 'testkey', '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() self._control.control() response_elements = [] for request in request_iterator: self._control.control() response_elements.append(request) self._control.control() if servicer_context is not None: servicer_context.set_trailing_metadata((( 'testkey', 'testvalue', ),)) return b''.join(response_elements) def handle_stream_stream(self, request_iterator, servicer_context): self._control.control() if servicer_context is not None: servicer_context.set_trailing_metadata((( 'testkey', 'testvalue', ),)) for request in request_iterator: self._control.control() 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): def __init__(self, request_streaming, response_streaming, request_deserializer, response_serializer, unary_unary, unary_stream, stream_unary, stream_stream): self.request_streaming = request_streaming self.response_streaming = response_streaming self.request_deserializer = request_deserializer self.response_serializer = response_serializer self.unary_unary = unary_unary self.unary_stream = unary_stream self.stream_unary = stream_unary self.stream_stream = stream_stream class _GenericHandler(grpc.GenericRpcHandler): def __init__(self, handler): self._handler = handler def service(self, handler_call_details): if handler_call_details.method == _UNARY_UNARY: return _MethodHandler(False, False, None, None, self._handler.handle_unary_unary, None, None, None) elif handler_call_details.method == _UNARY_STREAM: 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, self._handler.handle_stream_unary, None) 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 def _unary_unary_multi_callable(channel): return channel.unary_unary(_UNARY_UNARY) def _unary_stream_multi_callable(channel): return channel.unary_stream(_UNARY_STREAM, request_serializer=_SERIALIZE_REQUEST, 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, request_serializer=_SERIALIZE_REQUEST, response_deserializer=_DESERIALIZE_RESPONSE) 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._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') self._server.add_generic_rpc_handlers((_GenericHandler(self._handler),)) self._server.start() self._channel = grpc.insecure_channel('localhost:%d' % port) def tearDown(self): 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' with self.assertRaises(grpc.RpcError) as exception_context: self._channel.unary_unary('NoSuchMethod')(request) self.assertEqual(grpc.StatusCode.UNIMPLEMENTED, exception_context.exception.code()) def testSuccessfulUnaryRequestBlockingUnaryResponse(self): request = b'\x07\x08' expected_response = self._handler.handle_unary_unary(request, None) multi_callable = _unary_unary_multi_callable(self._channel) response = multi_callable( request, metadata=(('test', 'SuccessfulUnaryRequestBlockingUnaryResponse'),)) self.assertEqual(expected_response, response) def testSuccessfulUnaryRequestBlockingUnaryResponseWithCall(self): request = b'\x07\x08' expected_response = self._handler.handle_unary_unary(request, None) multi_callable = _unary_unary_multi_callable(self._channel) response, call = multi_callable.with_call( request, metadata=(('test', 'SuccessfulUnaryRequestBlockingUnaryResponseWithCall'),)) self.assertEqual(expected_response, response) self.assertIs(grpc.StatusCode.OK, call.code()) self.assertEqual('', call.debug_error_string()) def testSuccessfulUnaryRequestFutureUnaryResponse(self): request = b'\x07\x08' expected_response = self._handler.handle_unary_unary(request, None) multi_callable = _unary_unary_multi_callable(self._channel) response_future = multi_callable.future( request, metadata=(('test', 'SuccessfulUnaryRequestFutureUnaryResponse'),)) response = response_future.result() self.assertIsInstance(response_future, grpc.Future) self.assertIsInstance(response_future, grpc.Call) self.assertEqual(expected_response, response) self.assertIsNone(response_future.exception()) self.assertIsNone(response_future.traceback()) def testSuccessfulUnaryRequestStreamResponse(self): request = b'\x37\x58' expected_responses = tuple( self._handler.handle_unary_stream(request, None)) multi_callable = _unary_stream_multi_callable(self._channel) response_iterator = multi_callable( request, metadata=(('test', 'SuccessfulUnaryRequestStreamResponse'),)) responses = tuple(response_iterator) self.assertSequenceEqual(expected_responses, responses) def testSuccessfulStreamRequestBlockingUnaryResponse(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) expected_response = self._handler.handle_stream_unary( iter(requests), None) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) response = multi_callable( request_iterator, metadata=(('test', 'SuccessfulStreamRequestBlockingUnaryResponse'),)) self.assertEqual(expected_response, response) def testSuccessfulStreamRequestBlockingUnaryResponseWithCall(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) expected_response = self._handler.handle_stream_unary( iter(requests), None) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) response, call = multi_callable.with_call( request_iterator, metadata=( ('test', 'SuccessfulStreamRequestBlockingUnaryResponseWithCall'),)) self.assertEqual(expected_response, response) self.assertIs(grpc.StatusCode.OK, call.code()) def testSuccessfulStreamRequestFutureUnaryResponse(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) expected_response = self._handler.handle_stream_unary( iter(requests), None) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) response_future = multi_callable.future( request_iterator, metadata=(('test', 'SuccessfulStreamRequestFutureUnaryResponse'),)) response = response_future.result() self.assertEqual(expected_response, response) self.assertIsNone(response_future.exception()) self.assertIsNone(response_future.traceback()) 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) multi_callable = _stream_stream_multi_callable(self._channel) response_iterator = multi_callable( request_iterator, metadata=(('test', 'SuccessfulStreamRequestStreamResponse'),)) responses = tuple(response_iterator) self.assertSequenceEqual(expected_responses, responses) def testSequentialInvocations(self): first_request = b'\x07\x08' second_request = b'\x0809' expected_first_response = self._handler.handle_unary_unary( first_request, None) expected_second_response = self._handler.handle_unary_unary( second_request, None) multi_callable = _unary_unary_multi_callable(self._channel) first_response = multi_callable(first_request, metadata=(('test', 'SequentialInvocations'),)) second_response = multi_callable(second_request, metadata=(('test', 'SequentialInvocations'),)) self.assertEqual(expected_first_response, first_response) self.assertEqual(expected_second_response, second_response) def testConcurrentBlockingInvocations(self): pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY) requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) expected_response = self._handler.handle_stream_unary( iter(requests), None) expected_responses = [expected_response ] * test_constants.THREAD_CONCURRENCY response_futures = [None] * test_constants.THREAD_CONCURRENCY multi_callable = _stream_unary_multi_callable(self._channel) for index in range(test_constants.THREAD_CONCURRENCY): request_iterator = iter(requests) response_future = pool.submit( multi_callable, request_iterator, metadata=(('test', 'ConcurrentBlockingInvocations'),)) response_futures[index] = response_future responses = tuple( response_future.result() for response_future in response_futures) pool.shutdown(wait=True) self.assertSequenceEqual(expected_responses, responses) def testConcurrentFutureInvocations(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) expected_response = self._handler.handle_stream_unary( iter(requests), None) expected_responses = [expected_response ] * test_constants.THREAD_CONCURRENCY response_futures = [None] * test_constants.THREAD_CONCURRENCY multi_callable = _stream_unary_multi_callable(self._channel) for index in range(test_constants.THREAD_CONCURRENCY): request_iterator = iter(requests) response_future = multi_callable.future( request_iterator, metadata=(('test', 'ConcurrentFutureInvocations'),)) response_futures[index] = response_future responses = tuple( response_future.result() for response_future in response_futures) self.assertSequenceEqual(expected_responses, responses) def testWaitingForSomeButNotAllConcurrentFutureInvocations(self): pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY) request = b'\x67\x68' expected_response = self._handler.handle_unary_unary(request, None) response_futures = [None] * test_constants.THREAD_CONCURRENCY lock = threading.Lock() test_is_running_cell = [True] def wrap_future(future): def wrap(): try: return future.result() except grpc.RpcError: with lock: if test_is_running_cell[0]: raise return None return wrap multi_callable = _unary_unary_multi_callable(self._channel) for index in range(test_constants.THREAD_CONCURRENCY): inner_response_future = multi_callable.future( request, metadata=( ('test', 'WaitingForSomeButNotAllConcurrentFutureInvocations'),)) outer_response_future = pool.submit( wrap_future(inner_response_future)) response_futures[index] = outer_response_future some_completed_response_futures_iterator = itertools.islice( futures.as_completed(response_futures), test_constants.THREAD_CONCURRENCY // 2) for response_future in some_completed_response_futures_iterator: self.assertEqual(expected_response, response_future.result()) with lock: test_is_running_cell[0] = False def testConsumingOneStreamResponseUnaryRequest(self): self._consume_one_stream_response_unary_request( _unary_stream_multi_callable(self._channel)) def testConsumingOneStreamResponseUnaryRequestNonBlocking(self): self._consume_one_stream_response_unary_request( _unary_stream_non_blocking_multi_callable(self._channel)) def testConsumingSomeButNotAllStreamResponsesUnaryRequest(self): self._consume_some_but_not_all_stream_responses_unary_request( _unary_stream_multi_callable(self._channel)) def testConsumingSomeButNotAllStreamResponsesUnaryRequestNonBlocking(self): self._consume_some_but_not_all_stream_responses_unary_request( _unary_stream_non_blocking_multi_callable(self._channel)) def testConsumingSomeButNotAllStreamResponsesStreamRequest(self): self._consume_some_but_not_all_stream_responses_stream_request( _stream_stream_multi_callable(self._channel)) def testConsumingSomeButNotAllStreamResponsesStreamRequestNonBlocking(self): self._consume_some_but_not_all_stream_responses_stream_request( _stream_stream_non_blocking_multi_callable(self._channel)) def testConsumingTooManyStreamResponsesStreamRequest(self): self._consume_too_many_stream_responses_stream_request( _stream_stream_multi_callable(self._channel)) 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' multi_callable = _unary_unary_multi_callable(self._channel) with self._control.pause(): response_future = multi_callable.future( request, metadata=(('test', 'CancelledUnaryRequestUnaryResponse'),)) response_future.cancel() self.assertIs(grpc.StatusCode.CANCELLED, response_future.code()) self.assertTrue(response_future.cancelled()) with self.assertRaises(grpc.FutureCancelledError): response_future.result() with self.assertRaises(grpc.FutureCancelledError): response_future.exception() with self.assertRaises(grpc.FutureCancelledError): response_future.traceback() def testCancelledUnaryRequestStreamResponse(self): self._cancelled_unary_request_stream_response( _unary_stream_multi_callable(self._channel)) def testCancelledUnaryRequestStreamResponseNonBlocking(self): self._cancelled_unary_request_stream_response( _unary_stream_non_blocking_multi_callable(self._channel)) def testCancelledStreamRequestUnaryResponse(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) with self._control.pause(): response_future = multi_callable.future( request_iterator, metadata=(('test', 'CancelledStreamRequestUnaryResponse'),)) self._control.block_until_paused() response_future.cancel() self.assertIs(grpc.StatusCode.CANCELLED, response_future.code()) self.assertTrue(response_future.cancelled()) with self.assertRaises(grpc.FutureCancelledError): response_future.result() with self.assertRaises(grpc.FutureCancelledError): response_future.exception() with self.assertRaises(grpc.FutureCancelledError): response_future.traceback() self.assertIsNotNone(response_future.initial_metadata()) self.assertIsNotNone(response_future.details()) self.assertIsNotNone(response_future.trailing_metadata()) def testCancelledStreamRequestStreamResponse(self): self._cancelled_stream_request_stream_response( _stream_stream_multi_callable(self._channel)) def testCancelledStreamRequestStreamResponseNonBlocking(self): self._cancelled_stream_request_stream_response( _stream_stream_non_blocking_multi_callable(self._channel)) def testExpiredUnaryRequestBlockingUnaryResponse(self): request = b'\x07\x17' multi_callable = _unary_unary_multi_callable(self._channel) with self._control.pause(): with self.assertRaises(grpc.RpcError) as exception_context: multi_callable.with_call( request, timeout=test_constants.SHORT_TIMEOUT, metadata=(('test', 'ExpiredUnaryRequestBlockingUnaryResponse'),)) self.assertIsInstance(exception_context.exception, grpc.Call) self.assertIsNotNone(exception_context.exception.initial_metadata()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code()) self.assertIsNotNone(exception_context.exception.details()) self.assertIsNotNone(exception_context.exception.trailing_metadata()) def testExpiredUnaryRequestFutureUnaryResponse(self): request = b'\x07\x17' callback = _Callback() multi_callable = _unary_unary_multi_callable(self._channel) with self._control.pause(): response_future = multi_callable.future( request, timeout=test_constants.SHORT_TIMEOUT, metadata=(('test', 'ExpiredUnaryRequestFutureUnaryResponse'),)) response_future.add_done_callback(callback) value_passed_to_callback = callback.value() self.assertIs(response_future, value_passed_to_callback) self.assertIsNotNone(response_future.initial_metadata()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, response_future.code()) self.assertIsNotNone(response_future.details()) self.assertIsNotNone(response_future.trailing_metadata()) with self.assertRaises(grpc.RpcError) as exception_context: response_future.result() self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code()) self.assertIsInstance(response_future.exception(), grpc.RpcError) self.assertIsNotNone(response_future.traceback()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, response_future.exception().code()) def testExpiredUnaryRequestStreamResponse(self): self._expired_unary_request_stream_response( _unary_stream_multi_callable(self._channel)) def testExpiredUnaryRequestStreamResponseNonBlocking(self): self._expired_unary_request_stream_response( _unary_stream_non_blocking_multi_callable(self._channel)) def testExpiredStreamRequestBlockingUnaryResponse(self): requests = tuple( b'\x07\x08' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) with self._control.pause(): with self.assertRaises(grpc.RpcError) as exception_context: multi_callable( request_iterator, timeout=test_constants.SHORT_TIMEOUT, metadata=(('test', 'ExpiredStreamRequestBlockingUnaryResponse'),)) self.assertIsInstance(exception_context.exception, grpc.RpcError) self.assertIsInstance(exception_context.exception, grpc.Call) self.assertIsNotNone(exception_context.exception.initial_metadata()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code()) self.assertIsNotNone(exception_context.exception.details()) self.assertIsNotNone(exception_context.exception.trailing_metadata()) def testExpiredStreamRequestFutureUnaryResponse(self): requests = tuple( b'\x07\x18' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) callback = _Callback() multi_callable = _stream_unary_multi_callable(self._channel) with self._control.pause(): response_future = multi_callable.future( request_iterator, timeout=test_constants.SHORT_TIMEOUT, metadata=(('test', 'ExpiredStreamRequestFutureUnaryResponse'),)) with self.assertRaises(grpc.FutureTimeoutError): response_future.result(timeout=test_constants.SHORT_TIMEOUT / 2.0) response_future.add_done_callback(callback) value_passed_to_callback = callback.value() with self.assertRaises(grpc.RpcError) as exception_context: response_future.result() self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, response_future.code()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, exception_context.exception.code()) self.assertIsInstance(response_future.exception(), grpc.RpcError) self.assertIsNotNone(response_future.traceback()) self.assertIs(response_future, value_passed_to_callback) self.assertIsNotNone(response_future.initial_metadata()) self.assertIs(grpc.StatusCode.DEADLINE_EXCEEDED, response_future.code()) self.assertIsNotNone(response_future.details()) self.assertIsNotNone(response_future.trailing_metadata()) def testExpiredStreamRequestStreamResponse(self): self._expired_stream_request_stream_response( _stream_stream_multi_callable(self._channel)) def testExpiredStreamRequestStreamResponseNonBlocking(self): self._expired_stream_request_stream_response( _stream_stream_non_blocking_multi_callable(self._channel)) def testFailedUnaryRequestBlockingUnaryResponse(self): request = b'\x37\x17' multi_callable = _unary_unary_multi_callable(self._channel) with self._control.fail(): with self.assertRaises(grpc.RpcError) as exception_context: multi_callable.with_call( request, metadata=(('test', 'FailedUnaryRequestBlockingUnaryResponse'),)) self.assertIs(grpc.StatusCode.UNKNOWN, exception_context.exception.code()) # 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) def testFailedUnaryRequestFutureUnaryResponse(self): request = b'\x37\x17' callback = _Callback() multi_callable = _unary_unary_multi_callable(self._channel) with self._control.fail(): response_future = multi_callable.future( request, metadata=(('test', 'FailedUnaryRequestFutureUnaryResponse'),)) response_future.add_done_callback(callback) value_passed_to_callback = callback.value() self.assertIsInstance(response_future, grpc.Future) self.assertIsInstance(response_future, grpc.Call) with self.assertRaises(grpc.RpcError) as exception_context: response_future.result() self.assertIs(grpc.StatusCode.UNKNOWN, exception_context.exception.code()) self.assertIsInstance(response_future.exception(), grpc.RpcError) self.assertIsNotNone(response_future.traceback()) self.assertIs(grpc.StatusCode.UNKNOWN, response_future.exception().code()) self.assertIs(response_future, value_passed_to_callback) def testFailedUnaryRequestStreamResponse(self): self._failed_unary_request_stream_response( _unary_stream_multi_callable(self._channel)) def testFailedUnaryRequestStreamResponseNonBlocking(self): self._failed_unary_request_stream_response( _unary_stream_non_blocking_multi_callable(self._channel)) def testFailedStreamRequestBlockingUnaryResponse(self): requests = tuple( b'\x47\x58' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) with self._control.fail(): with self.assertRaises(grpc.RpcError) as exception_context: multi_callable( request_iterator, metadata=(('test', 'FailedStreamRequestBlockingUnaryResponse'),)) self.assertIs(grpc.StatusCode.UNKNOWN, exception_context.exception.code()) def testFailedStreamRequestFutureUnaryResponse(self): requests = tuple( b'\x07\x18' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) callback = _Callback() multi_callable = _stream_unary_multi_callable(self._channel) with self._control.fail(): response_future = multi_callable.future( request_iterator, metadata=(('test', 'FailedStreamRequestFutureUnaryResponse'),)) response_future.add_done_callback(callback) value_passed_to_callback = callback.value() with self.assertRaises(grpc.RpcError) as exception_context: response_future.result() self.assertIs(grpc.StatusCode.UNKNOWN, response_future.code()) self.assertIs(grpc.StatusCode.UNKNOWN, exception_context.exception.code()) self.assertIsInstance(response_future.exception(), grpc.RpcError) self.assertIsNotNone(response_future.traceback()) self.assertIs(response_future, value_passed_to_callback) def testFailedStreamRequestStreamResponse(self): self._failed_stream_request_stream_response( _stream_stream_multi_callable(self._channel)) def testFailedStreamRequestStreamResponseNonBlocking(self): self._failed_stream_request_stream_response( _stream_stream_non_blocking_multi_callable(self._channel)) def testIgnoredUnaryRequestFutureUnaryResponse(self): request = b'\x37\x17' multi_callable = _unary_unary_multi_callable(self._channel) multi_callable.future( request, metadata=(('test', 'IgnoredUnaryRequestFutureUnaryResponse'),)) def testIgnoredUnaryRequestStreamResponse(self): self._ignored_unary_stream_request_future_unary_response( _unary_stream_multi_callable(self._channel)) def testIgnoredUnaryRequestStreamResponseNonBlocking(self): self._ignored_unary_stream_request_future_unary_response( _unary_stream_non_blocking_multi_callable(self._channel)) def testIgnoredStreamRequestFutureUnaryResponse(self): requests = tuple( b'\x07\x18' for _ in range(test_constants.STREAM_LENGTH)) request_iterator = iter(requests) multi_callable = _stream_unary_multi_callable(self._channel) multi_callable.future( request_iterator, 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(request_iterator, metadata=(('test', 'IgnoredStreamRequestStreamResponse'),)) if __name__ == '__main__': logging.basicConfig() unittest.main(verbosity=2)