• 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 logging
17import unittest
18
19from google.protobuf import descriptor_pb2
20import grpc
21from grpc.experimental import aio
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_aio.unit._test_base import AioTestBase
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
58class ReflectionServicerTest(AioTestBase):
59    async def setUp(self):
60        self._server = aio.server()
61        reflection.enable_server_reflection(_SERVICE_NAMES, self._server)
62        port = self._server.add_insecure_port("[::]:0")
63        await self._server.start()
64
65        self._channel = aio.insecure_channel("localhost:%d" % port)
66        self._stub = reflection_pb2_grpc.ServerReflectionStub(self._channel)
67
68    async def tearDown(self):
69        await self._server.stop(None)
70        await self._channel.close()
71
72    async def test_file_by_name(self):
73        requests = (
74            reflection_pb2.ServerReflectionRequest(
75                file_by_filename=_EMPTY_PROTO_FILE_NAME
76            ),
77            reflection_pb2.ServerReflectionRequest(
78                file_by_filename="i-donut-exist"
79            ),
80        )
81        responses = []
82        async for response in self._stub.ServerReflectionInfo(iter(requests)):
83            responses.append(response)
84        expected_responses = (
85            reflection_pb2.ServerReflectionResponse(
86                valid_host="",
87                file_descriptor_response=reflection_pb2.FileDescriptorResponse(
88                    file_descriptor_proto=(
89                        _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),
90                    )
91                ),
92            ),
93            reflection_pb2.ServerReflectionResponse(
94                valid_host="",
95                error_response=reflection_pb2.ErrorResponse(
96                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
97                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
98                ),
99            ),
100        )
101        self.assertSequenceEqual(expected_responses, responses)
102
103    async def test_file_by_symbol(self):
104        requests = (
105            reflection_pb2.ServerReflectionRequest(
106                file_containing_symbol=_EMPTY_PROTO_SYMBOL_NAME
107            ),
108            reflection_pb2.ServerReflectionRequest(
109                file_containing_symbol="i.donut.exist.co.uk.org.net.me.name.foo"
110            ),
111        )
112        responses = []
113        async for response in self._stub.ServerReflectionInfo(iter(requests)):
114            responses.append(response)
115        expected_responses = (
116            reflection_pb2.ServerReflectionResponse(
117                valid_host="",
118                file_descriptor_response=reflection_pb2.FileDescriptorResponse(
119                    file_descriptor_proto=(
120                        _file_descriptor_to_proto(empty_pb2.DESCRIPTOR),
121                    )
122                ),
123            ),
124            reflection_pb2.ServerReflectionResponse(
125                valid_host="",
126                error_response=reflection_pb2.ErrorResponse(
127                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
128                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
129                ),
130            ),
131        )
132        self.assertSequenceEqual(expected_responses, responses)
133
134    async def test_file_containing_extension(self):
135        requests = (
136            reflection_pb2.ServerReflectionRequest(
137                file_containing_extension=reflection_pb2.ExtensionRequest(
138                    containing_type=_EMPTY_EXTENSIONS_SYMBOL_NAME,
139                    extension_number=125,
140                ),
141            ),
142            reflection_pb2.ServerReflectionRequest(
143                file_containing_extension=reflection_pb2.ExtensionRequest(
144                    containing_type="i.donut.exist.co.uk.org.net.me.name.foo",
145                    extension_number=55,
146                ),
147            ),
148        )
149        responses = []
150        async for response in self._stub.ServerReflectionInfo(iter(requests)):
151            responses.append(response)
152        expected_responses = (
153            reflection_pb2.ServerReflectionResponse(
154                valid_host="",
155                file_descriptor_response=reflection_pb2.FileDescriptorResponse(
156                    file_descriptor_proto=(
157                        _file_descriptor_to_proto(
158                            empty2_extensions_pb2.DESCRIPTOR
159                        ),
160                        _file_descriptor_to_proto(empty2_pb2.DESCRIPTOR),
161                    )
162                ),
163            ),
164            reflection_pb2.ServerReflectionResponse(
165                valid_host="",
166                error_response=reflection_pb2.ErrorResponse(
167                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
168                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
169                ),
170            ),
171        )
172        self.assertSequenceEqual(expected_responses, responses)
173
174    async def test_extension_numbers_of_type(self):
175        requests = (
176            reflection_pb2.ServerReflectionRequest(
177                all_extension_numbers_of_type=_EMPTY_EXTENSIONS_SYMBOL_NAME
178            ),
179            reflection_pb2.ServerReflectionRequest(
180                all_extension_numbers_of_type="i.donut.exist.co.uk.net.name.foo"
181            ),
182        )
183        responses = []
184        async for response in self._stub.ServerReflectionInfo(iter(requests)):
185            responses.append(response)
186        expected_responses = (
187            reflection_pb2.ServerReflectionResponse(
188                valid_host="",
189                all_extension_numbers_response=reflection_pb2.ExtensionNumberResponse(
190                    base_type_name=_EMPTY_EXTENSIONS_SYMBOL_NAME,
191                    extension_number=_EMPTY_EXTENSIONS_NUMBERS,
192                ),
193            ),
194            reflection_pb2.ServerReflectionResponse(
195                valid_host="",
196                error_response=reflection_pb2.ErrorResponse(
197                    error_code=grpc.StatusCode.NOT_FOUND.value[0],
198                    error_message=grpc.StatusCode.NOT_FOUND.value[1].encode(),
199                ),
200            ),
201        )
202        self.assertSequenceEqual(expected_responses, responses)
203
204    async def test_list_services(self):
205        requests = (
206            reflection_pb2.ServerReflectionRequest(
207                list_services="",
208            ),
209        )
210        responses = []
211        async for response in self._stub.ServerReflectionInfo(iter(requests)):
212            responses.append(response)
213        expected_responses = (
214            reflection_pb2.ServerReflectionResponse(
215                valid_host="",
216                list_services_response=reflection_pb2.ListServiceResponse(
217                    service=tuple(
218                        reflection_pb2.ServiceResponse(name=name)
219                        for name in _SERVICE_NAMES
220                    )
221                ),
222            ),
223        )
224        self.assertSequenceEqual(expected_responses, responses)
225
226    def test_reflection_service_name(self):
227        self.assertEqual(
228            reflection.SERVICE_NAME, "grpc.reflection.v1alpha.ServerReflection"
229        )
230
231
232if __name__ == "__main__":
233    logging.basicConfig(level=logging.DEBUG)
234    unittest.main(verbosity=2)
235