userver: /home/antonyzhilin/arcadia/taxi/uservices/userver/testsuite/pytest_plugins/pytest_userver/plugins/grpc/mockserver.py Source File
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
mockserver.py
1"""
2Mocks for the gRPC servers.
3
4@sa @ref scripts/docs/en/userver/tutorial/grpc_service.md
5"""
6
7import grpc
8import pytest
9
11
12# @cond
13
14
15DEFAULT_PORT = 8091
16
17USERVER_CONFIG_HOOKS = ['userver_config_grpc_mockserver']
18
19
20# @endcond
21
22
23@pytest.fixture(scope='session')
24def grpc_mockserver_endpoint(pytestconfig, get_free_port) -> str:
25 """
26 Returns the gRPC endpoint to start the mocking server that is set by
27 command line `--grpc-mockserver-host` and `--grpc-mockserver-port` options.
28
29 For port 0, picks some free port.
30
31 Override this fixture to customize the endpoint used by gRPC mockserver.
32
33 @snippet grpc/functional_tests/basic_chaos/tests-grpcclient/conftest.py grpc_mockserver_endpoint example
34 @ingroup userver_testsuite_fixtures
35 """
36 port = pytestconfig.option.grpc_mockserver_port
37 if pytestconfig.option.service_wait or pytestconfig.option.service_disable:
38 port = port or DEFAULT_PORT
39 if port == 0:
40 port = get_free_port()
41 return f'{pytestconfig.option.grpc_mockserver_host}:{port}'
42
43
44@pytest.fixture(scope='session')
45async def grpc_mockserver_session(grpc_mockserver_endpoint) -> pytest_userver.grpc.MockserverSession:
46 """
47 Returns the gRPC mocking server.
48
49 @warning This is a sharp knife, use with caution! For most use-cases, prefer
50 @ref pytest_userver.plugins.grpc.mockserver.grpc_mockserver "grpc_mockserver" instead.
51
52 @ingroup userver_testsuite_fixtures
53 """
54 server = grpc.aio.server()
55 server.add_insecure_port(grpc_mockserver_endpoint)
56
57 async with pytest_userver.grpc.MockserverSession(server=server, experimental=True) as mockserver:
58 yield mockserver
59
60
61@pytest.fixture
63 grpc_mockserver_session,
64 asyncexc_append,
65 _grpc_mockserver_ignore_errors,
66) -> pytest_userver.grpc.Mockserver:
67 """
68 Returns the gRPC mocking server.
69 In order for gRPC clients in your service to work, mock handlers need to be installed for them using this fixture.
70
71 Example:
72
73 @snippet samples/grpc_service/testsuite/test_grpc.py Prepare modules
74 @snippet samples/grpc_service/testsuite/test_grpc.py grpc client test
75
76 Alternatively, you can create a shorthand for mocking frequently-used services:
77
78 @snippet grpc/functional_tests/metrics/tests/conftest.py Prepare modules
79 @snippet grpc/functional_tests/metrics/tests/conftest.py Prepare server mock
80 @snippet grpc/functional_tests/metrics/tests/test_metrics.py grpc client test
81
82 Mocks are only active within tests after their respective handler functions are created, not between tests.
83 If the service needs the mock during startup, add the fixture that defines your mock to
84 @ref pytest_userver.plugins.service.extra_client_deps "extra_client_deps".
85
86 To return an error status instead of response, use `context` (see
87 [ServicerContext](https://grpc.github.io/grpc/python/grpc_asyncio.html#grpc.aio.ServicerContext)
88 docs):
89
90 @snippet grpc/functional_tests/middleware_client/tests/test_error_status.py Mocked error status
91
92 To trigger special exceptions in the service's gRPC client, raise these mocked errors from the mock handler:
93
94 * @ref pytest_userver.grpc._mocked_errors.TimeoutError "pytest_userver.grpc.TimeoutError"
95 * @ref pytest_userver.grpc._mocked_errors.NetworkError "pytest_userver.grpc.NetworkError"
96
97 @ingroup userver_testsuite_fixtures
98 """
99 with grpc_mockserver_session.asyncexc_append_scope(None if _grpc_mockserver_ignore_errors else asyncexc_append):
100 try:
101 yield pytest_userver.grpc.Mockserver(mockserver_session=grpc_mockserver_session, experimental=True)
102 finally:
103 grpc_mockserver_session.reset_mocks()
104
105
106@pytest.fixture(scope='session')
107def userver_config_grpc_mockserver(grpc_mockserver_endpoint):
108 """
109 Returns a function that adjusts the static config for testsuite.
110 Finds `grpc-client-middleware-pipeline` in config_yaml and
111 enables `grpc-client-middleware-testsuite`.
112
113 @ingroup userver_testsuite_fixtures
114 """
115
116 def get_dict_field(parent: dict, field_name: str) -> dict:
117 if parent.setdefault(field_name, {}) is None:
118 parent[field_name] = {}
119
120 return parent[field_name]
121
122 def patch_config(config_yaml, _config_vars):
123 components = config_yaml['components_manager']['components']
124 if components.get('grpc-client-common', None) is not None:
125 client_middlewares_pipeline = get_dict_field(components, 'grpc-client-middlewares-pipeline')
126 middlewares = get_dict_field(client_middlewares_pipeline, 'middlewares')
127 testsuite_middleware = get_dict_field(middlewares, 'grpc-client-middleware-testsuite')
128 testsuite_middleware['enabled'] = True
129
130 return patch_config
131
132
133@pytest.fixture
134def grpc_mockserver_new(grpc_mockserver) -> pytest_userver.grpc.Mockserver:
135 """
136 @deprecated Legacy alias for @ref pytest_userver.plugins.grpc.mockserver.grpc_mockserver "grpc_mockserver".
137 """
138 return grpc_mockserver
139
140
141# @cond
142
143
144def pytest_addoption(parser):
145 group = parser.getgroup('grpc-mockserver')
146 group.addoption(
147 '--grpc-mockserver-host',
148 default='[::]',
149 help='gRPC mockserver hostname, default is [::]',
150 )
151 group.addoption(
152 '--grpc-mockserver-port',
153 type=int,
154 default=0,
155 help='gRPC mockserver port, by default random port is used',
156 )
157
158
159@pytest.fixture(scope='session')
160def _grpc_mockserver_ignore_errors() -> bool:
161 return False
162
163
164# @endcond