# Copyright 2015, Google Inc. # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are # met: # # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above # copyright notice, this list of conditions and the following disclaimer # in the documentation and/or other materials provided with the # distribution. # * Neither the name of Google Inc. nor the names of its # contributors may be used to endorse or promote products derived from # this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import time import unittest from grpc._cython import cygrpc from grpc_test._cython import test_utilities class TypeSmokeTest(unittest.TestCase): def testStringsInUtilitiesUpDown(self): self.assertEqual(0, cygrpc.StatusCode.ok) metadatum = cygrpc.Metadatum('a', 'b') self.assertEqual('a'.encode(), metadatum.key) self.assertEqual('b'.encode(), metadatum.value) metadata = cygrpc.Metadata([metadatum]) self.assertEqual(1, len(metadata)) self.assertEqual(metadatum.key, metadata[0].key) def testMetadataIteration(self): metadata = cygrpc.Metadata([ cygrpc.Metadatum('a', 'b'), cygrpc.Metadatum('c', 'd')]) iterator = iter(metadata) metadatum = next(iterator) self.assertIsInstance(metadatum, cygrpc.Metadatum) self.assertEqual(metadatum.key, 'a'.encode()) self.assertEqual(metadatum.value, 'b'.encode()) metadatum = next(iterator) self.assertIsInstance(metadatum, cygrpc.Metadatum) self.assertEqual(metadatum.key, 'c'.encode()) self.assertEqual(metadatum.value, 'd'.encode()) with self.assertRaises(StopIteration): next(iterator) def testOperationsIteration(self): operations = cygrpc.Operations([ cygrpc.operation_send_message('asdf')]) iterator = iter(operations) operation = next(iterator) self.assertIsInstance(operation, cygrpc.Operation) # `Operation`s are write-only structures; can't directly debug anything out # of them. Just check that we stop iterating. with self.assertRaises(StopIteration): next(iterator) def testTimespec(self): now = time.time() timespec = cygrpc.Timespec(now) self.assertAlmostEqual(now, float(timespec), places=8) def testCompletionQueueUpDown(self): completion_queue = cygrpc.CompletionQueue() del completion_queue def testServerUpDown(self): server = cygrpc.Server(cygrpc.ChannelArgs([])) del server def testChannelUpDown(self): channel = cygrpc.Channel('[::]:0', cygrpc.ChannelArgs([])) del channel @unittest.skip('TODO(atash): undo skip after #2229 is merged') def testServerStartNoExplicitShutdown(self): server = cygrpc.Server() completion_queue = cygrpc.CompletionQueue() server.register_completion_queue(completion_queue) port = server.add_http2_port('[::]:0') self.assertIsInstance(port, int) server.start() del server @unittest.skip('TODO(atash): undo skip after #2229 is merged') def testServerStartShutdown(self): completion_queue = cygrpc.CompletionQueue() server = cygrpc.Server() server.add_http2_port('[::]:0') server.register_completion_queue(completion_queue) server.start() shutdown_tag = object() server.shutdown(completion_queue, shutdown_tag) event = completion_queue.poll() self.assertEqual(cygrpc.CompletionType.operation_complete, event.type) self.assertIs(shutdown_tag, event.tag) del server del completion_queue class InsecureServerInsecureClient(unittest.TestCase): def setUp(self): self.server_completion_queue = cygrpc.CompletionQueue() self.server = cygrpc.Server() self.server.register_completion_queue(self.server_completion_queue) self.port = self.server.add_http2_port('[::]:0') self.server.start() self.client_completion_queue = cygrpc.CompletionQueue() self.client_channel = cygrpc.Channel('localhost:{}'.format(self.port)) def tearDown(self): del self.server del self.client_completion_queue del self.server_completion_queue def testEcho(self): DEADLINE = time.time()+5 DEADLINE_TOLERANCE = 0.25 CLIENT_METADATA_ASCII_KEY = b'key' CLIENT_METADATA_ASCII_VALUE = b'val' CLIENT_METADATA_BIN_KEY = b'key-bin' CLIENT_METADATA_BIN_VALUE = b'\0'*1000 SERVER_INITIAL_METADATA_KEY = b'init_me_me_me' SERVER_INITIAL_METADATA_VALUE = b'whodawha?' SERVER_TRAILING_METADATA_KEY = b'California_is_in_a_drought' SERVER_TRAILING_METADATA_VALUE = b'zomg it is' SERVER_STATUS_CODE = cygrpc.StatusCode.ok SERVER_STATUS_DETAILS = b'our work is never over' REQUEST = b'in death a member of project mayhem has a name' RESPONSE = b'his name is robert paulson' METHOD = b'twinkies' HOST = b'hostess' cygrpc_deadline = cygrpc.Timespec(DEADLINE) server_request_tag = object() request_call_result = self.server.request_call( self.server_completion_queue, self.server_completion_queue, server_request_tag) self.assertEqual(cygrpc.CallError.ok, request_call_result) client_call_tag = object() client_call = self.client_channel.create_call(self.client_completion_queue, METHOD, HOST, cygrpc_deadline) client_initial_metadata = cygrpc.Metadata([ cygrpc.Metadatum(CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE), cygrpc.Metadatum(CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)]) client_start_batch_result = client_call.start_batch(cygrpc.Operations([ cygrpc.operation_send_initial_metadata(client_initial_metadata), cygrpc.operation_send_message(REQUEST), cygrpc.operation_send_close_from_client(), cygrpc.operation_receive_initial_metadata(), cygrpc.operation_receive_message(), cygrpc.operation_receive_status_on_client() ]), client_call_tag) self.assertEqual(cygrpc.CallError.ok, client_start_batch_result) client_event_future = test_utilities.CompletionQueuePollFuture( self.client_completion_queue, cygrpc_deadline) request_event = self.server_completion_queue.poll(cygrpc_deadline) self.assertEqual(cygrpc.CompletionType.operation_complete, request_event.type) self.assertIsInstance(request_event.operation_call, cygrpc.Call) self.assertIs(server_request_tag, request_event.tag) self.assertEqual(0, len(request_event.batch_operations)) self.assertEqual(dict(client_initial_metadata), dict(request_event.request_metadata)) self.assertEqual(METHOD, request_event.request_call_details.method) self.assertEqual(HOST, request_event.request_call_details.host) self.assertLess( abs(DEADLINE - float(request_event.request_call_details.deadline)), DEADLINE_TOLERANCE) server_call_tag = object() server_call = request_event.operation_call server_initial_metadata = cygrpc.Metadata([ cygrpc.Metadatum(SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE)]) server_trailing_metadata = cygrpc.Metadata([ cygrpc.Metadatum(SERVER_TRAILING_METADATA_KEY, SERVER_TRAILING_METADATA_VALUE)]) server_start_batch_result = server_call.start_batch([ cygrpc.operation_send_initial_metadata(server_initial_metadata), cygrpc.operation_receive_message(), cygrpc.operation_send_message(RESPONSE), cygrpc.operation_receive_close_on_server(), cygrpc.operation_send_status_from_server( server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS) ], server_call_tag) self.assertEqual(cygrpc.CallError.ok, server_start_batch_result) client_event = client_event_future.result() server_event = self.server_completion_queue.poll(cygrpc_deadline) self.assertEqual(6, len(client_event.batch_operations)) found_client_op_types = set() for client_result in client_event.batch_operations: # we expect each op type to be unique self.assertNotIn(client_result.type, found_client_op_types) found_client_op_types.add(client_result.type) if client_result.type == cygrpc.OperationType.receive_initial_metadata: self.assertEqual(dict(server_initial_metadata), dict(client_result.received_metadata)) elif client_result.type == cygrpc.OperationType.receive_message: self.assertEqual(RESPONSE, client_result.received_message.bytes()) elif client_result.type == cygrpc.OperationType.receive_status_on_client: self.assertEqual(dict(server_trailing_metadata), dict(client_result.received_metadata)) self.assertEqual(SERVER_STATUS_DETAILS, client_result.received_status_details) self.assertEqual(SERVER_STATUS_CODE, client_result.received_status_code) self.assertEqual(set([ cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.send_message, cygrpc.OperationType.send_close_from_client, cygrpc.OperationType.receive_initial_metadata, cygrpc.OperationType.receive_message, cygrpc.OperationType.receive_status_on_client ]), found_client_op_types) self.assertEqual(5, len(server_event.batch_operations)) found_server_op_types = set() for server_result in server_event.batch_operations: self.assertNotIn(client_result.type, found_server_op_types) found_server_op_types.add(server_result.type) if server_result.type == cygrpc.OperationType.receive_message: self.assertEqual(REQUEST, server_result.received_message.bytes()) elif server_result.type == cygrpc.OperationType.receive_close_on_server: self.assertFalse(server_result.received_cancelled) self.assertEqual(set([ cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.receive_message, cygrpc.OperationType.send_message, cygrpc.OperationType.receive_close_on_server, cygrpc.OperationType.send_status_from_server ]), found_server_op_types) del client_call del server_call if __name__ == '__main__': unittest.main(verbosity=2)