• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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                original_request=requests[0],
94            ),
95            reflection_pb2.ServerReflectionResponse(
96                valid_host="",
97                error_response=reflection_pb2.ErrorResponse(
98                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
99                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
100                ),
101                original_request=requests[1],
102            ),
103        )
104        self.assertEqual(expected_responses, responses)
105
106    def testFileBySymbol(self):
107        requests = (
108            reflection_pb2.ServerReflectionRequest(
109                file_containing_symbol=_EMPTY_PROTO_SYMBOL_NAME
110            ),
111            reflection_pb2.ServerReflectionRequest(
112                file_containing_symbol="i.donut.exist.co.uk.org.net.me.name.foo"
113            ),
114        )
115        responses = tuple(self._stub.ServerReflectionInfo(iter(requests)))
116        expected_responses = (
117            reflection_pb2.ServerReflectionResponse(
118                valid_host="",
119                file_descriptor_response=reflection_pb2.FileDescriptorResponse(
120                    file_descriptor_proto=(
121                        _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),
122                    )
123                ),
124                original_request=requests[0],
125            ),
126            reflection_pb2.ServerReflectionResponse(
127                valid_host="",
128                error_response=reflection_pb2.ErrorResponse(
129                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
130                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
131                ),
132                original_request=requests[1],
133            ),
134        )
135        self.assertEqual(expected_responses, responses)
136
137    def testFileContainingExtension(self):
138        requests = (
139            reflection_pb2.ServerReflectionRequest(
140                file_containing_extension=reflection_pb2.ExtensionRequest(
141                    containing_type=_EMPTY_EXTENSIONS_SYMBOL_NAME,
142                    extension_number=125,
143                ),
144            ),
145            reflection_pb2.ServerReflectionRequest(
146                file_containing_extension=reflection_pb2.ExtensionRequest(
147                    containing_type="i.donut.exist.co.uk.org.net.me.name.foo",
148                    extension_number=55,
149                ),
150            ),
151        )
152        responses = tuple(self._stub.ServerReflectionInfo(iter(requests)))
153        expected_responses = (
154            reflection_pb2.ServerReflectionResponse(
155                valid_host="",
156                file_descriptor_response=reflection_pb2.FileDescriptorResponse(
157                    file_descriptor_proto=(
158                        _file_descriptor_to_proto(
159                            empty2_extensions_pb2.DESCRIPTOR
160                        ),
161                        _file_descriptor_to_proto(empty2_pb2.DESCRIPTOR),
162                    )
163                ),
164                original_request=requests[0],
165            ),
166            reflection_pb2.ServerReflectionResponse(
167                valid_host="",
168                error_response=reflection_pb2.ErrorResponse(
169                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
170                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
171                ),
172                original_request=requests[1],
173            ),
174        )
175        self.assertEqual(expected_responses, responses)
176
177    def testExtensionNumbersOfType(self):
178        requests = (
179            reflection_pb2.ServerReflectionRequest(
180                all_extension_numbers_of_type=_EMPTY_EXTENSIONS_SYMBOL_NAME
181            ),
182            reflection_pb2.ServerReflectionRequest(
183                all_extension_numbers_of_type="i.donut.exist.co.uk.net.name.foo"
184            ),
185        )
186        responses = tuple(self._stub.ServerReflectionInfo(iter(requests)))
187        expected_responses = (
188            reflection_pb2.ServerReflectionResponse(
189                valid_host="",
190                all_extension_numbers_response=reflection_pb2.ExtensionNumberResponse(
191                    base_type_name=_EMPTY_EXTENSIONS_SYMBOL_NAME,
192                    extension_number=_EMPTY_EXTENSIONS_NUMBERS,
193                ),
194                original_request=requests[0],
195            ),
196            reflection_pb2.ServerReflectionResponse(
197                valid_host="",
198                error_response=reflection_pb2.ErrorResponse(
199                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
200                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
201                ),
202                original_request=requests[1],
203            ),
204        )
205        self.assertEqual(expected_responses, responses)
206
207    def testListServices(self):
208        requests = (
209            reflection_pb2.ServerReflectionRequest(
210                list_services="",
211            ),
212        )
213        responses = tuple(self._stub.ServerReflectionInfo(iter(requests)))
214        expected_responses = (
215            reflection_pb2.ServerReflectionResponse(
216                valid_host="",
217                list_services_response=reflection_pb2.ListServiceResponse(
218                    service=tuple(
219                        reflection_pb2.ServiceResponse(name=name)
220                        for name in _SERVICE_NAMES
221                    )
222                ),
223                original_request=requests[0],
224            ),
225        )
226        self.assertEqual(expected_responses, responses)
227
228    def testReflectionServiceName(self):
229        self.assertEqual(
230            reflection.SERVICE_NAME, "grpc.reflection.v1alpha.ServerReflection"
231        )
232
233
234if __name__ == "__main__":
235    unittest.main(verbosity=2)
236