| # 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.""" |
| |
| from concurrent import futures |
| import itertools |
| import logging |
| import threading |
| import unittest |
| |
| import grpc |
| from grpc.framework.foundation import logging_pool |
| |
| from tests.unit._rpc_test_helpers import BaseRPCTest |
| from tests.unit._rpc_test_helpers import Callback |
| from tests.unit._rpc_test_helpers import TIMEOUT_SHORT |
| from tests.unit._rpc_test_helpers import \ |
| stream_stream_non_blocking_multi_callable |
| from tests.unit._rpc_test_helpers import \ |
| unary_stream_non_blocking_multi_callable |
| from tests.unit._rpc_test_helpers import stream_stream_multi_callable |
| from tests.unit._rpc_test_helpers import stream_unary_multi_callable |
| from tests.unit._rpc_test_helpers import unary_stream_multi_callable |
| from tests.unit._rpc_test_helpers import unary_unary_multi_callable |
| from tests.unit.framework.common import test_constants |
| |
| |
| class RPCPart2Test(BaseRPCTest, unittest.TestCase): |
| |
| 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=TIMEOUT_SHORT, |
| 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=TIMEOUT_SHORT, |
| 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)) |
| |
| |
| if __name__ == '__main__': |
| logging.basicConfig() |
| unittest.main(verbosity=2) |
