diff options
Diffstat (limited to 'src/python/grpcio_tests/tests/unit/beta/_face_interface_test.py')
| -rw-r--r-- | src/python/grpcio_tests/tests/unit/beta/_face_interface_test.py | 132 |
1 files changed, 0 insertions, 132 deletions
diff --git a/src/python/grpcio_tests/tests/unit/beta/_face_interface_test.py b/src/python/grpcio_tests/tests/unit/beta/_face_interface_test.py deleted file mode 100644 index c99738e085..0000000000 --- a/src/python/grpcio_tests/tests/unit/beta/_face_interface_test.py +++ /dev/null @@ -1,132 +0,0 @@ -# Copyright 2015 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. -"""Tests Face interface compliance of the gRPC Python Beta API.""" - -import collections -import unittest - -import six - -from grpc.beta import implementations -from grpc.beta import interfaces -from tests.unit import resources -from tests.unit import test_common as grpc_test_common -from tests.unit.beta import test_utilities -from tests.unit.framework.common import test_constants -from tests.unit.framework.interfaces.face import test_cases -from tests.unit.framework.interfaces.face import test_interfaces - -_SERVER_HOST_OVERRIDE = 'foo.test.google.fr' - - -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 six.iteritems(test_methods): - 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): - serialization_behaviors = _serialization_behaviors_from_test_methods( - methods) - # TODO(nathaniel): Add a "groups" attribute to _digest.TestServiceDigest. - service = 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 six.iteritems(methods) - } - - server_options = implementations.server_options( - request_deserializers=serialization_behaviors.request_deserializers, - response_serializers=serialization_behaviors.response_serializers, - thread_pool_size=test_constants.POOL_SIZE) - server = implementations.server( - method_implementations, options=server_options) - server_credentials = implementations.ssl_server_credentials([ - ( - resources.private_key(), - resources.certificate_chain(), - ), - ]) - port = server.add_secure_port('[::]:0', server_credentials) - server.start() - channel_credentials = implementations.ssl_channel_credentials( - resources.test_root_certificates()) - channel = test_utilities.not_really_secure_channel( - 'localhost', port, channel_credentials, _SERVER_HOST_OVERRIDE) - stub_options = implementations.stub_options( - request_serializers=serialization_behaviors.request_serializers, - response_deserializers=serialization_behaviors. - response_deserializers, - thread_pool_size=test_constants.POOL_SIZE) - generic_stub = implementations.generic_stub( - channel, options=stub_options) - dynamic_stub = implementations.dynamic_stub( - channel, service, cardinalities, options=stub_options) - return generic_stub, {service: dynamic_stub}, server - - def destantiate(self, memo): - memo.stop(test_constants.SHORT_TIMEOUT).wait() - - def invocation_metadata(self): - return grpc_test_common.INVOCATION_INITIAL_METADATA - - def initial_metadata(self): - return grpc_test_common.SERVICE_INITIAL_METADATA - - def terminal_metadata(self): - return grpc_test_common.SERVICE_TERMINAL_METADATA - - def code(self): - return interfaces.StatusCode.OK - - def details(self): - return grpc_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) |