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