_rpc_part_2_test.py

Go to the documentation of this file.

# 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)