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 sys 17import unittest 18 19from google.protobuf import descriptor_pb2 20from google.protobuf import descriptor_pool 21import grpc 22from grpc_reflection.v1alpha import reflection 23from grpc_reflection.v1alpha import reflection_pb2 24from grpc_reflection.v1alpha import reflection_pb2_grpc 25 26from src.proto.grpc.testing import empty_pb2 27from src.proto.grpc.testing.proto2 import empty2_extensions_pb2 28from src.proto.grpc.testing.proto2 import empty2_pb2 29from tests.unit import test_common 30 31_EMPTY_PROTO_FILE_NAME = "src/proto/grpc/testing/empty.proto" 32_EMPTY_PROTO_SYMBOL_NAME = "grpc.testing.Empty" 33_SERVICE_NAMES = ( 34 "Angstrom", 35 "Bohr", 36 "Curie", 37 "Dyson", 38 "Einstein", 39 "Feynman", 40 "Galilei", 41) 42_EMPTY_EXTENSIONS_SYMBOL_NAME = "grpc.testing.proto2.EmptyWithExtensions" 43_EMPTY_EXTENSIONS_NUMBERS = ( 44 124, 45 125, 46 126, 47 127, 48 128, 49) 50 51 52def _file_descriptor_to_proto(descriptor): 53 proto = descriptor_pb2.FileDescriptorProto() 54 descriptor.CopyToProto(proto) 55 return proto.SerializeToString() 56 57 58@unittest.skipIf( 59 sys.version_info[0] < 3, "ProtoBuf descriptor has moved on from Python2" 60) 61class ReflectionServicerTest(unittest.TestCase): 62 def setUp(self): 63 self._server = test_common.test_server() 64 reflection.enable_server_reflection(_SERVICE_NAMES, self._server) 65 port = self._server.add_insecure_port("[::]:0") 66 self._server.start() 67 68 self._channel = grpc.insecure_channel("localhost:%d" % port) 69 self._stub = reflection_pb2_grpc.ServerReflectionStub(self._channel) 70 71 def tearDown(self): 72 self._server.stop(None) 73 self._channel.close() 74 75 def testFileByName(self): 76 requests = ( 77 reflection_pb2.ServerReflectionRequest( 78 file_by_filename=_EMPTY_PROTO_FILE_NAME 79 ), 80 reflection_pb2.ServerReflectionRequest( 81 file_by_filename="i-donut-exist" 82 ), 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 ) 92 ), 93 ), 94 reflection_pb2.ServerReflectionResponse( 95 valid_host="", 96 error_response=reflection_pb2.ErrorResponse( 97 error_code=grpc.StatusCode.NOT_FOUND.value[0], 98 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 99 ), 100 ), 101 ) 102 self.assertEqual(expected_responses, responses) 103 104 def testFileBySymbol(self): 105 requests = ( 106 reflection_pb2.ServerReflectionRequest( 107 file_containing_symbol=_EMPTY_PROTO_SYMBOL_NAME 108 ), 109 reflection_pb2.ServerReflectionRequest( 110 file_containing_symbol="i.donut.exist.co.uk.org.net.me.name.foo" 111 ), 112 ) 113 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 114 expected_responses = ( 115 reflection_pb2.ServerReflectionResponse( 116 valid_host="", 117 file_descriptor_response=reflection_pb2.FileDescriptorResponse( 118 file_descriptor_proto=( 119 _file_descriptor_to_proto(empty_pb2.DESCRIPTOR), 120 ) 121 ), 122 ), 123 reflection_pb2.ServerReflectionResponse( 124 valid_host="", 125 error_response=reflection_pb2.ErrorResponse( 126 error_code=grpc.StatusCode.NOT_FOUND.value[0], 127 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 128 ), 129 ), 130 ) 131 self.assertEqual(expected_responses, responses) 132 133 def testFileContainingExtension(self): 134 requests = ( 135 reflection_pb2.ServerReflectionRequest( 136 file_containing_extension=reflection_pb2.ExtensionRequest( 137 containing_type=_EMPTY_EXTENSIONS_SYMBOL_NAME, 138 extension_number=125, 139 ), 140 ), 141 reflection_pb2.ServerReflectionRequest( 142 file_containing_extension=reflection_pb2.ExtensionRequest( 143 containing_type="i.donut.exist.co.uk.org.net.me.name.foo", 144 extension_number=55, 145 ), 146 ), 147 ) 148 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 149 expected_responses = ( 150 reflection_pb2.ServerReflectionResponse( 151 valid_host="", 152 file_descriptor_response=reflection_pb2.FileDescriptorResponse( 153 file_descriptor_proto=( 154 _file_descriptor_to_proto( 155 empty2_extensions_pb2.DESCRIPTOR 156 ), 157 _file_descriptor_to_proto(empty2_pb2.DESCRIPTOR), 158 ) 159 ), 160 ), 161 reflection_pb2.ServerReflectionResponse( 162 valid_host="", 163 error_response=reflection_pb2.ErrorResponse( 164 error_code=grpc.StatusCode.NOT_FOUND.value[0], 165 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 166 ), 167 ), 168 ) 169 self.assertEqual(expected_responses, responses) 170 171 def testExtensionNumbersOfType(self): 172 requests = ( 173 reflection_pb2.ServerReflectionRequest( 174 all_extension_numbers_of_type=_EMPTY_EXTENSIONS_SYMBOL_NAME 175 ), 176 reflection_pb2.ServerReflectionRequest( 177 all_extension_numbers_of_type="i.donut.exist.co.uk.net.name.foo" 178 ), 179 ) 180 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 181 expected_responses = ( 182 reflection_pb2.ServerReflectionResponse( 183 valid_host="", 184 all_extension_numbers_response=reflection_pb2.ExtensionNumberResponse( 185 base_type_name=_EMPTY_EXTENSIONS_SYMBOL_NAME, 186 extension_number=_EMPTY_EXTENSIONS_NUMBERS, 187 ), 188 ), 189 reflection_pb2.ServerReflectionResponse( 190 valid_host="", 191 error_response=reflection_pb2.ErrorResponse( 192 error_code=grpc.StatusCode.NOT_FOUND.value[0], 193 error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(), 194 ), 195 ), 196 ) 197 self.assertEqual(expected_responses, responses) 198 199 def testListServices(self): 200 requests = ( 201 reflection_pb2.ServerReflectionRequest( 202 list_services="", 203 ), 204 ) 205 responses = tuple(self._stub.ServerReflectionInfo(iter(requests))) 206 expected_responses = ( 207 reflection_pb2.ServerReflectionResponse( 208 valid_host="", 209 list_services_response=reflection_pb2.ListServiceResponse( 210 service=tuple( 211 reflection_pb2.ServiceResponse(name=name) 212 for name in _SERVICE_NAMES 213 ) 214 ), 215 ), 216 ) 217 self.assertEqual(expected_responses, responses) 218 219 def testReflectionServiceName(self): 220 self.assertEqual( 221 reflection.SERVICE_NAME, "grpc.reflection.v1alpha.ServerReflection" 222 ) 223 224 225if __name__ == "__main__": 226 unittest.main(verbosity=2) 227