_read_some_but_not_all_responses_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 a corner-case at the level of the Cython API."""
 
import threading
import unittest
 
from grpc._cython import cygrpc
 
from tests.unit._cython import test_utilities
 
_EMPTY_FLAGS = 0
_EMPTY_METADATA = ()
 
 
class _ServerDriver(object):
 
    def __init__(self, completion_queue, shutdown_tag):
        self._condition = threading.Condition()
        self._completion_queue = completion_queue
        self._shutdown_tag = shutdown_tag
        self._events = []
        self._saw_shutdown_tag = False
 
    def start(self):
 
        def in_thread():
            while True:
                event = self._completion_queue.poll()
                with self._condition:
                    self._events.append(event)
                    self._condition.notify()
                    if event.tag is self._shutdown_tag:
                        self._saw_shutdown_tag = True
                        break
 
        thread = threading.Thread(target=in_thread)
        thread.start()
 
    def done(self):
        with self._condition:
            return self._saw_shutdown_tag
 
    def first_event(self):
        with self._condition:
            while not self._events:
                self._condition.wait()
            return self._events[0]
 
    def events(self):
        with self._condition:
            while not self._saw_shutdown_tag:
                self._condition.wait()
            return tuple(self._events)
 
 
class _QueueDriver(object):
 
    def __init__(self, condition, completion_queue, due):
        self._condition = condition
        self._completion_queue = completion_queue
        self._due = due
        self._events = []
        self._returned = False
 
    def start(self):
 
        def in_thread():
            while True:
                event = self._completion_queue.poll()
                with self._condition:
                    self._events.append(event)
                    self._due.remove(event.tag)
                    self._condition.notify_all()
                    if not self._due:
                        self._returned = True
                        return
 
        thread = threading.Thread(target=in_thread)
        thread.start()
 
    def done(self):
        with self._condition:
            return self._returned
 
    def event_with_tag(self, tag):
        with self._condition:
            while True:
                for event in self._events:
                    if event.tag is tag:
                        return event
                self._condition.wait()
 
    def events(self):
        with self._condition:
            while not self._returned:
                self._condition.wait()
            return tuple(self._events)
 
 
class ReadSomeButNotAllResponsesTest(unittest.TestCase):
 
    def testReadSomeButNotAllResponses(self):
        server_completion_queue = cygrpc.CompletionQueue()
        server = cygrpc.Server([(
            b'grpc.so_reuseport',
            0,
        )], False)
        server.register_completion_queue(server_completion_queue)
        port = server.add_http2_port(b'[::]:0')
        server.start()
        channel = cygrpc.Channel('localhost:{}'.format(port).encode(), set(),
                                 None)
 
        server_shutdown_tag = 'server_shutdown_tag'
        server_driver = _ServerDriver(server_completion_queue,
                                      server_shutdown_tag)
        server_driver.start()
 
        client_condition = threading.Condition()
        client_due = set()
 
        server_call_condition = threading.Condition()
        server_send_initial_metadata_tag = 'server_send_initial_metadata_tag'
        server_send_first_message_tag = 'server_send_first_message_tag'
        server_send_second_message_tag = 'server_send_second_message_tag'
        server_complete_rpc_tag = 'server_complete_rpc_tag'
        server_call_due = set((
            server_send_initial_metadata_tag,
            server_send_first_message_tag,
            server_send_second_message_tag,
            server_complete_rpc_tag,
        ))
        server_call_completion_queue = cygrpc.CompletionQueue()
        server_call_driver = _QueueDriver(server_call_condition,
                                          server_call_completion_queue,
                                          server_call_due)
        server_call_driver.start()
 
        server_rpc_tag = 'server_rpc_tag'
        request_call_result = server.request_call(server_call_completion_queue,
                                                  server_completion_queue,
                                                  server_rpc_tag)
 
        client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag'
        client_complete_rpc_tag = 'client_complete_rpc_tag'
        client_call = channel.segregated_call(
            _EMPTY_FLAGS, b'/twinkies', None, None, _EMPTY_METADATA, None, (
                (
                    [
                        cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
                    ],
                    client_receive_initial_metadata_tag,
                ),
                (
                    [
                        cygrpc.SendInitialMetadataOperation(
                            _EMPTY_METADATA, _EMPTY_FLAGS),
                        cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
                        cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
                    ],
                    client_complete_rpc_tag,
                ),
            ))
        client_receive_initial_metadata_event_future = test_utilities.SimpleFuture(
            client_call.next_event)
 
        server_rpc_event = server_driver.first_event()
 
        with server_call_condition:
            server_send_initial_metadata_start_batch_result = (
                server_rpc_event.call.start_server_batch([
                    cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
                                                        _EMPTY_FLAGS),
                ], server_send_initial_metadata_tag))
            server_send_first_message_start_batch_result = (
                server_rpc_event.call.start_server_batch([
                    cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS),
                ], server_send_first_message_tag))
        server_send_initial_metadata_event = server_call_driver.event_with_tag(
            server_send_initial_metadata_tag)
        server_send_first_message_event = server_call_driver.event_with_tag(
            server_send_first_message_tag)
        with server_call_condition:
            server_send_second_message_start_batch_result = (
                server_rpc_event.call.start_server_batch([
                    cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS),
                ], server_send_second_message_tag))
            server_complete_rpc_start_batch_result = (
                server_rpc_event.call.start_server_batch([
                    cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),
                    cygrpc.SendStatusFromServerOperation(
                        (), cygrpc.StatusCode.ok, b'test details',
                        _EMPTY_FLAGS),
                ], server_complete_rpc_tag))
        server_send_second_message_event = server_call_driver.event_with_tag(
            server_send_second_message_tag)
        server_complete_rpc_event = server_call_driver.event_with_tag(
            server_complete_rpc_tag)
        server_call_driver.events()
 
        client_recieve_initial_metadata_event = client_receive_initial_metadata_event_future.result(
        )
 
        client_receive_first_message_tag = 'client_receive_first_message_tag'
        client_call.operate([
            cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
        ], client_receive_first_message_tag)
        client_receive_first_message_event = client_call.next_event()
 
        client_call_cancel_result = client_call.cancel(
            cygrpc.StatusCode.cancelled, 'Cancelled during test!')
        client_complete_rpc_event = client_call.next_event()
 
        channel.close(cygrpc.StatusCode.unknown, 'Channel closed!')
        server.shutdown(server_completion_queue, server_shutdown_tag)
        server.cancel_all_calls()
        server_driver.events()
 
        self.assertEqual(cygrpc.CallError.ok, request_call_result)
        self.assertEqual(cygrpc.CallError.ok,
                         server_send_initial_metadata_start_batch_result)
        self.assertIs(server_rpc_tag, server_rpc_event.tag)
        self.assertEqual(cygrpc.CompletionType.operation_complete,
                         server_rpc_event.completion_type)
        self.assertIsInstance(server_rpc_event.call, cygrpc.Call)
 
 
if __name__ == '__main__':
    unittest.main(verbosity=2)