1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
import socket
from google.protobuf.descriptor_pool import DescriptorPool
import pytest
import pytest_asyncio
from google.protobuf.descriptor_pb2 import FileDescriptorProto
from grpclib.client import Channel
from grpclib.server import Server
from grpclib.reflection.service import ServerReflection
from grpclib.reflection.v1.reflection_pb2 import ServerReflectionRequest
from grpclib.reflection.v1.reflection_pb2 import ServerReflectionResponse
from grpclib.reflection.v1.reflection_pb2 import ErrorResponse
from grpclib.reflection.v1.reflection_grpc import ServerReflectionStub
from dummy_pb2 import DESCRIPTOR
from test_functional import DummyService
@pytest.fixture(name='port')
def port_fixture():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
_, port = s.getsockname()
return port
@pytest_asyncio.fixture(name='channel')
async def channel_fixture(port):
services = [DummyService()]
services = ServerReflection.extend(services)
server = Server(services)
await server.start(port=port)
channel = Channel(port=port)
try:
yield channel
finally:
channel.close()
server.close()
await server.wait_closed()
@pytest.mark.asyncio
async def test_file_by_filename_response(channel):
r1, r2 = await ServerReflectionStub(channel).ServerReflectionInfo([
ServerReflectionRequest(
file_by_filename=DESCRIPTOR.name,
),
ServerReflectionRequest(
file_by_filename='my/missing.proto',
),
])
proto_bytes, = r1.file_descriptor_response.file_descriptor_proto
dummy_proto = FileDescriptorProto()
dummy_proto.ParseFromString(proto_bytes)
assert dummy_proto.name == DESCRIPTOR.name
assert dummy_proto.package == DESCRIPTOR.package
assert r2 == ServerReflectionResponse(
error_response=ErrorResponse(
error_code=5,
error_message='not found',
),
)
@pytest.mark.asyncio
async def test_file_containing_symbol_response(channel):
r1, r2 = await ServerReflectionStub(channel).ServerReflectionInfo([
ServerReflectionRequest(
file_containing_symbol=(
DESCRIPTOR.message_types_by_name['DummyRequest'].full_name
),
),
ServerReflectionRequest(
file_containing_symbol='unknown.Symbol',
),
])
proto_bytes, = r1.file_descriptor_response.file_descriptor_proto
dummy_proto = FileDescriptorProto()
dummy_proto.ParseFromString(proto_bytes)
assert dummy_proto.name == DESCRIPTOR.name
assert dummy_proto.package == DESCRIPTOR.package
assert r2 == ServerReflectionResponse(
error_response=ErrorResponse(
error_code=5,
error_message='not found',
),
)
def test_all_extension_numbers_of_type_response():
pass # message extension is a deprecated feature and not exist in proto3
@pytest.mark.asyncio
async def test_list_services_response(channel):
r1, = await ServerReflectionStub(channel).ServerReflectionInfo([
ServerReflectionRequest(
list_services='',
),
])
service, = r1.list_services_response.service
assert service.name == DESCRIPTOR.services_by_name['DummyService'].full_name
@pytest.mark.asyncio
async def test_file_containing_symbol_response_custom_pool(port):
my_pool = DescriptorPool()
services = [DummyService()]
services = ServerReflection.extend(services, pool=my_pool)
server = Server(services)
await server.start(port=port)
channel = Channel(port=port)
try:
# because we use our own pool (my_pool), there's no descriptors to find.
req = ServerReflectionRequest(
file_containing_symbol=(
DESCRIPTOR.message_types_by_name['DummyRequest'].full_name
),
)
resp, = await ServerReflectionStub(channel).ServerReflectionInfo([req])
assert resp == ServerReflectionResponse(
error_response=ErrorResponse(
error_code=5,
error_message='not found',
),
)
# once we update the pool, we should find the descriptor.
my_pool.AddSerializedFile(DESCRIPTOR.serialized_pb)
resp, = await ServerReflectionStub(channel).ServerReflectionInfo([req])
proto_bytes, = resp.file_descriptor_response.file_descriptor_proto
dummy_proto = FileDescriptorProto()
dummy_proto.ParseFromString(proto_bytes)
assert dummy_proto.name == DESCRIPTOR.name
assert dummy_proto.package == DESCRIPTOR.package
finally:
channel.close()
server.close()
await server.wait_closed()
|
