63 grpc_mockserver_session,
65 _grpc_mockserver_ignore_errors,
66) -> pytest_userver.grpc.Mockserver:
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.
73 @snippet samples/grpc_service/testsuite/test_grpc.py Prepare modules
74 @snippet samples/grpc_service/testsuite/test_grpc.py grpc client test
76 Alternatively, you can create a shorthand for mocking frequently-used services:
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
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".
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)
90 @snippet grpc/functional_tests/middleware_client/tests/test_error_status.py Mocked error status
92 To trigger special exceptions in the service's gRPC client, raise these mocked errors from the mock handler:
94 * @ref pytest_userver.grpc._mocked_errors.TimeoutError "pytest_userver.grpc.TimeoutError"
95 * @ref pytest_userver.grpc._mocked_errors.NetworkError "pytest_userver.grpc.NetworkError"
97 @ingroup userver_testsuite_fixtures
99 with grpc_mockserver_session.asyncexc_append_scope(
None if _grpc_mockserver_ignore_errors
else asyncexc_append):
103 grpc_mockserver_session.reset_mocks()
106@pytest.fixture(scope='session')
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`.
113 @ingroup userver_testsuite_fixtures
116 def get_dict_field(parent: dict, field_name: str) -> dict:
117 if parent.setdefault(field_name, {})
is None:
118 parent[field_name] = {}
120 return parent[field_name]
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