| # 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. |
| |
| """Tests for the old '_low'.""" |
| |
| import threading |
| import time |
| import unittest |
| |
| import six |
| from six.moves import queue |
| |
| from grpc._adapter import _intermediary_low as _low |
| |
| _STREAM_LENGTH = 300 |
| _TIMEOUT = 5 |
| _AFTER_DELAY = 2 |
| _FUTURE = time.time() + 60 * 60 * 24 |
| _BYTE_SEQUENCE = b'\abcdefghijklmnopqrstuvwxyz0123456789' * 200 |
| _BYTE_SEQUENCE_SEQUENCE = tuple( |
| bytes(bytearray((row + column) % 256 for column in range(row))) |
| for row in range(_STREAM_LENGTH)) |
| |
| |
| class LonelyClientTest(unittest.TestCase): |
| |
| def testLonelyClient(self): |
| host = 'nosuchhostexists' |
| port = 54321 |
| method = 'test method' |
| deadline = time.time() + _TIMEOUT |
| after_deadline = deadline + _AFTER_DELAY |
| metadata_tag = object() |
| finish_tag = object() |
| |
| completion_queue = _low.CompletionQueue() |
| channel = _low.Channel('%s:%d' % (host, port), None) |
| client_call = _low.Call(channel, completion_queue, method, host, deadline) |
| |
| client_call.invoke(completion_queue, metadata_tag, finish_tag) |
| first_event = completion_queue.get(after_deadline) |
| self.assertIsNotNone(first_event) |
| second_event = completion_queue.get(after_deadline) |
| self.assertIsNotNone(second_event) |
| kinds = [event.kind for event in (first_event, second_event)] |
| six.assertCountEqual(self, |
| (_low.Event.Kind.METADATA_ACCEPTED, _low.Event.Kind.FINISH), |
| kinds) |
| |
| self.assertIsNone(completion_queue.get(after_deadline)) |
| |
| completion_queue.stop() |
| stop_event = completion_queue.get(_FUTURE) |
| self.assertEqual(_low.Event.Kind.STOP, stop_event.kind) |
| |
| del client_call |
| del channel |
| del completion_queue |
| |
| |
| def _drive_completion_queue(completion_queue, event_queue): |
| while True: |
| event = completion_queue.get(_FUTURE) |
| if event.kind is _low.Event.Kind.STOP: |
| break |
| event_queue.put(event) |
| |
| |
| class EchoTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.host = 'localhost' |
| |
| self.server_completion_queue = _low.CompletionQueue() |
| self.server = _low.Server(self.server_completion_queue) |
| port = self.server.add_http2_addr('[::]:0') |
| self.server.start() |
| self.server_events = queue.Queue() |
| self.server_completion_queue_thread = threading.Thread( |
| target=_drive_completion_queue, |
| args=(self.server_completion_queue, self.server_events)) |
| self.server_completion_queue_thread.start() |
| |
| self.client_completion_queue = _low.CompletionQueue() |
| self.channel = _low.Channel('%s:%d' % (self.host, port), None) |
| self.client_events = queue.Queue() |
| self.client_completion_queue_thread = threading.Thread( |
| target=_drive_completion_queue, |
| args=(self.client_completion_queue, self.client_events)) |
| self.client_completion_queue_thread.start() |
| |
| def tearDown(self): |
| self.server.stop() |
| self.server.cancel_all_calls() |
| self.server_completion_queue.stop() |
| self.client_completion_queue.stop() |
| self.server_completion_queue_thread.join() |
| self.client_completion_queue_thread.join() |
| del self.server |
| |
| def _perform_echo_test(self, test_data): |
| method = 'test method' |
| details = 'test details' |
| server_leading_metadata_key = 'my_server_leading_key' |
| server_leading_metadata_value = 'my_server_leading_value' |
| server_trailing_metadata_key = 'my_server_trailing_key' |
| server_trailing_metadata_value = 'my_server_trailing_value' |
| client_metadata_key = 'my_client_key' |
| client_metadata_value = 'my_client_value' |
| server_leading_binary_metadata_key = 'my_server_leading_key-bin' |
| server_leading_binary_metadata_value = b'\0'*2047 |
| server_trailing_binary_metadata_key = 'my_server_trailing_key-bin' |
| server_trailing_binary_metadata_value = b'\0'*2047 |
| client_binary_metadata_key = 'my_client_key-bin' |
| client_binary_metadata_value = b'\0'*2047 |
| deadline = _FUTURE |
| metadata_tag = object() |
| finish_tag = object() |
| write_tag = object() |
| complete_tag = object() |
| service_tag = object() |
| read_tag = object() |
| status_tag = object() |
| |
| server_data = [] |
| client_data = [] |
| |
| client_call = _low.Call(self.channel, self.client_completion_queue, |
| method, self.host, deadline) |
| client_call.add_metadata(client_metadata_key, client_metadata_value) |
| client_call.add_metadata(client_binary_metadata_key, |
| client_binary_metadata_value) |
| |
| client_call.invoke(self.client_completion_queue, metadata_tag, finish_tag) |
| |
| self.server.service(service_tag) |
| service_accepted = self.server_events.get() |
| self.assertIsNotNone(service_accepted) |
| self.assertIs(service_accepted.kind, _low.Event.Kind.SERVICE_ACCEPTED) |
| self.assertIs(service_accepted.tag, service_tag) |
| self.assertEqual(method, service_accepted.service_acceptance.method) |
| self.assertEqual(self.host, service_accepted.service_acceptance.host) |
| self.assertIsNotNone(service_accepted.service_acceptance.call) |
| metadata = dict(service_accepted.metadata) |
| self.assertIn(client_metadata_key, metadata) |
| self.assertEqual(client_metadata_value, metadata[client_metadata_key]) |
| self.assertIn(client_binary_metadata_key, metadata) |
| self.assertEqual(client_binary_metadata_value, |
| metadata[client_binary_metadata_key]) |
| server_call = service_accepted.service_acceptance.call |
| server_call.accept(self.server_completion_queue, finish_tag) |
| server_call.add_metadata(server_leading_metadata_key, |
| server_leading_metadata_value) |
| server_call.add_metadata(server_leading_binary_metadata_key, |
| server_leading_binary_metadata_value) |
| server_call.premetadata() |
| |
| metadata_accepted = self.client_events.get() |
| self.assertIsNotNone(metadata_accepted) |
| self.assertEqual(_low.Event.Kind.METADATA_ACCEPTED, metadata_accepted.kind) |
| self.assertEqual(metadata_tag, metadata_accepted.tag) |
| metadata = dict(metadata_accepted.metadata) |
| self.assertIn(server_leading_metadata_key, metadata) |
| self.assertEqual(server_leading_metadata_value, |
| metadata[server_leading_metadata_key]) |
| self.assertIn(server_leading_binary_metadata_key, metadata) |
| self.assertEqual(server_leading_binary_metadata_value, |
| metadata[server_leading_binary_metadata_key]) |
| |
| for datum in test_data: |
| client_call.write(datum, write_tag, _low.WriteFlags.WRITE_NO_COMPRESS) |
| write_accepted = self.client_events.get() |
| self.assertIsNotNone(write_accepted) |
| self.assertIs(write_accepted.kind, _low.Event.Kind.WRITE_ACCEPTED) |
| self.assertIs(write_accepted.tag, write_tag) |
| self.assertIs(write_accepted.write_accepted, True) |
| |
| server_call.read(read_tag) |
| read_accepted = self.server_events.get() |
| self.assertIsNotNone(read_accepted) |
| self.assertEqual(_low.Event.Kind.READ_ACCEPTED, read_accepted.kind) |
| self.assertEqual(read_tag, read_accepted.tag) |
| self.assertIsNotNone(read_accepted.bytes) |
| server_data.append(read_accepted.bytes) |
| |
| server_call.write(read_accepted.bytes, write_tag, 0) |
| write_accepted = self.server_events.get() |
| self.assertIsNotNone(write_accepted) |
| self.assertEqual(_low.Event.Kind.WRITE_ACCEPTED, write_accepted.kind) |
| self.assertEqual(write_tag, write_accepted.tag) |
| self.assertTrue(write_accepted.write_accepted) |
| |
| client_call.read(read_tag) |
| read_accepted = self.client_events.get() |
| self.assertIsNotNone(read_accepted) |
| self.assertEqual(_low.Event.Kind.READ_ACCEPTED, read_accepted.kind) |
| self.assertEqual(read_tag, read_accepted.tag) |
| self.assertIsNotNone(read_accepted.bytes) |
| client_data.append(read_accepted.bytes) |
| |
| client_call.complete(complete_tag) |
| complete_accepted = self.client_events.get() |
| self.assertIsNotNone(complete_accepted) |
| self.assertIs(complete_accepted.kind, _low.Event.Kind.COMPLETE_ACCEPTED) |
| self.assertIs(complete_accepted.tag, complete_tag) |
| self.assertIs(complete_accepted.complete_accepted, True) |
| |
| server_call.read(read_tag) |
| read_accepted = self.server_events.get() |
| self.assertIsNotNone(read_accepted) |
| self.assertEqual(_low.Event.Kind.READ_ACCEPTED, read_accepted.kind) |
| self.assertEqual(read_tag, read_accepted.tag) |
| self.assertIsNone(read_accepted.bytes) |
| |
| server_call.add_metadata(server_trailing_metadata_key, |
| server_trailing_metadata_value) |
| server_call.add_metadata(server_trailing_binary_metadata_key, |
| server_trailing_binary_metadata_value) |
| |
| server_call.status(_low.Status(_low.Code.OK, details), status_tag) |
| server_terminal_event_one = self.server_events.get() |
| server_terminal_event_two = self.server_events.get() |
| if server_terminal_event_one.kind == _low.Event.Kind.COMPLETE_ACCEPTED: |
| status_accepted = server_terminal_event_one |
| rpc_accepted = server_terminal_event_two |
| else: |
| status_accepted = server_terminal_event_two |
| rpc_accepted = server_terminal_event_one |
| self.assertIsNotNone(status_accepted) |
| self.assertIsNotNone(rpc_accepted) |
| self.assertEqual(_low.Event.Kind.COMPLETE_ACCEPTED, status_accepted.kind) |
| self.assertEqual(status_tag, status_accepted.tag) |
| self.assertTrue(status_accepted.complete_accepted) |
| self.assertEqual(_low.Event.Kind.FINISH, rpc_accepted.kind) |
| self.assertEqual(finish_tag, rpc_accepted.tag) |
| self.assertEqual(_low.Status(_low.Code.OK, ''), rpc_accepted.status) |
| |
| client_call.read(read_tag) |
| client_terminal_event_one = self.client_events.get() |
| client_terminal_event_two = self.client_events.get() |
| if client_terminal_event_one.kind == _low.Event.Kind.READ_ACCEPTED: |
| read_accepted = client_terminal_event_one |
| finish_accepted = client_terminal_event_two |
| else: |
| read_accepted = client_terminal_event_two |
| finish_accepted = client_terminal_event_one |
| self.assertIsNotNone(read_accepted) |
| self.assertIsNotNone(finish_accepted) |
| self.assertEqual(_low.Event.Kind.READ_ACCEPTED, read_accepted.kind) |
| self.assertEqual(read_tag, read_accepted.tag) |
| self.assertIsNone(read_accepted.bytes) |
| self.assertEqual(_low.Event.Kind.FINISH, finish_accepted.kind) |
| self.assertEqual(finish_tag, finish_accepted.tag) |
| self.assertEqual(_low.Status(_low.Code.OK, details), finish_accepted.status) |
| metadata = dict(finish_accepted.metadata) |
| self.assertIn(server_trailing_metadata_key, metadata) |
| self.assertEqual(server_trailing_metadata_value, |
| metadata[server_trailing_metadata_key]) |
| self.assertIn(server_trailing_binary_metadata_key, metadata) |
| self.assertEqual(server_trailing_binary_metadata_value, |
| metadata[server_trailing_binary_metadata_key]) |
| self.assertSetEqual(set(key for key, _ in finish_accepted.metadata), |
| set((server_trailing_metadata_key, |
| server_trailing_binary_metadata_key,))) |
| |
| self.assertSequenceEqual(test_data, server_data) |
| self.assertSequenceEqual(test_data, client_data) |
| |
| def testNoEcho(self): |
| self._perform_echo_test(()) |
| |
| def testOneByteEcho(self): |
| self._perform_echo_test([b'\x07']) |
| |
| def testOneManyByteEcho(self): |
| self._perform_echo_test([_BYTE_SEQUENCE]) |
| |
| def testManyOneByteEchoes(self): |
| self._perform_echo_test(_BYTE_SEQUENCE) |
| |
| def testManyManyByteEchoes(self): |
| self._perform_echo_test(_BYTE_SEQUENCE_SEQUENCE) |
| |
| |
| class CancellationTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.host = 'localhost' |
| |
| self.server_completion_queue = _low.CompletionQueue() |
| self.server = _low.Server(self.server_completion_queue) |
| port = self.server.add_http2_addr('[::]:0') |
| self.server.start() |
| self.server_events = queue.Queue() |
| self.server_completion_queue_thread = threading.Thread( |
| target=_drive_completion_queue, |
| args=(self.server_completion_queue, self.server_events)) |
| self.server_completion_queue_thread.start() |
| |
| self.client_completion_queue = _low.CompletionQueue() |
| self.channel = _low.Channel('%s:%d' % (self.host, port), None) |
| self.client_events = queue.Queue() |
| self.client_completion_queue_thread = threading.Thread( |
| target=_drive_completion_queue, |
| args=(self.client_completion_queue, self.client_events)) |
| self.client_completion_queue_thread.start() |
| |
| def tearDown(self): |
| self.server.stop() |
| self.server.cancel_all_calls() |
| self.server_completion_queue.stop() |
| self.client_completion_queue.stop() |
| self.server_completion_queue_thread.join() |
| self.client_completion_queue_thread.join() |
| del self.server |
| |
| def testCancellation(self): |
| method = 'test method' |
| deadline = _FUTURE |
| metadata_tag = object() |
| finish_tag = object() |
| write_tag = object() |
| service_tag = object() |
| read_tag = object() |
| test_data = _BYTE_SEQUENCE_SEQUENCE |
| |
| server_data = [] |
| client_data = [] |
| |
| client_call = _low.Call(self.channel, self.client_completion_queue, |
| method, self.host, deadline) |
| |
| client_call.invoke(self.client_completion_queue, metadata_tag, finish_tag) |
| |
| self.server.service(service_tag) |
| service_accepted = self.server_events.get() |
| server_call = service_accepted.service_acceptance.call |
| |
| server_call.accept(self.server_completion_queue, finish_tag) |
| server_call.premetadata() |
| |
| metadata_accepted = self.client_events.get() |
| self.assertIsNotNone(metadata_accepted) |
| |
| for datum in test_data: |
| client_call.write(datum, write_tag, 0) |
| write_accepted = self.client_events.get() |
| |
| server_call.read(read_tag) |
| read_accepted = self.server_events.get() |
| server_data.append(read_accepted.bytes) |
| |
| server_call.write(read_accepted.bytes, write_tag, 0) |
| write_accepted = self.server_events.get() |
| self.assertIsNotNone(write_accepted) |
| |
| client_call.read(read_tag) |
| read_accepted = self.client_events.get() |
| client_data.append(read_accepted.bytes) |
| |
| client_call.cancel() |
| # cancel() is idempotent. |
| client_call.cancel() |
| client_call.cancel() |
| client_call.cancel() |
| |
| server_call.read(read_tag) |
| |
| server_terminal_event_one = self.server_events.get() |
| server_terminal_event_two = self.server_events.get() |
| if server_terminal_event_one.kind == _low.Event.Kind.READ_ACCEPTED: |
| read_accepted = server_terminal_event_one |
| rpc_accepted = server_terminal_event_two |
| else: |
| read_accepted = server_terminal_event_two |
| rpc_accepted = server_terminal_event_one |
| self.assertIsNotNone(read_accepted) |
| self.assertIsNotNone(rpc_accepted) |
| self.assertEqual(_low.Event.Kind.READ_ACCEPTED, read_accepted.kind) |
| self.assertIsNone(read_accepted.bytes) |
| self.assertEqual(_low.Event.Kind.FINISH, rpc_accepted.kind) |
| self.assertEqual(_low.Status(_low.Code.CANCELLED, ''), rpc_accepted.status) |
| |
| finish_event = self.client_events.get() |
| self.assertEqual(_low.Event.Kind.FINISH, finish_event.kind) |
| self.assertEqual(_low.Status(_low.Code.CANCELLED, 'Cancelled'), |
| finish_event.status) |
| |
| self.assertSequenceEqual(test_data, server_data) |
| self.assertSequenceEqual(test_data, client_data) |
| |
| |
| class ExpirationTest(unittest.TestCase): |
| |
| @unittest.skip('TODO(nathaniel): Expiration test!') |
| def testExpiration(self): |
| pass |
| |
| |
| if __name__ == '__main__': |
| unittest.main(verbosity=2) |
| |