1# Copyright 2016 gRPC authors. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""Tests of grpc_reflection.v1alpha.reflection.""" 15 16import unittest 17 18import grpc 19 20from grpc_reflection.v1alpha import reflection 21from grpc_reflection.v1alpha import reflection_pb2 22from grpc_reflection.v1alpha import reflection_pb2_grpc 23 24from google.protobuf import descriptor_pool 25from google.protobuf import descriptor_pb2 26 27from src.proto.grpc.testing import empty_pb2 28from src.proto.grpc.testing.proto2 import empty2_extensions_pb2 29 30from tests.unit import test_common 31 32_EMPTY_PROTO_FILE_NAME = 'src/proto/grpc/testing/empty.proto' 33_EMPTY_PROTO_SYMBOL_NAME = 'grpc.testing.Empty' 34_SERVICE_NAMES = ('Angstrom', 'Bohr', 'Curie', 'Dyson', 'Einstein', 'Feynman', 35 'Galilei') 36_EMPTY_EXTENSIONS_SYMBOL_NAME = 'grpc.testing.proto2.EmptyWithExtensions' 37_EMPTY_EXTENSIONS_NUMBERS = ( 38 124, 39 125, 40 126, 41 127, 42 128, 43) 44 45 46def _file_descriptor_to_proto(descriptor): 47 proto = descriptor_pb2.FileDescriptorProto() 48 descriptor.CopyToProto(proto) 49 return proto.SerializeToString() 50 51 52class ReflectionServicerTest(unittest.TestCase): 53 54 # TODO(https://github.com/grpc/grpc/issues/17844) 55 # Bazel + Python 3 will result in creating two different instance of 56 # DESCRIPTOR for each message. So, the equal comparison between protobuf 57 # returned by stub and manually crafted protobuf will always fail. 58 def _assert_sequence_of_proto_equal(self, x, y): 59 self.assertSequenceEqual( 60 tuple(proto.SerializeToString() for proto in x), 61 tuple(proto.SerializeToString() for proto in y), 62 ) 63 64 def setUp(self): 65 self._server = test_common.test_server() 66 reflection.enable_server_reflection(_SERVICE_NAMES, self._server) 67 port = self._server.add_insecure_port('[::]:0') 68 self._server.start() 69 70 self._channel = grpc.insecure_channel('localhost:%d' % port) 71 self._stub = reflection_pb2_grpc.ServerReflectionStub(self._channel) 72 73 def tearDown(self): 74 self._server.stop(None) 75 self._channel.close() 76 77 def testFileByName(self): 78 requests = ( 79 reflection_pb2.ServerReflectionRequest( 80 file_by_filename=_EMPTY_PROTO_FILE_NAME), 81 reflection_pb2.ServerReflectionRequest( 82 file_by_filename='i-donut-exist'), 83 ) 84 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 85 expected_responses = ( 86 reflection_pb2.ServerReflectionResponse( 87 valid_host='', 88 file_descriptor_response=reflection_pb2.FileDescriptorResponse( 89 file_descriptor_proto=( 90 _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),))), 91 reflection_pb2.ServerReflectionResponse( 92 valid_host='', 93 error_response=reflection_pb2.ErrorResponse( 94 error_code=grpc.StatusCode.NOT_FOUND.value[0], 95 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 96 )), 97 ) 98 self._assert_sequence_of_proto_equal(expected_responses, responses) 99 100 def testFileBySymbol(self): 101 requests = ( 102 reflection_pb2.ServerReflectionRequest( 103 file_containing_symbol=_EMPTY_PROTO_SYMBOL_NAME), 104 reflection_pb2.ServerReflectionRequest( 105 file_containing_symbol='i.donut.exist.co.uk.org.net.me.name.foo' 106 ), 107 ) 108 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 109 expected_responses = ( 110 reflection_pb2.ServerReflectionResponse( 111 valid_host='', 112 file_descriptor_response=reflection_pb2.FileDescriptorResponse( 113 file_descriptor_proto=( 114 _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),))), 115 reflection_pb2.ServerReflectionResponse( 116 valid_host='', 117 error_response=reflection_pb2.ErrorResponse( 118 error_code=grpc.StatusCode.NOT_FOUND.value[0], 119 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 120 )), 121 ) 122 self._assert_sequence_of_proto_equal(expected_responses, responses) 123 124 def testFileContainingExtension(self): 125 requests = ( 126 reflection_pb2.ServerReflectionRequest( 127 file_containing_extension=reflection_pb2.ExtensionRequest( 128 containing_type=_EMPTY_EXTENSIONS_SYMBOL_NAME, 129 extension_number=125, 130 ),), 131 reflection_pb2.ServerReflectionRequest( 132 file_containing_extension=reflection_pb2.ExtensionRequest( 133 containing_type='i.donut.exist.co.uk.org.net.me.name.foo', 134 extension_number=55, 135 ),), 136 ) 137 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 138 expected_responses = ( 139 reflection_pb2.ServerReflectionResponse( 140 valid_host='', 141 file_descriptor_response=reflection_pb2.FileDescriptorResponse( 142 file_descriptor_proto=(_file_descriptor_to_proto( 143 empty2_extensions_pb2.DESCRIPTOR),))), 144 reflection_pb2.ServerReflectionResponse( 145 valid_host='', 146 error_response=reflection_pb2.ErrorResponse( 147 error_code=grpc.StatusCode.NOT_FOUND.value[0], 148 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 149 )), 150 ) 151 self._assert_sequence_of_proto_equal(expected_responses, responses) 152 153 def testExtensionNumbersOfType(self): 154 requests = ( 155 reflection_pb2.ServerReflectionRequest( 156 all_extension_numbers_of_type=_EMPTY_EXTENSIONS_SYMBOL_NAME), 157 reflection_pb2.ServerReflectionRequest( 158 all_extension_numbers_of_type='i.donut.exist.co.uk.net.name.foo' 159 ), 160 ) 161 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 162 expected_responses = ( 163 reflection_pb2.ServerReflectionResponse( 164 valid_host='', 165 all_extension_numbers_response=reflection_pb2. 166 ExtensionNumberResponse( 167 base_type_name=_EMPTY_EXTENSIONS_SYMBOL_NAME, 168 extension_number=_EMPTY_EXTENSIONS_NUMBERS)), 169 reflection_pb2.ServerReflectionResponse( 170 valid_host='', 171 error_response=reflection_pb2.ErrorResponse( 172 error_code=grpc.StatusCode.NOT_FOUND.value[0], 173 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 174 )), 175 ) 176 self._assert_sequence_of_proto_equal(expected_responses, responses) 177 178 def testListServices(self): 179 requests = (reflection_pb2.ServerReflectionRequest(list_services='',),) 180 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 181 expected_responses = (reflection_pb2.ServerReflectionResponse( 182 valid_host='', 183 list_services_response=reflection_pb2.ListServiceResponse( 184 service=tuple( 185 reflection_pb2.ServiceResponse(name=name) 186 for name in _SERVICE_NAMES))),) 187 self._assert_sequence_of_proto_equal(expected_responses, responses) 188 189 def testReflectionServiceName(self): 190 self.assertEqual(reflection.SERVICE_NAME, 191 'grpc.reflection.v1alpha.ServerReflection') 192 193 194if __name__ == '__main__': 195 unittest.main(verbosity=2) 196