| # 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. |
| """Tests of grpc_reflection.v1alpha.reflection.""" |
| |
| import unittest |
| |
| import grpc |
| from grpc_reflection.v1alpha import reflection |
| from grpc_reflection.v1alpha import reflection_pb2 |
| from grpc_reflection.v1alpha import reflection_pb2_grpc |
| |
| from google.protobuf import descriptor_pool |
| from google.protobuf import descriptor_pb2 |
| |
| from src.proto.grpc.testing import empty_pb2 |
| from src.proto.grpc.testing.proto2 import empty2_extensions_pb2 |
| |
| from tests.unit import test_common |
| |
| _EMPTY_PROTO_FILE_NAME = 'src/proto/grpc/testing/empty.proto' |
| _EMPTY_PROTO_SYMBOL_NAME = 'grpc.testing.Empty' |
| _SERVICE_NAMES = ('Angstrom', 'Bohr', 'Curie', 'Dyson', 'Einstein', 'Feynman', |
| 'Galilei') |
| _EMPTY_EXTENSIONS_SYMBOL_NAME = 'grpc.testing.proto2.EmptyWithExtensions' |
| _EMPTY_EXTENSIONS_NUMBERS = ( |
| 124, |
| 125, |
| 126, |
| 127, |
| 128, |
| ) |
| |
| |
| def _file_descriptor_to_proto(descriptor): |
| proto = descriptor_pb2.FileDescriptorProto() |
| descriptor.CopyToProto(proto) |
| return proto.SerializeToString() |
| |
| |
| class ReflectionServicerTest(unittest.TestCase): |
| |
| def setUp(self): |
| self._server = test_common.test_server() |
| reflection.enable_server_reflection(_SERVICE_NAMES, self._server) |
| port = self._server.add_insecure_port('[::]:0') |
| self._server.start() |
| |
| self._channel = grpc.insecure_channel('localhost:%d' % port) |
| self._stub = reflection_pb2_grpc.ServerReflectionStub(self._channel) |
| |
| def tearDown(self): |
| self._server.stop(None) |
| self._channel.close() |
| |
| def testFileByName(self): |
| requests = ( |
| reflection_pb2.ServerReflectionRequest( |
| file_by_filename=_EMPTY_PROTO_FILE_NAME), |
| reflection_pb2.ServerReflectionRequest( |
| file_by_filename='i-donut-exist'), |
| ) |
| responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) |
| expected_responses = ( |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| file_descriptor_response=reflection_pb2.FileDescriptorResponse( |
| file_descriptor_proto=( |
| _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),))), |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| error_response=reflection_pb2.ErrorResponse( |
| error_code=grpc.StatusCode.NOT_FOUND.value[0], |
| error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), |
| )), |
| ) |
| self.assertSequenceEqual(expected_responses, responses) |
| |
| def testFileBySymbol(self): |
| requests = ( |
| reflection_pb2.ServerReflectionRequest( |
| file_containing_symbol=_EMPTY_PROTO_SYMBOL_NAME), |
| reflection_pb2.ServerReflectionRequest( |
| file_containing_symbol='i.donut.exist.co.uk.org.net.me.name.foo' |
| ), |
| ) |
| responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) |
| expected_responses = ( |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| file_descriptor_response=reflection_pb2.FileDescriptorResponse( |
| file_descriptor_proto=( |
| _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),))), |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| error_response=reflection_pb2.ErrorResponse( |
| error_code=grpc.StatusCode.NOT_FOUND.value[0], |
| error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), |
| )), |
| ) |
| self.assertSequenceEqual(expected_responses, responses) |
| |
| def testFileContainingExtension(self): |
| requests = ( |
| reflection_pb2.ServerReflectionRequest( |
| file_containing_extension=reflection_pb2.ExtensionRequest( |
| containing_type=_EMPTY_EXTENSIONS_SYMBOL_NAME, |
| extension_number=125, |
| ),), |
| reflection_pb2.ServerReflectionRequest( |
| file_containing_extension=reflection_pb2.ExtensionRequest( |
| containing_type='i.donut.exist.co.uk.org.net.me.name.foo', |
| extension_number=55, |
| ),), |
| ) |
| responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) |
| expected_responses = ( |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| file_descriptor_response=reflection_pb2.FileDescriptorResponse( |
| file_descriptor_proto=(_file_descriptor_to_proto( |
| empty2_extensions_pb2.DESCRIPTOR),))), |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| error_response=reflection_pb2.ErrorResponse( |
| error_code=grpc.StatusCode.NOT_FOUND.value[0], |
| error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), |
| )), |
| ) |
| self.assertSequenceEqual(expected_responses, responses) |
| |
| def testExtensionNumbersOfType(self): |
| requests = ( |
| reflection_pb2.ServerReflectionRequest( |
| all_extension_numbers_of_type=_EMPTY_EXTENSIONS_SYMBOL_NAME), |
| reflection_pb2.ServerReflectionRequest( |
| all_extension_numbers_of_type='i.donut.exist.co.uk.net.name.foo' |
| ), |
| ) |
| responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) |
| expected_responses = ( |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| all_extension_numbers_response=reflection_pb2. |
| ExtensionNumberResponse( |
| base_type_name=_EMPTY_EXTENSIONS_SYMBOL_NAME, |
| extension_number=_EMPTY_EXTENSIONS_NUMBERS)), |
| reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| error_response=reflection_pb2.ErrorResponse( |
| error_code=grpc.StatusCode.NOT_FOUND.value[0], |
| error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), |
| )), |
| ) |
| self.assertSequenceEqual(expected_responses, responses) |
| |
| def testListServices(self): |
| requests = (reflection_pb2.ServerReflectionRequest(list_services='',),) |
| responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) |
| expected_responses = (reflection_pb2.ServerReflectionResponse( |
| valid_host='', |
| list_services_response=reflection_pb2.ListServiceResponse( |
| service=tuple( |
| reflection_pb2.ServiceResponse(name=name) |
| for name in _SERVICE_NAMES))),) |
| self.assertSequenceEqual(expected_responses, responses) |
| |
| def testReflectionServiceName(self): |
| self.assertEqual(reflection.SERVICE_NAME, |
| 'grpc.reflection.v1alpha.ServerReflection') |
| |
| |
| if __name__ == '__main__': |
| unittest.main(verbosity=2) |
