userver: /data/code/service_template/third_party/userver/testsuite/pytest_plugins/pytest_userver/plugins/grpc/client.py Source File
Loading...
Searching...
No Matches
client.py
1"""
2Make gRPC requests to the service.
3
4@sa @ref scripts/docs/en/userver/tutorial/grpc_service.md
5"""
6
7# pylint: disable=no-member
8import asyncio
9
10import grpc
11import pytest
12
13DEFAULT_TIMEOUT = 15.0
14
15USERVER_CONFIG_HOOKS = ['prepare_config_vars']
16
17
18@pytest.fixture(name='grpc_service_port', scope='session')
19def _grpc_service_port(service_config_yaml) -> int:
20 """
21 Returns the gRPC listener port number of the service that is set in the
22 static configuration file.
23
24 Override this fixture to change the way the gRPC listener port number
25 is retrieved by the testsuite for tests.
26
27 @ingroup userver_testsuite_fixtures
28 """
29 components = service_config_yaml['components_manager']['components']
30 if 'grpc-server' not in components:
31 raise RuntimeError('No grpc-server component')
32 return components['grpc-server']['port']
33
34
35@pytest.fixture(name='grpc_service_endpoint', scope='session')
36def _grpc_service_endpoint(grpc_service_port) -> str:
37 """
38 Returns the gRPC endpoint of the service.
39
40 Override this fixture to change the way the gRPC endpoint
41 is retrieved by the testsuite for tests.
42
43 @ingroup userver_testsuite_fixtures
44 """
45 return f'localhost:{grpc_service_port}'
46
47
48@pytest.fixture(name='grpc_service_timeout', scope='session')
49def _grpc_service_timeout(pytestconfig) -> float:
50 """
51 Returns the gRPC timeout for the service that is set by the command
52 line option `--service-timeout`.
53
54 Override this fixture to change the way the gRPC timeout
55 is set.
56
57 @ingroup userver_testsuite_fixtures
58 """
59 return float(pytestconfig.option.service_timeout) or DEFAULT_TIMEOUT
60
61
62@pytest.fixture(scope='session')
63async def _grpc_session_channel(grpc_service_endpoint):
64 async with grpc.aio.insecure_channel(grpc_service_endpoint) as channel:
65 yield channel
66
67
68@pytest.fixture
69async def grpc_channel(
70 grpc_service_endpoint,
71 grpc_service_deps,
72 grpc_service_timeout,
73 _grpc_session_channel,
74):
75 """
76 Returns the gRPC channel configured by the parameters from the
77 @ref plugins.grpc.grpc_service_endpoint "grpc_service_endpoint" fixture.
78
79 @ingroup userver_testsuite_fixtures
80 """
81 try:
82 await asyncio.wait_for(
83 _grpc_session_channel.channel_ready(),
84 timeout=grpc_service_timeout,
85 )
86 except asyncio.TimeoutError:
87 raise RuntimeError(
88 f'Failed to connect to remote gRPC server by '
89 f'address {grpc_service_endpoint}',
90 )
91 return _grpc_session_channel
92
93
94@pytest.fixture(name='grpc_service_deps')
95def _grpc_service_deps(service_client):
96 """
97 gRPC service dependencies hook. Feel free to override it.
98
99 @ingroup userver_testsuite_fixtures
100 """
101
102
103@pytest.fixture(scope='session')
104def prepare_config_vars(grpc_mockserver_endpoint):
105 def patch_config(config, config_vars):
106 for name in config_vars:
107 if config_vars[name] == '$grpc_mockserver':
108 config_vars[name] = grpc_mockserver_endpoint
109
110 return patch_config