diff options
Diffstat (limited to 'src/python/grpcio_test')
17 files changed, 700 insertions, 32 deletions
diff --git a/src/python/grpcio_test/grpc_interop/client.py b/src/python/grpcio_test/grpc_interop/client.py index 2dd2103cbe..36afe6c096 100644 --- a/src/python/grpcio_test/grpc_interop/client.py +++ b/src/python/grpcio_test/grpc_interop/client.py @@ -70,7 +70,13 @@ def _oauth_access_token(args): def _stub(args): if args.oauth_scope: - metadata_transformer = lambda x: [('Authorization', 'Bearer %s' % _oauth_access_token(args))] + if args.test_case == 'oauth2_auth_token': + access_token = _oauth_access_token(args) + metadata_transformer = lambda x: [ + ('Authorization', 'Bearer %s' % access_token)] + else: + metadata_transformer = lambda x: [ + ('Authorization', 'Bearer %s' % _oauth_access_token(args))] else: metadata_transformer = lambda x: [] if args.use_tls: diff --git a/src/python/grpcio_test/grpc_interop/methods.py b/src/python/grpcio_test/grpc_interop/methods.py index 7a831f3cbd..642458e892 100644 --- a/src/python/grpcio_test/grpc_interop/methods.py +++ b/src/python/grpcio_test/grpc_interop/methods.py @@ -360,6 +360,19 @@ def _service_account_creds(stub, args): (response.oauth_scope, args.oauth_scope)) +def _oauth2_auth_token(stub, args): + json_key_filename = os.environ[ + oauth2client_client.GOOGLE_APPLICATION_CREDENTIALS] + wanted_email = json.load(open(json_key_filename, 'rb'))['client_email'] + response = _large_unary_common_behavior(stub, True, True) + if wanted_email != response.username: + raise ValueError( + 'expected username %s, got %s' % (wanted_email, response.username)) + if args.oauth_scope.find(response.oauth_scope) == -1: + raise ValueError( + 'expected to find oauth scope "%s" in received "%s"' % + (response.oauth_scope, args.oauth_scope)) + @enum.unique class TestCase(enum.Enum): EMPTY_UNARY = 'empty_unary' @@ -371,6 +384,7 @@ class TestCase(enum.Enum): CANCEL_AFTER_FIRST_RESPONSE = 'cancel_after_first_response' COMPUTE_ENGINE_CREDS = 'compute_engine_creds' SERVICE_ACCOUNT_CREDS = 'service_account_creds' + OAUTH2_AUTH_TOKEN = 'oauth2_auth_token' TIMEOUT_ON_SLEEPING_SERVER = 'timeout_on_sleeping_server' def test_interoperability(self, stub, args): @@ -394,5 +408,7 @@ class TestCase(enum.Enum): _compute_engine_creds(stub, args) elif self is TestCase.SERVICE_ACCOUNT_CREDS: _service_account_creds(stub, args) + elif self is TestCase.OAUTH2_AUTH_TOKEN: + _oauth2_auth_token(stub, args) else: raise NotImplementedError('Test case "%s" not implemented!' % self.name) diff --git a/src/python/grpcio_test/grpc_test/_adapter/_links_test.py b/src/python/grpcio_test/grpc_test/_adapter/_links_test.py index c4686b327a..4077b8e350 100644 --- a/src/python/grpcio_test/grpc_test/_adapter/_links_test.py +++ b/src/python/grpcio_test/grpc_test/_adapter/_links_test.py @@ -46,8 +46,8 @@ _TIMEOUT = 32 # TODO(nathaniel): End-to-end metadata testing. def _transform_metadata(unused_metadata): return ( - ('one unused key', 'one unused value'), - ('another unused key', 'another unused value'), + ('one_unused_key', 'one unused value'), + ('another_unused_key', 'another unused value'), ) diff --git a/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py index 72b1ae5642..7fa90fe35f 100644 --- a/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py +++ b/src/python/grpcio_test/grpc_test/_core_over_links_base_interface_test.py @@ -27,7 +27,7 @@ # (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 the RPC Framework Core's implementation of the Base interface.""" +"""Tests Base interface compliance of the core-over-gRPC-links stack.""" import collections import logging diff --git a/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py b/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py new file mode 100644 index 0000000000..25b99cbbaf --- /dev/null +++ b/src/python/grpcio_test/grpc_test/_crust_over_core_over_links_face_interface_test.py @@ -0,0 +1,160 @@ +# 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 Face compliance of the crust-over-core-over-gRPC-links stack.""" + +import collections +import unittest + +from grpc._adapter import _intermediary_low +from grpc._links import invocation +from grpc._links import service +from grpc.framework.core import implementations as core_implementations +from grpc.framework.crust import implementations as crust_implementations +from grpc.framework.foundation import logging_pool +from grpc.framework.interfaces.links import utilities +from grpc_test import test_common +from grpc_test.framework.common import test_constants +from grpc_test.framework.interfaces.face import test_cases +from grpc_test.framework.interfaces.face import test_interfaces +from grpc_test.framework.interfaces.links import test_utilities + + +class _SerializationBehaviors( + collections.namedtuple( + '_SerializationBehaviors', + ('request_serializers', 'request_deserializers', 'response_serializers', + 'response_deserializers',))): + pass + + +def _serialization_behaviors_from_test_methods(test_methods): + request_serializers = {} + request_deserializers = {} + response_serializers = {} + response_deserializers = {} + for (group, method), test_method in test_methods.iteritems(): + request_serializers[group, method] = test_method.serialize_request + request_deserializers[group, method] = test_method.deserialize_request + response_serializers[group, method] = test_method.serialize_response + response_deserializers[group, method] = test_method.deserialize_response + return _SerializationBehaviors( + request_serializers, request_deserializers, response_serializers, + response_deserializers) + + +class _Implementation(test_interfaces.Implementation): + + def instantiate( + self, methods, method_implementations, multi_method_implementation): + pool = logging_pool.pool(test_constants.POOL_SIZE) + servicer = crust_implementations.servicer( + method_implementations, multi_method_implementation, pool) + serialization_behaviors = _serialization_behaviors_from_test_methods( + methods) + invocation_end_link = core_implementations.invocation_end_link() + service_end_link = core_implementations.service_end_link( + servicer, test_constants.DEFAULT_TIMEOUT, + test_constants.MAXIMUM_TIMEOUT) + service_grpc_link = service.service_link( + serialization_behaviors.request_deserializers, + serialization_behaviors.response_serializers) + port = service_grpc_link.add_port(0, None) + channel = _intermediary_low.Channel('localhost:%d' % port, None) + invocation_grpc_link = invocation.invocation_link( + channel, b'localhost', + serialization_behaviors.request_serializers, + serialization_behaviors.response_deserializers) + + invocation_end_link.join_link(invocation_grpc_link) + invocation_grpc_link.join_link(invocation_end_link) + service_grpc_link.join_link(service_end_link) + service_end_link.join_link(service_grpc_link) + service_end_link.start() + invocation_end_link.start() + invocation_grpc_link.start() + service_grpc_link.start() + + generic_stub = crust_implementations.generic_stub(invocation_end_link, pool) + # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest. + group = next(iter(methods))[0] + # TODO(nathaniel): Add a "cardinalities_by_group" attribute to + # _digest.TestServiceDigest. + cardinalities = { + method: method_object.cardinality() + for (group, method), method_object in methods.iteritems()} + dynamic_stub = crust_implementations.dynamic_stub( + invocation_end_link, group, cardinalities, pool) + + return generic_stub, {group: dynamic_stub}, ( + invocation_end_link, invocation_grpc_link, service_grpc_link, + service_end_link, pool) + + def destantiate(self, memo): + (invocation_end_link, invocation_grpc_link, service_grpc_link, + service_end_link, pool) = memo + invocation_end_link.stop(0).wait() + invocation_grpc_link.stop() + service_grpc_link.stop_gracefully() + service_end_link.stop(0).wait() + invocation_end_link.join_link(utilities.NULL_LINK) + invocation_grpc_link.join_link(utilities.NULL_LINK) + service_grpc_link.join_link(utilities.NULL_LINK) + service_end_link.join_link(utilities.NULL_LINK) + pool.shutdown(wait=True) + + def invocation_metadata(self): + return test_common.INVOCATION_INITIAL_METADATA + + def initial_metadata(self): + return test_common.SERVICE_INITIAL_METADATA + + def terminal_metadata(self): + return test_common.SERVICE_TERMINAL_METADATA + + def code(self): + return _intermediary_low.Code.OK + + def details(self): + return test_common.DETAILS + + def metadata_transmitted(self, original_metadata, transmitted_metadata): + return original_metadata is None or grpc_test_common.metadata_transmitted( + original_metadata, transmitted_metadata) + + +def load_tests(loader, tests, pattern): + return unittest.TestSuite( + tests=tuple( + loader.loadTestsFromTestCase(test_case_class) + for test_case_class in test_cases.test_cases(_Implementation()))) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_test/grpc_test/_cython/adapter_low_test.py b/src/python/grpcio_test/grpc_test/_cython/adapter_low_test.py index 9bab930e56..f1bec238cf 100644 --- a/src/python/grpcio_test/grpc_test/_cython/adapter_low_test.py +++ b/src/python/grpcio_test/grpc_test/_cython/adapter_low_test.py @@ -76,7 +76,7 @@ class InsecureServerInsecureClient(unittest.TestCase): CLIENT_METADATA_BIN_VALUE = b'\0'*1000 SERVER_INITIAL_METADATA_KEY = 'init_me_me_me' SERVER_INITIAL_METADATA_VALUE = 'whodawha?' - SERVER_TRAILING_METADATA_KEY = 'California_is_in_a_drought' + SERVER_TRAILING_METADATA_KEY = 'california_is_in_a_drought' SERVER_TRAILING_METADATA_VALUE = 'zomg it is' SERVER_STATUS_CODE = _types.StatusCode.OK SERVER_STATUS_DETAILS = 'our work is never over' diff --git a/src/python/grpcio_test/grpc_test/_links/_transmission_test.py b/src/python/grpcio_test/grpc_test/_links/_transmission_test.py index 02ddd512c2..db011bca66 100644 --- a/src/python/grpcio_test/grpc_test/_links/_transmission_test.py +++ b/src/python/grpcio_test/grpc_test/_links/_transmission_test.py @@ -66,9 +66,9 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase): def create_invocation_initial_metadata(self): return ( - ('first invocation initial metadata key', 'just a string value'), - ('second invocation initial metadata key', '0123456789'), - ('third invocation initial metadata key-bin', '\x00\x57' * 100), + ('first_invocation_initial_metadata_key', 'just a string value'), + ('second_invocation_initial_metadata_key', '0123456789'), + ('third_invocation_initial_metadata_key-bin', '\x00\x57' * 100), ) def create_invocation_terminal_metadata(self): @@ -76,16 +76,16 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase): def create_service_initial_metadata(self): return ( - ('first service initial metadata key', 'just another string value'), - ('second service initial metadata key', '9876543210'), - ('third service initial metadata key-bin', '\x00\x59\x02' * 100), + ('first_service_initial_metadata_key', 'just another string value'), + ('second_service_initial_metadata_key', '9876543210'), + ('third_service_initial_metadata_key-bin', '\x00\x59\x02' * 100), ) def create_service_terminal_metadata(self): return ( - ('first service terminal metadata key', 'yet another string value'), - ('second service terminal metadata key', 'abcdefghij'), - ('third service terminal metadata key-bin', '\x00\x37' * 100), + ('first_service_terminal_metadata_key', 'yet another string value'), + ('second_service_terminal_metadata_key', 'abcdefghij'), + ('third_service_terminal_metadata_key-bin', '\x00\x37' * 100), ) def create_invocation_completion(self): diff --git a/src/python/grpcio_test/grpc_test/beta/__init__.py b/src/python/grpcio_test/grpc_test/beta/__init__.py new file mode 100644 index 0000000000..7086519106 --- /dev/null +++ b/src/python/grpcio_test/grpc_test/beta/__init__.py @@ -0,0 +1,30 @@ +# 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. + + diff --git a/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py new file mode 100644 index 0000000000..038464889d --- /dev/null +++ b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py @@ -0,0 +1,180 @@ +# 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 of grpc.beta._connectivity_channel.""" + +import threading +import time +import unittest + +from grpc._adapter import _low +from grpc._adapter import _types +from grpc.beta import _connectivity_channel +from grpc_test.framework.common import test_constants + +_MAPPING_FUNCTION = lambda integer: integer * 200 + 17 +_MAPPING = { + state: _MAPPING_FUNCTION(state) for state in _types.ConnectivityState} +_IDLE, _CONNECTING, _READY, _TRANSIENT_FAILURE, _FATAL_FAILURE = map( + _MAPPING_FUNCTION, _types.ConnectivityState) + + +def _drive_completion_queue(completion_queue): + while True: + event = completion_queue.next(time.time() + 24 * 60 * 60) + if event.type == _types.EventType.QUEUE_SHUTDOWN: + break + + +class _Callback(object): + + def __init__(self): + self._condition = threading.Condition() + self._connectivities = [] + + def update(self, connectivity): + with self._condition: + self._connectivities.append(connectivity) + self._condition.notify() + + def connectivities(self): + with self._condition: + return tuple(self._connectivities) + + def block_until_connectivities_satisfy(self, predicate): + with self._condition: + while True: + connectivities = tuple(self._connectivities) + if predicate(connectivities): + return connectivities + else: + self._condition.wait() + + +class ChannelConnectivityTest(unittest.TestCase): + + def test_lonely_channel_connectivity(self): + low_channel = _low.Channel('localhost:12345', ()) + callback = _Callback() + + connectivity_channel = _connectivity_channel.ConnectivityChannel( + low_channel, _MAPPING) + connectivity_channel.subscribe(callback.update, try_to_connect=False) + first_connectivities = callback.block_until_connectivities_satisfy(bool) + connectivity_channel.subscribe(callback.update, try_to_connect=True) + second_connectivities = callback.block_until_connectivities_satisfy( + lambda connectivities: 2 <= len(connectivities)) + # Wait for a connection that will never happen. + time.sleep(test_constants.SHORT_TIMEOUT) + third_connectivities = callback.connectivities() + connectivity_channel.unsubscribe(callback.update) + fourth_connectivities = callback.connectivities() + connectivity_channel.unsubscribe(callback.update) + fifth_connectivities = callback.connectivities() + + self.assertSequenceEqual((_IDLE,), first_connectivities) + self.assertNotIn(_READY, second_connectivities) + self.assertNotIn(_READY, third_connectivities) + self.assertNotIn(_READY, fourth_connectivities) + self.assertNotIn(_READY, fifth_connectivities) + + def test_immediately_connectable_channel_connectivity(self): + server_completion_queue = _low.CompletionQueue() + server = _low.Server(server_completion_queue, []) + port = server.add_http2_port('[::]:0') + server.start() + server_completion_queue_thread = threading.Thread( + target=_drive_completion_queue, args=(server_completion_queue,)) + server_completion_queue_thread.start() + low_channel = _low.Channel('localhost:%d' % port, ()) + first_callback = _Callback() + second_callback = _Callback() + + connectivity_channel = _connectivity_channel.ConnectivityChannel( + low_channel, _MAPPING) + connectivity_channel.subscribe(first_callback.update, try_to_connect=False) + first_connectivities = first_callback.block_until_connectivities_satisfy( + bool) + # Wait for a connection that will never happen because try_to_connect=True + # has not yet been passed. + time.sleep(test_constants.SHORT_TIMEOUT) + second_connectivities = first_callback.connectivities() + connectivity_channel.subscribe(second_callback.update, try_to_connect=True) + third_connectivities = first_callback.block_until_connectivities_satisfy( + lambda connectivities: 2 <= len(connectivities)) + fourth_connectivities = second_callback.block_until_connectivities_satisfy( + bool) + # Wait for a connection that will happen (or may already have happened). + first_callback.block_until_connectivities_satisfy( + lambda connectivities: _READY in connectivities) + second_callback.block_until_connectivities_satisfy( + lambda connectivities: _READY in connectivities) + connectivity_channel.unsubscribe(first_callback.update) + connectivity_channel.unsubscribe(second_callback.update) + + server.shutdown() + server_completion_queue.shutdown() + server_completion_queue_thread.join() + + self.assertSequenceEqual((_IDLE,), first_connectivities) + self.assertSequenceEqual((_IDLE,), second_connectivities) + self.assertNotIn(_TRANSIENT_FAILURE, third_connectivities) + self.assertNotIn(_FATAL_FAILURE, third_connectivities) + self.assertNotIn(_TRANSIENT_FAILURE, fourth_connectivities) + self.assertNotIn(_FATAL_FAILURE, fourth_connectivities) + + def test_reachable_then_unreachable_channel_connectivity(self): + server_completion_queue = _low.CompletionQueue() + server = _low.Server(server_completion_queue, []) + port = server.add_http2_port('[::]:0') + server.start() + server_completion_queue_thread = threading.Thread( + target=_drive_completion_queue, args=(server_completion_queue,)) + server_completion_queue_thread.start() + low_channel = _low.Channel('localhost:%d' % port, ()) + callback = _Callback() + + connectivity_channel = _connectivity_channel.ConnectivityChannel( + low_channel, _MAPPING) + connectivity_channel.subscribe(callback.update, try_to_connect=True) + callback.block_until_connectivities_satisfy( + lambda connectivities: _READY in connectivities) + # Now take down the server and confirm that channel readiness is repudiated. + server.shutdown() + callback.block_until_connectivities_satisfy( + lambda connectivities: connectivities[-1] is not _READY) + connectivity_channel.unsubscribe(callback.update) + + server.shutdown() + server_completion_queue.shutdown() + server_completion_queue_thread.join() + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_test/grpc_test/beta/_utilities_test.py b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py new file mode 100644 index 0000000000..998e74ccf4 --- /dev/null +++ b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py @@ -0,0 +1,123 @@ +# 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 of grpc.beta.utilities.""" + +import threading +import time +import unittest + +from grpc._adapter import _low +from grpc._adapter import _types +from grpc.beta import beta +from grpc.beta import utilities +from grpc.framework.foundation import future +from grpc_test.framework.common import test_constants + + +def _drive_completion_queue(completion_queue): + while True: + event = completion_queue.next(time.time() + 24 * 60 * 60) + if event.type == _types.EventType.QUEUE_SHUTDOWN: + break + + +class _Callback(object): + + def __init__(self): + self._condition = threading.Condition() + self._value = None + + def accept_value(self, value): + with self._condition: + self._value = value + self._condition.notify_all() + + def block_until_called(self): + with self._condition: + while self._value is None: + self._condition.wait() + return self._value + + +class ChannelConnectivityTest(unittest.TestCase): + + def test_lonely_channel_connectivity(self): + channel = beta.create_insecure_channel('localhost', 12345) + callback = _Callback() + + ready_future = utilities.channel_ready_future(channel) + ready_future.add_done_callback(callback.accept_value) + with self.assertRaises(future.TimeoutError): + ready_future.result(test_constants.SHORT_TIMEOUT) + self.assertFalse(ready_future.cancelled()) + self.assertFalse(ready_future.done()) + self.assertTrue(ready_future.running()) + ready_future.cancel() + value_passed_to_callback = callback.block_until_called() + self.assertIs(ready_future, value_passed_to_callback) + self.assertTrue(ready_future.cancelled()) + self.assertTrue(ready_future.done()) + self.assertFalse(ready_future.running()) + + def test_immediately_connectable_channel_connectivity(self): + server_completion_queue = _low.CompletionQueue() + server = _low.Server(server_completion_queue, []) + port = server.add_http2_port('[::]:0') + server.start() + server_completion_queue_thread = threading.Thread( + target=_drive_completion_queue, args=(server_completion_queue,)) + server_completion_queue_thread.start() + channel = beta.create_insecure_channel('localhost', port) + callback = _Callback() + + try: + ready_future = utilities.channel_ready_future(channel) + ready_future.add_done_callback(callback.accept_value) + self.assertIsNone( + ready_future.result(test_constants.SHORT_TIMEOUT)) + value_passed_to_callback = callback.block_until_called() + self.assertIs(ready_future, value_passed_to_callback) + self.assertFalse(ready_future.cancelled()) + self.assertTrue(ready_future.done()) + self.assertFalse(ready_future.running()) + # Cancellation after maturity has no effect. + ready_future.cancel() + self.assertFalse(ready_future.cancelled()) + self.assertTrue(ready_future.done()) + self.assertFalse(ready_future.running()) + finally: + ready_future.cancel() + server.shutdown() + server_completion_queue.shutdown() + server_completion_queue_thread.join() + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py b/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py new file mode 100644 index 0000000000..30bb85f6c3 --- /dev/null +++ b/src/python/grpcio_test/grpc_test/framework/_crust_over_core_face_interface_test.py @@ -0,0 +1,111 @@ +# 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 Face interface compliance of the crust-over-core stack.""" + +import collections +import unittest + +from grpc.framework.core import implementations as core_implementations +from grpc.framework.crust import implementations as crust_implementations +from grpc.framework.foundation import logging_pool +from grpc.framework.interfaces.links import utilities +from grpc_test.framework.common import test_constants +from grpc_test.framework.interfaces.face import test_cases +from grpc_test.framework.interfaces.face import test_interfaces +from grpc_test.framework.interfaces.links import test_utilities + + +class _Implementation(test_interfaces.Implementation): + + def instantiate( + self, methods, method_implementations, multi_method_implementation): + pool = logging_pool.pool(test_constants.POOL_SIZE) + servicer = crust_implementations.servicer( + method_implementations, multi_method_implementation, pool) + + service_end_link = core_implementations.service_end_link( + servicer, test_constants.DEFAULT_TIMEOUT, + test_constants.MAXIMUM_TIMEOUT) + invocation_end_link = core_implementations.invocation_end_link() + invocation_end_link.join_link(service_end_link) + service_end_link.join_link(invocation_end_link) + service_end_link.start() + invocation_end_link.start() + + generic_stub = crust_implementations.generic_stub(invocation_end_link, pool) + # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest. + group = next(iter(methods))[0] + # TODO(nathaniel): Add a "cardinalities_by_group" attribute to + # _digest.TestServiceDigest. + cardinalities = { + method: method_object.cardinality() + for (group, method), method_object in methods.iteritems()} + dynamic_stub = crust_implementations.dynamic_stub( + invocation_end_link, group, cardinalities, pool) + + return generic_stub, {group: dynamic_stub}, ( + invocation_end_link, service_end_link, pool) + + def destantiate(self, memo): + invocation_end_link, service_end_link, pool = memo + invocation_end_link.stop(0).wait() + service_end_link.stop(0).wait() + invocation_end_link.join_link(utilities.NULL_LINK) + service_end_link.join_link(utilities.NULL_LINK) + pool.shutdown(wait=True) + + def invocation_metadata(self): + return object() + + def initial_metadata(self): + return object() + + def terminal_metadata(self): + return object() + + def code(self): + return object() + + def details(self): + return object() + + def metadata_transmitted(self, original_metadata, transmitted_metadata): + return original_metadata is transmitted_metadata + + +def load_tests(loader, tests, pattern): + return unittest.TestSuite( + tests=tuple( + loader.loadTestsFromTestCase(test_case_class) + for test_case_class in test_cases.test_cases(_Implementation()))) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py index 5c8b176da4..87332cf612 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py @@ -134,7 +134,7 @@ class _Servicer(base.Servicer): if group != self._group or method != self._method: controller.fail( '%s != %s or %s != %s' % (group, self._group, method, self._method)) - raise base.NoSuchMethodError() + raise base.NoSuchMethodError(None, None) else: operator = _Operator( controller, controller.on_service_advance, self._pool, diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py new file mode 100644 index 0000000000..363d9ce8f1 --- /dev/null +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_3069_test_constant.py @@ -0,0 +1,37 @@ +# 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. + +"""A test constant working around issue 3069.""" + +# test_constants is referenced from specification in this module. +from grpc_test.framework.common import test_constants # pylint: disable=unused-import + +# TODO(issue 3069): Replace uses of this constant with +# test_constants.SHORT_TIMEOUT. +REALLY_SHORT_TIMEOUT = 0.1 diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py index 857ad5cf3e..8804f3f223 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_blocking_invocation_inline_service.py @@ -37,6 +37,7 @@ from grpc.framework.interfaces.face import face from grpc_test.framework.common import test_constants from grpc_test.framework.common import test_control from grpc_test.framework.common import test_coverage +from grpc_test.framework.interfaces.face import _3069_test_constant from grpc_test.framework.interfaces.face import _digest from grpc_test.framework.interfaces.face import _stock_service from grpc_test.framework.interfaces.face import test_interfaces # pylint: disable=unused-import @@ -170,7 +171,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(), self.assertRaises( face.ExpirationError): self._invoker.blocking(group, method)( - request, test_constants.SHORT_TIMEOUT) + request, _3069_test_constant.REALLY_SHORT_TIMEOUT) def testExpiredUnaryRequestStreamResponse(self): for (group, method), test_messages_sequence in ( @@ -181,7 +182,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(), self.assertRaises( face.ExpirationError): response_iterator = self._invoker.blocking(group, method)( - request, test_constants.SHORT_TIMEOUT) + request, _3069_test_constant.REALLY_SHORT_TIMEOUT) list(response_iterator) def testExpiredStreamRequestUnaryResponse(self): @@ -193,7 +194,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(), self.assertRaises( face.ExpirationError): self._invoker.blocking(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) def testExpiredStreamRequestStreamResponse(self): for (group, method), test_messages_sequence in ( @@ -204,7 +205,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(), self.assertRaises( face.ExpirationError): response_iterator = self._invoker.blocking(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) list(response_iterator) def testFailedUnaryRequestUnaryResponse(self): diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py index ea5cdeaea3..5a78b4bed2 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_event_invocation_synchronous_event_service.py @@ -37,6 +37,7 @@ from grpc.framework.interfaces.face import face from grpc_test.framework.common import test_constants from grpc_test.framework.common import test_control from grpc_test.framework.common import test_coverage +from grpc_test.framework.interfaces.face import _3069_test_constant from grpc_test.framework.interfaces.face import _digest from grpc_test.framework.interfaces.face import _receiver from grpc_test.framework.interfaces.face import _stock_service @@ -264,7 +265,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): self._invoker.event(group, method)( - request, receiver, receiver.abort, test_constants.SHORT_TIMEOUT) + request, receiver, receiver.abort, + _3069_test_constant.REALLY_SHORT_TIMEOUT) receiver.block_until_terminated() self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind) @@ -278,7 +280,8 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): self._invoker.event(group, method)( - request, receiver, receiver.abort, test_constants.SHORT_TIMEOUT) + request, receiver, receiver.abort, + _3069_test_constant.REALLY_SHORT_TIMEOUT) receiver.block_until_terminated() self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind) @@ -290,7 +293,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): receiver = _receiver.Receiver() self._invoker.event(group, method)( - receiver, receiver.abort, test_constants.SHORT_TIMEOUT) + receiver, receiver.abort, _3069_test_constant.REALLY_SHORT_TIMEOUT) receiver.block_until_terminated() self.assertIs(face.Abortion.Kind.EXPIRED, receiver.abortion().kind) @@ -303,7 +306,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): receiver = _receiver.Receiver() call_consumer = self._invoker.event(group, method)( - receiver, receiver.abort, test_constants.SHORT_TIMEOUT) + receiver, receiver.abort, _3069_test_constant.REALLY_SHORT_TIMEOUT) for request in requests: call_consumer.consume(request) receiver.block_until_terminated() diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py index a649362cef..d1107e1576 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_future_invocation_asynchronous_event_service.py @@ -40,6 +40,7 @@ from grpc.framework.interfaces.face import face from grpc_test.framework.common import test_constants from grpc_test.framework.common import test_control from grpc_test.framework.common import test_coverage +from grpc_test.framework.interfaces.face import _3069_test_constant from grpc_test.framework.interfaces.face import _digest from grpc_test.framework.interfaces.face import _stock_service from grpc_test.framework.interfaces.face import test_interfaces # pylint: disable=unused-import @@ -265,7 +266,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): response_future = self._invoker.future( - group, method)(request, test_constants.SHORT_TIMEOUT) + group, method)(request, _3069_test_constant.REALLY_SHORT_TIMEOUT) self.assertIsInstance( response_future.exception(), face.ExpirationError) with self.assertRaises(face.ExpirationError): @@ -279,7 +280,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): response_iterator = self._invoker.future(group, method)( - request, test_constants.SHORT_TIMEOUT) + request, _3069_test_constant.REALLY_SHORT_TIMEOUT) with self.assertRaises(face.ExpirationError): list(response_iterator) @@ -291,7 +292,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): response_future = self._invoker.future(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) self.assertIsInstance( response_future.exception(), face.ExpirationError) with self.assertRaises(face.ExpirationError): @@ -305,7 +306,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.pause(): response_iterator = self._invoker.future(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) with self.assertRaises(face.ExpirationError): list(response_iterator) @@ -317,7 +318,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.fail(): response_future = self._invoker.future(group, method)( - request, test_constants.SHORT_TIMEOUT) + request, _3069_test_constant.REALLY_SHORT_TIMEOUT) # Because the servicer fails outside of the thread from which the # servicer-side runtime called into it its failure is @@ -340,7 +341,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): # expiration of the RPC. with self._control.fail(), self.assertRaises(face.ExpirationError): response_iterator = self._invoker.future(group, method)( - request, test_constants.SHORT_TIMEOUT) + request, _3069_test_constant.REALLY_SHORT_TIMEOUT) list(response_iterator) def testFailedStreamRequestUnaryResponse(self): @@ -351,7 +352,7 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): with self._control.fail(): response_future = self._invoker.future(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) # Because the servicer fails outside of the thread from which the # servicer-side runtime called into it its failure is @@ -374,5 +375,5 @@ class TestCase(test_coverage.Coverage, unittest.TestCase): # expiration of the RPC. with self._control.fail(), self.assertRaises(face.ExpirationError): response_iterator = self._invoker.future(group, method)( - iter(requests), test_constants.SHORT_TIMEOUT) + iter(requests), _3069_test_constant.REALLY_SHORT_TIMEOUT) list(response_iterator) diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py index 1dd2ec3633..808e2c4e36 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/face/_stock_service.py @@ -1,4 +1,4 @@ -B# Copyright 2015, Google Inc. +# Copyright 2015, Google Inc. # All rights reserved. # # Redistribution and use in source and binary forms, with or without |