Detailed Description

Fixtures to use with testsuite.

Namespaces
namespace	pytest_userver.plugins.core
	Python plugin that provides core fixtures for functional tests with testsuite; see Functional service tests (testsuite) for an introduction.
namespace	pytest_userver.plugins.grpc
	Python plugin that provides gRPC fixtures for functional tests with testsuite; see Functional service tests (testsuite) for an introduction.
namespace	pytest_userver.plugins.sql_coverage
	Plugin that imports the required fixtures for checking SQL/YQL coverage.
namespace	pytest_userver.plugins.ydb
	pytest plugin that provides YDB fixtures for functional tests with testsuite; see YDB for an introduction.

Functions
pathlib.Path	pytest_userver.plugins.base.service_source_dir (pytestconfig)
	Returns the path to the service source directory that is set by command line --service-source-dir option.
pathlib.Path	pytest_userver.plugins.base.build_dir (pytestconfig)
	Returns the build directory set by command line --build-dir option.
pathlib.Path	pytest_userver.plugins.base.service_binary (pytestconfig)
	Returns the path to service binary set by command line --service-binary option.
int	pytest_userver.plugins.base.service_port (pytestconfig, _original_service_config, choose_free_port)
	Returns the main listener port number of the service set by command line --service-port option.
int	pytest_userver.plugins.base.monitor_port (pytestconfig, _original_service_config, choose_free_port)
	Returns the monitor listener port number of the service set by command line --monitor-port option.
bool	pytest_userver.plugins.base.congestion_control_fake_mode ()
	Returns congestion control fake-mode value.
Callable[[int\|None], int]	pytest_userver.plugins.base.choose_free_port (pytestconfig, get_free_port, _testsuite_socket_cleanup, _testsuite_default_af)
	A function that chooses a free port based on the optional hint given in the parameter.
InvalidationState	pytest_userver.plugins.caches.cache_invalidation_state ()
	A fixture for notifying the service of changes in cache data sources.
CacheControl	pytest_userver.plugins.caches.userver_cache_control (_userver_cache_control_context, _userver_cache_fixtures, request)
	Userver cache control handler.
pathlib.Path	pytest_userver.plugins.config.service_config_path (pytestconfig, service_binary)
	Returns the path to service.yaml file set by command line --service-config option.
pathlib.Path	pytest_userver.plugins.config.db_dump_schema_path (service_binary, service_tmpdir)
	Runs the service binary with --dump-db-schema argument, dumps the 0_db_schema.sql file with database schema and returns path to it.
pathlib.Path\|None	pytest_userver.plugins.config.service_config_vars_path (pytestconfig)
	Returns the path to config_vars.yaml file set by command line --service-config-vars option.
pathlib.Path\|None	pytest_userver.plugins.config.service_secdist_path (pytestconfig)
	Returns the path to secure_data.json file set by command line --service-secdist option.
pathlib.Path	pytest_userver.plugins.config.config_fallback_path (pytestconfig)
	Returns the path to dynamic config fallback file set by command line --config-fallback option.
	pytest_userver.plugins.config.service_tmpdir (service_binary, tmp_path_factory)
	Returns the path for temporary files.
pathlib.Path	pytest_userver.plugins.config.service_config_path_temp (service_tmpdir, service_config, service_config_yaml, service_config_vars)
	Dumps the contents of the service_config_yaml and service_config_vars into a static config for testsuite and returns the path to the config file.
dict	pytest_userver.plugins.config.service_config_yaml (_service_config_hooked)
	Returns the static config values after the USERVER_CONFIG_HOOKS were applied (if any).
dict	pytest_userver.plugins.config.service_config_vars (_service_config_hooked)
	Returns the static config variables (config_vars.yaml) values after the USERVER_CONFIG_HOOKS were applied (if any).
Callable[[Any, dict], Any]	pytest_userver.plugins.config.substitute_config_vars (service_env)
	A function that takes config_yaml, config_vars and applies all substitutions just like the service would.
dict	pytest_userver.plugins.config.service_config (service_config_yaml, service_config_vars, substitute_config_vars)
	Returns the static config values after the USERVER_CONFIG_HOOKS were applied (if any) and with all the '$', environment and fallback variables substituted.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_substitutions (_service_config_substitution_vars)
	Replaces substitution vars in all strings within config_vars using string.Template.substitute.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_http_server (service_port, monitor_port)
	Returns a function that adjusts the static configuration file for testsuite.
list[str]	pytest_userver.plugins.config.allowed_url_prefixes_extra ()
	By default, userver HTTP client is only allowed to talk to mockserver when running in testsuite.
list[str]	pytest_userver.plugins.config.allowed_url_prefixes (mockserver_info, mockserver_ssl_info, allowed_url_prefixes_extra)
	Returns final list of allowed urls prefixes when running in testsuite.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_http_client (allowed_url_prefixes)
	Returns a function that adjusts the static configuration file for testsuite.
str	pytest_userver.plugins.config.userver_default_log_level ()
	Default log level to use in userver if no command line option was provided.
str	pytest_userver.plugins.config.userver_log_level (pytestconfig, userver_default_log_level)
	Returns –service-log-level value if provided, otherwise returns userver_default_log_level() value from fixture.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_logging (userver_log_level, _service_logfile_path)
	Returns a function that adjusts the static configuration file for testsuite.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_logging_otlp ()
	Returns a function that adjusts the static configuration file for testsuite.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_testsuite (pytestconfig, mockserver_info)
	Returns a function that adjusts the static configuration file for testsuite.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_secdist (service_secdist_path)
	Returns a function that adjusts the static configuration file for testsuite.
ServiceConfigPatch	pytest_userver.plugins.config.userver_config_deadlock_detector (str userver_deadlock_detector_mode)
	Returns a function that adjusts the static configuration file for testsuite.
str	pytest_userver.plugins.config.userver_deadlock_detector_mode ()
	Returns Deadlock detector mode for testsuite.
pathlib.Path	pytest_userver.plugins.dumps.userver_dumps_root (tmp_path_factory)
	The directory which the service will use for cache dumps.
	pytest_userver.plugins.dumps.read_latest_dump (userver_dumps_root)
	Read the latest dump produced by a specified dumper.
	pytest_userver.plugins.dumps.cleanup_userver_dumps (userver_dumps_root, request)
	To avoid leaking dumps between tests, cache_dump_dir must be cleaned after each test.
dynconf.DynamicConfig	pytest_userver.plugins.dynamic_config.dynamic_config (request, search_path, object_substitute, cache_invalidation_state, _dynamic_config_defaults_storage, config_service_defaults, dynamic_config_changelog, _dynconf_load_json_cached, dynconf_cache_names)
	Fixture that allows to control dynamic config values used by the service.
dynconf.ConfigValuesDict	pytest_userver.plugins.dynamic_config.dynamic_config_fallback_patch ()
	Override this fixture to replace some dynamic config values specifically for testsuite tests:
dynconf.ConfigValuesDict	pytest_userver.plugins.dynamic_config.config_service_defaults (config_fallback_path, dynamic_config_fallback_patch)
	Fixture that returns default values for dynamic config.
	pytest_userver.plugins.dynamic_config.userver_config_dynconf_cache (service_tmpdir)
	Returns a function that adjusts the static configuration file for the testsuite.
	pytest_userver.plugins.dynamic_config.userver_config_dynconf_fallback (config_service_defaults)
	Returns a function that adjusts the static configuration file for the testsuite.
	pytest_userver.plugins.dynamic_config.userver_config_dynconf_url (mockserver_info)
	Returns a function that adjusts the static configuration file for the testsuite.
None	pytest_userver.plugins.dynamic_config.mock_configs_service (mockserver, dynconf.DynamicConfig dynamic_config, dynconf._Changelog dynamic_config_changelog)
	Adds a mockserver handler that forwards dynamic_config to service's dynamic-config-client component.
int	pytest_userver.plugins.grpc.client.grpc_service_port_fallback ()
	Returns the gRPC port that should be used in service runner mode in case no port is specified in the source config_yaml.
str	pytest_userver.plugins.grpc.client.grpc_service_endpoint (service_config)
	Returns the gRPC endpoint of the service.
float	pytest_userver.plugins.grpc.client.grpc_service_timeout (pytestconfig)
	Returns the gRPC timeout for the service that is set by the command line option --service-timeout.
	pytest_userver.plugins.grpc.client.grpc_channel (service_client, grpc_service_endpoint, grpc_service_timeout, grpc_session_channel, request)
	Returns the gRPC channel configured by the parameters from the grpc_service_endpoint fixture.
Generator[pathlib.Path, None, None]	pytest_userver.plugins.grpc.client.grpc_socket_path ()
	Path for the UNIX socket over which testsuite will talk to the gRPC service, if it chooses to use a UNIX socket.
	pytest_userver.plugins.grpc.client.userver_config_grpc_endpoint (pytestconfig, grpc_service_port_fallback, substitute_config_vars, request, choose_free_port)
	Returns a function that adjusts the static config for testsuite.
str	pytest_userver.plugins.grpc.mockserver.grpc_mockserver_endpoint (pytestconfig, get_free_port)
	Returns the gRPC endpoint to start the mocking server that is set by command line --grpc-mockserver-host and --grpc-mockserver-port options.
AsyncIterator[pytest_userver.grpc.MockserverSession]	pytest_userver.plugins.grpc.mockserver.grpc_mockserver_session (grpc_mockserver_endpoint)
	Returns the gRPC mocking server.
Iterator[pytest_userver.grpc.Mockserver]	pytest_userver.plugins.grpc.mockserver.grpc_mockserver (grpc_mockserver_session, asyncexc_append)
	Returns the gRPC mocking server.
	pytest_userver.plugins.grpc.mockserver.userver_config_grpc_mockserver ()
	Returns a function that adjusts the static config for testsuite.
list[str]	pytest_userver.plugins.kafka.kafka_components ()
	Should contain manually listed names of kafka producer and consumer components.
dict	pytest_userver.plugins.kafka.kafka_secdist (_bootstrap_servers, kafka_components)
	Automatically generates secdist config from user static config.
logcapture.CaptureServer	pytest_userver.plugins.log_capture.userver_log_capture (pytestconfig, userver_log_level)
	Session-scoped TCP server that receives service logs in TSKV format.
	pytest_userver.plugins.mongo.userver_mongo_config (mongo_connection_info)
	Returns a function that adjusts the static configuration file for the testsuite.
	pytest_userver.plugins.postgresql.userver_pg_config (pgsql_local)
	Returns a function that adjusts the static configuration file for the testsuite.
sql.RegisteredTrx	pytest_userver.plugins.postgresql.userver_pg_trx (testpoint)
	The fixture maintains transaction fault injection state using RegisteredTrx class.
RegisteredNtrx	pytest_userver.plugins.postgresql.userver_pg_ntrx (testpoint)
	The fixture maintains single query fault injection state using RegisteredNtrx class.
ConnectionInfo	pytest_userver.plugins.scylla.scylla_connection_info (pytestconfig)
	Where the ScyllaDB cluster is expected to be running.
float	pytest_userver.plugins.scylla.scylla_wait_timeout (pytestconfig)
ConnectionInfo	pytest_userver.plugins.scylla.scylla (_scylla_tcp_ready, scylla_connection_info)
	pytest_userver.plugins.scylla.userver_config_scylla (scylla_connection_info, _scylla_tcp_ready)
dict[str, str]	pytest_userver.plugins.service.service_env ()
	Override this to pass extra environment variables to the service.
str\|None	pytest_userver.plugins.service.service_http_ping_url (service_config, service_baseurl)
	Returns the service HTTP ping URL that is used by the testsuite to detect that the service is ready to work.
net.HealthChecks	pytest_userver.plugins.service.service_non_http_health_checks (service_config)
	Returns a health checks info.
float	pytest_userver.plugins.service.service_start_timeout ()
	Returns service start timeout in seconds.
	pytest_userver.plugins.service.service_binary_launcher ()
	If non-empty, defines a list of arguments starting with executable that is used instead of service_binary.
Callable[..., Awaitable[bool]]\|None	pytest_userver.plugins.service.service_health_check (service_http_ping_url, service_non_http_health_checks)
	Returns the health check function used by service_daemon_scope to detect that the service has started and is ready to accept requests.
	pytest_userver.plugins.service.service_daemon_scope (create_daemon_scope, daemon_scoped_mark, service_env, service_http_ping_url, service_config_path_temp, service_binary, service_binary_launcher, service_health_check, service_start_timeout)
	Prepares the start of the service daemon.
None	pytest_userver.plugins.service.extra_client_deps ()
	Service client dependencies hook.
None	pytest_userver.plugins.service.auto_client_deps (request)
	Ensures that the following fixtures, if available, are run before service start:
	pytest_userver.plugins.service.builtin_client_deps (testpoint, cleanup_userver_dumps, userver_log_capture, dynamic_config, mock_configs_service)
	Service client dependencies hook, like extra_client_deps.
	pytest_userver.plugins.service.service_daemon_instance (ensure_daemon_started, service_daemon_scope, builtin_client_deps, auto_client_deps, extra_client_deps)
	Calls ensure_daemon_started on service_daemon_scope to actually start the service.
dict[str, Any]\|None	pytest_userver.plugins.service.daemon_scoped_mark (request)
	Depend on this fixture directly or transitively to make your fixture a per-daemon fixture.
client.Client	pytest_userver.plugins.service_client.service_client (service_daemon_instance, service_baseurl, service_client_options, userver_service_client_options, userver_client_cleanup, client.TestsuiteClientConfig _testsuite_client_config)
	Main fixture that provides access to userver based service.
Callable[[client.Client], AsyncGenerator]	pytest_userver.plugins.service_client.userver_client_cleanup (request, service_logs_update_position, servicelogs_register_flusher, _dynamic_config_defaults_storage, _check_config_marks, dynamic_config)
	Contains the pre-test and post-test setup that depends on service_client.
	pytest_userver.plugins.service_client.websocket_client (service_client, service_port)
	Fixture that provides access to userver based websocket service.
client.ClientMonitor	pytest_userver.plugins.service_client.monitor_client (service_client, userver_monitor_client_options, monitor_baseurl)
	Main fixture that provides access to userver monitor listener.
str	pytest_userver.plugins.service_client.service_baseurl (service_port)
	Returns the main listener URL of the service.
str	pytest_userver.plugins.service_client.monitor_baseurl (monitor_port)
	Returns the main monitor URL of the service.
	pytest_userver.plugins.sql_coverage.on_uncovered ()
	Called when the coverage is incomplete.
Coverage	pytest_userver.plugins.sql_coverage.sql_coverage (sql_files)
	Returns data about the current coverage of statements.
	pytest_userver.plugins.sql_coverage.sql_statement_hook (testpoint, sql_coverage)
	Hook that accepts requests from the testpoint with information on PostgreSQL statements coverage.
	pytest_userver.plugins.sql_coverage.yql_statement_hook (testpoint, sql_coverage)
	Hook that accepts requests from the testpoint with information on YDB statements coverage.
	pytest_userver.plugins.sqlite.userver_sqlite_config (tmp_path_factory)
	Returns a function that adjusts the SQLite static configuration file for the testsuite.
	pytest_userver.plugins.sqlite.sqlite_db (_list_dbpath_components, service_client, testpoint)
	Removes all the data from SQLite tables at the end of the test to guarantee tests isolation.
	pytest_userver.plugins.testpoint.userver_config_testpoint (mockserver_info)
	Returns a function that adjusts the static configuration file for the testsuite.
client.YdbClient	pytest_userver.plugins.ydb.ydbsupport.ydb (_ydb_client, _ydb_init)
	YDB client fixture.
pathlib.Path	pytest_userver.plugins.ydb.ydbsupport.ydb_migration_dir (service_source_dir)
	Directory with migration files.
pathlib.Path	pytest_userver.plugins.ydb.ydbsupport.goose_binary_path ()
	Path to 'goose' migration tool.
pathlib.Path	pytest_userver.plugins.ydb.ydbsupport.ydb_cli ()
	Path to YDB CLI executable.
sql.RegisteredTrx	pytest_userver.plugins.ydb.ydbsupport.userver_ydb_trx (testpoint)
	The fixture maintains transaction fault injection state using RegisteredTrx class.
	pytest_userver.plugins.ydb.ydbsupport.userver_config_ydb (ydb_service_settings)
	Returns a function that adjusts the static configuration file for testsuite.

Function Documentation

◆ allowed_url_prefixes()

list[str] allowed_url_prefixes	(	mockserver_info,
		mockserver_ssl_info,
		allowed_url_prefixes_extra )

Returns final list of allowed urls prefixes when running in testsuite.

For most of cases you DO NOT want to override this fixture. Use "allowed_url_prefixes_extra" instead.

Definition at line 521 of file config.py.

◆ allowed_url_prefixes_extra()

list[str] allowed_url_prefixes_extra ( )

By default, userver HTTP client is only allowed to talk to mockserver when running in testsuite.

This makes tests repeatable and encapsulated.

Override this fixture to whitelist some additional URLs. It is still strongly advised to only talk to localhost in tests.

Definition at line 507 of file config.py.

◆ auto_client_deps()

None auto_client_deps ( request )

Ensures that the following fixtures, if available, are run before service start:

pgsql
mongodb
clickhouse
rabbitmq
kafka (kafka_producer, kafka_consumer)
redis_store
mysql
ydb
scylla
grpc_mockserver

To add other dependencies prefer overriding the extra_client_deps fixture.

Definition at line 231 of file service.py.

◆ build_dir()

pathlib.Path build_dir ( pytestconfig )

Returns the build directory set by command line --build-dir option.

Override this fixture to change the way the build directory is detected by the testsuite.

Definition at line 69 of file base.py.

◆ builtin_client_deps()

builtin_client_deps	(	testpoint,
		cleanup_userver_dumps,
		userver_log_capture,
		dynamic_config,
		mock_configs_service )

Service client dependencies hook, like extra_client_deps.

Feel free to override globally in a more specific pytest plugin (one that comes after userver plugins), but make sure to depend on the original fixture:

@pytest.fixture(name='builtin_client_deps')
def _builtin_client_deps(builtin_client_deps, some_extra_fixtures):
    pass

Definition at line 289 of file service.py.

◆ cache_invalidation_state()

InvalidationState cache_invalidation_state ( )

A fixture for notifying the service of changes in cache data sources.

Intended to be used by other fixtures that represent those data sources, not by tests directly.

Definition at line 154 of file caches.py.

◆ choose_free_port()

Callable[[int \| None], int] choose_free_port	(	pytestconfig,
		get_free_port,
		_testsuite_socket_cleanup,
		_testsuite_default_af )

A function that chooses a free port based on the optional hint given in the parameter.

Definition at line 182 of file base.py.

◆ cleanup_userver_dumps()

cleanup_userver_dumps	(		userver_dumps_root,
			request )

To avoid leaking dumps between tests, cache_dump_dir must be cleaned after each test.

To observe the dumps, add a final time.sleep(1000000) to your test locally. The returned function may also be used to clean dumps manually as appropriate.

See also: userver::dump::Dumper

Definition at line 50 of file dumps.py.

◆ config_fallback_path()

pathlib.Path config_fallback_path ( pytestconfig )

Returns the path to dynamic config fallback file set by command line --config-fallback option.

Override this fixture to change the way path to dynamic config fallback is provided.

Definition at line 191 of file config.py.

◆ config_service_defaults()

dynconf.ConfigValuesDict config_service_defaults	(		config_fallback_path,
			dynamic_config_fallback_patch )

Fixture that returns default values for dynamic config.

You may override it in your local conftest.py or fixture:

@pytest.fixture(scope='session')
def config_service_defaults():
    with open('defaults.json') as fp:
        return json.load(fp)

Definition at line 156 of file dynamic_config.py.

◆ congestion_control_fake_mode()

bool congestion_control_fake_mode ( )

Returns congestion control fake-mode value.

Definition at line 140 of file base.py.

◆ daemon_scoped_mark()

dict[str, Any] | None daemon_scoped_mark ( request )

Depend on this fixture directly or transitively to make your fixture a per-daemon fixture.

Example:

@pytest.fixture(scope='session')
def users_cache_state(daemon_scoped_mark, ...):
    return UsersCacheState(users_list=[])

For tests marked with @pytest.mark.uservice_oneshot(...), the service will be restarted, and all the per-daemon fixtures will be recreated.

This fixture returns kwargs passed to the uservice_oneshot mark (which may be an empty dict). For normal tests, this fixture returns None.

Definition at line 340 of file service.py.

◆ db_dump_schema_path()

pathlib.Path db_dump_schema_path	(		service_binary,
			service_tmpdir )

Runs the service binary with --dump-db-schema argument, dumps the 0_db_schema.sql file with database schema and returns path to it.

Override this fixture to change the way to dump the database schema.

Definition at line 142 of file config.py.

◆ dynamic_config()

dynconf.DynamicConfig dynamic_config	(	request,
		search_path,
		object_substitute,
		cache_invalidation_state,
		_dynamic_config_defaults_storage,
		config_service_defaults,
		dynamic_config_changelog,
		_dynconf_load_json_cached,
		dynconf_cache_names )

Fixture that allows to control dynamic config values used by the service.

Example:

    dynamic_config.set(
        USERVER_LOG_REQUEST_HEADERS=True,
        USERVER_LOG_REQUEST_HEADERS_WHITELIST=headers_whitelist,
    )
    await service_client.update_server_state()

Example with kill switches:

    # Enable kill switches and set their values
    dynamic_config.set(
        FIRST_KILL_SWITCH=1,
        SECOND_KILL_SWITCH=2,
    )
    await service_client.update_server_state()
 
    # Disable kill switches
    dynamic_config.switch_to_static_default(
        'FIRST_KILL_SWITCH',
        'SECOND_KILL_SWITCH',
    )
    await service_client.update_server_state()
 
    # Enable kill switches without changing their values
    dynamic_config.switch_to_dynamic_value(
        'FIRST_KILL_SWITCH',
        'SECOND_KILL_SWITCH',
    )
    await service_client.update_server_state()

HTTP and gRPC client requests call update_server_state automatically before each request.

For main dynamic config documentation:

See also: dynamic_config_testsuite

◆ dynamic_config_fallback_patch()

dynconf.ConfigValuesDict dynamic_config_fallback_patch ( )

Override this fixture to replace some dynamic config values specifically for testsuite tests:

@pytest.fixture(scope='session')
def dynamic_config_fallback_patch():
    return {"MY_CONFIG_NAME": 42}

Definition at line 139 of file dynamic_config.py.

◆ extra_client_deps()

None extra_client_deps ( )

Service client dependencies hook.

Feel free to override, e.g.:

@pytest.fixture
def extra_client_deps(some_fixtures_to_wait_before_service_start):
    pass

Definition at line 216 of file service.py.

◆ goose_binary_path()

pathlib.Path goose_binary_path ( )

Path to 'goose' migration tool.

Override this fixture to change the way 'goose' binary is discovered.

Definition at line 174 of file ydbsupport.py.

◆ grpc_channel()

grpc_channel	(	service_client,
		grpc_service_endpoint,
		grpc_service_timeout,
		grpc_session_channel,
		request )

Returns the gRPC channel configured by the parameters from the grpc_service_endpoint fixture.

You can add interceptors to the channel by implementing the pytest_grpc_client_interceptors hook in your pytest plugin or initial (root) conftest.

Definition at line 90 of file client.py.

◆ grpc_mockserver()

Iterator[pytest_userver.grpc.Mockserver] grpc_mockserver	(		grpc_mockserver_session,
			asyncexc_append )

Returns the gRPC mocking server.

In order for gRPC clients in your service to work, mock handlers need to be installed for them using this fixture.

Example:

import samples.greeter_pb2 as greeter_protos
import samples.greeter_pb2_grpc as greeter_services
 

async def test_grpc_client_mock_say_hello(service_client, grpc_mockserver):
    @grpc_mockserver(greeter_services.GreeterServiceServicer.SayHello)
    async def mock_say_hello(request, context):
        return greeter_protos.GreetingResponse(
            greeting=f'Hello, {request.name} from mockserver!',
        )
 
    response = await service_client.post('/hello?case=say_hello', data='tests')
    assert response.status == 200
    assert 'text/plain' in response.headers['Content-Type']
    assert response.text == 'Hello, tests from mockserver!'
    assert mock_say_hello.times_called == 1

Alternatively, you can create a shorthand for mocking frequently-used services:

import pytest
 
import samples.greeter_pb2_grpc as greeter_services
 
pytest_plugins = ['pytest_userver.plugins.grpc']

@pytest.fixture
def greeter_mock(grpc_mockserver):
    return grpc_mockserver.mock_factory(greeter_services.GreeterServiceServicer)

    @greeter_mock('SayHello')
    async def mock_say_hello(mock_request, _mock_context):
        return greeter_protos.GreetingResponse(
            greeting=f'Hello, {mock_request.name} from mockserver!',
        )

Mocks are only active within tests after their respective handler functions are created, not between tests. If the service needs the mock during startup, add the fixture that defines your mock to extra_client_deps.

To return an error status instead of response, use context (see ServicerContext docs):

async def test_error_status(service_client, grpc_mockserver):
    @grpc_mockserver(greeter_services.GreeterServiceServicer.SayHello)
    async def mock_say_hello(request, context: grpc.aio.ServicerContext):
        # Don't forget the `await`!
        await context.abort(grpc.StatusCode.UNAVAILABLE, 'Greeter is down')
 
    with pytest.raises(pytest_userver.client.TestsuiteTaskFailed) as ex_info:
        await service_client.run_task('call-say-hello')
    ex_info.match("'samples.api.GreeterService/SayHello' failed: code=UNAVAILABLE, message='Greeter is down'")
    assert mock_say_hello.times_called == 2

To trigger special exceptions in the service's gRPC client, raise these mocked errors from the mock handler:

Definition at line 65 of file mockserver.py.

◆ grpc_mockserver_endpoint()

str grpc_mockserver_endpoint	(		pytestconfig,
			get_free_port )

Returns the gRPC endpoint to start the mocking server that is set by command line --grpc-mockserver-host and --grpc-mockserver-port options.

For port 0, picks some free port.

Override this fixture to customize the endpoint used by gRPC mockserver.

# Overriding userver fixture
@pytest.fixture(scope='session')
def grpc_mockserver_endpoint(grpc_client_port):
    return f'[::]:{grpc_client_port}'

Definition at line 27 of file mockserver.py.

◆ grpc_mockserver_session()

AsyncIterator[pytest_userver.grpc.MockserverSession] grpc_mockserver_session ( grpc_mockserver_endpoint )

Returns the gRPC mocking server.

Warning: This is a sharp knife, use with caution! For most use-cases, prefer grpc_mockserver instead.

Definition at line 48 of file mockserver.py.

◆ grpc_service_endpoint()

str grpc_service_endpoint ( service_config )

Returns the gRPC endpoint of the service.

Used by grpc_channel fixture.

By default, gets the actual gRPC endpoint from service_config. Override this fixture if you add gRPC server listening ports in a custom way.

Definition at line 47 of file client.py.

◆ grpc_service_port_fallback()

int grpc_service_port_fallback ( )

Returns the gRPC port that should be used in service runner mode in case no port is specified in the source config_yaml.

Definition at line 36 of file client.py.

◆ grpc_service_timeout()

float grpc_service_timeout ( pytestconfig )

Returns the gRPC timeout for the service that is set by the command line option --service-timeout.

Override this fixture to change the way the gRPC timeout is set.

Definition at line 70 of file client.py.

◆ grpc_socket_path()

Generator[pathlib.Path, None, None] grpc_socket_path ( )

Path for the UNIX socket over which testsuite will talk to the gRPC service, if it chooses to use a UNIX socket.

See also: pytest_userver.plugins.grpc.client.userver_config_grpc_endpoint "userver_config_grpc_endpoint"

Definition at line 130 of file client.py.

◆ kafka_components()

list[str] kafka_components ( )

Should contain manually listed names of kafka producer and consumer components.

Definition at line 36 of file __init__.py.

◆ kafka_secdist()

dict kafka_secdist	(		_bootstrap_servers,
			kafka_components )

Automatically generates secdist config from user static config.

_bootstrap_servers is testsuite's fixture that determines current bootstrap servers list depends on Kafka testsuite plugin's settings.

@pytest.fixture(scope='session')
def kafka_components() -> list[str]:
    return ['kafka-consumer', 'kafka-producer']
 
 
@pytest.fixture(scope='session')
def service_env(kafka_secdist) -> dict:
    """
    Note: kafka_secist fixture generates the secdist config
 
    Expected secdist format is:
    "kafka_settings": {
        "<kafka-component-name>": {
            "brokers": "<brokers comma-separated endpoint list>",
            "username": "SASL2 username (may be empty if use PLAINTEXT)",
            "password": "SASL2 password (may be empty if use PLAINTEXT)"
        }
    }
    """
 
    return {'SECDIST_CONFIG': json.dumps(kafka_secdist)}

Definition at line 48 of file __init__.py.

◆ mock_configs_service()

None mock_configs_service	(		mockserver,
		dynconf.DynamicConfig	dynamic_config,
		dynconf._Changelog	dynamic_config_changelog )

Adds a mockserver handler that forwards dynamic_config to service's dynamic-config-client component.

Definition at line 316 of file dynamic_config.py.

◆ monitor_baseurl()

str monitor_baseurl ( monitor_port )

Returns the main monitor URL of the service.

Override this fixture to change the main monitor URL that the testsuite uses for tests.

Definition at line 243 of file service_client.py.

◆ monitor_client()

client.ClientMonitor monitor_client	(	service_client,
		userver_monitor_client_options,
		monitor_baseurl )

Main fixture that provides access to userver monitor listener.

async def test_engine_metrics(service_client, monitor_client: pytest_userver.client.ClientMonitor):
    metric: pytest_userver.metrics.Metric = await monitor_client.single_metric(
        'engine.task-processors.tasks.finished.v2',
        labels={'task_processor': 'main-task-processor'},
    )
    assert metric.value > 0
    assert metric.labels == {'task_processor': 'main-task-processor'}
 
    metrics_dict: pytest_userver.metrics.MetricsSnapshot = await monitor_client.metrics(
        prefix='http.',
        labels={'http_path': '/ping'},
    )
 
    assert metrics_dict
    assert 'http.handler.cancelled-by-deadline' in metrics_dict
 
    assert (
        metrics_dict.value_at(
            'http.handler.in-flight',
            labels={
                'http_path': '/ping',
                'http_handler': 'handler-ping',
                'version': '2',
            },
        )
        == 0
    )

Definition at line 137 of file service_client.py.

◆ monitor_port()

int monitor_port	(	pytestconfig,
		_original_service_config,
		choose_free_port )

Returns the monitor listener port number of the service set by command line --monitor-port option.

If no port is specified in the command line option, keeps the original port specified in the static config.

Override this fixture to change the way the monitor listener port number is detected by testsuite.

Definition at line 118 of file base.py.

◆ on_uncovered()

on_uncovered ( )

Called when the coverage is incomplete.

Override this fixture to change the way uncovered statements are reported or to ignore some of the statements from coverage report.

See SQL coverage tests for more info.

Definition at line 19 of file sql_coverage.py.

◆ read_latest_dump()

read_latest_dump ( userver_dumps_root )

Read the latest dump produced by a specified dumper.

See also: userver::dump::Dumper

Definition at line 25 of file dumps.py.

◆ scylla()

ConnectionInfo scylla	(		_scylla_tcp_ready,
			scylla_connection_info )

Definition at line 111 of file scylla.py.

◆ scylla_connection_info()

ConnectionInfo scylla_connection_info ( pytestconfig )

Where the ScyllaDB cluster is expected to be running.

Definition at line 63 of file scylla.py.

◆ scylla_wait_timeout()

float scylla_wait_timeout ( pytestconfig )

Definition at line 81 of file scylla.py.

◆ service_baseurl()

str service_baseurl ( service_port )

Returns the main listener URL of the service.

Override this fixture to change the main listener URL that the testsuite uses for tests.

Definition at line 230 of file service_client.py.

◆ service_binary()

pathlib.Path service_binary ( pytestconfig )

Returns the path to service binary set by command line --service-binary option.

Override this fixture to change the way the path to service binary is detected by the testsuite.

Definition at line 82 of file base.py.

◆ service_binary_launcher()

service_binary_launcher ( )

If non-empty, defines a list of arguments starting with executable that is used instead of service_binary.

The final argument list consists of service_binary_launcher plus the common service cmdline.

Can be used for:

resource limiting (rlimit, cgroup)
security limiting (capabilities, LSM)
resource profiling (perf)
tracing (strace, ltrace)

See also: service_daemon_instance

Definition at line 113 of file service.py.

◆ service_client()

client.Client service_client	(		service_daemon_instance,
			service_baseurl,
			service_client_options,
			userver_service_client_options,
			userver_client_cleanup,
		client.TestsuiteClientConfig	_testsuite_client_config )

Main fixture that provides access to userver based service.

async def test_ping(service_client: pytest_userver.client.Client):
    response: testsuite.utils.http.ClientResponse = await service_client.get('/ping')
    assert response.status == 200

Definition at line 22 of file service_client.py.

◆ service_config()

dict service_config	(	service_config_yaml,
		service_config_vars,
		substitute_config_vars )

Returns the static config values after the USERVER_CONFIG_HOOKS were applied (if any) and with all the '$', environment and fallback variables substituted.

Definition at line 360 of file config.py.

◆ service_config_path()

pathlib.Path service_config_path	(		pytestconfig,
			service_binary )

Returns the path to service.yaml file set by command line --service-config option.

Override this fixture to change the way path to the static config is provided.

Definition at line 123 of file config.py.

◆ service_config_path_temp()

pathlib.Path service_config_path_temp	(	service_tmpdir,
		service_config,
		service_config_yaml,
		service_config_vars )

Dumps the contents of the service_config_yaml and service_config_vars into a static config for testsuite and returns the path to the config file.

Definition at line 220 of file config.py.

◆ service_config_vars()

dict service_config_vars ( _service_config_hooked )

Returns the static config variables (config_vars.yaml) values after the USERVER_CONFIG_HOOKS were applied (if any).

Prefer using pytest_userver.plugins.config.service_config

Definition at line 271 of file config.py.

◆ service_config_vars_path()

pathlib.Path | None service_config_vars_path ( pytestconfig )

Returns the path to config_vars.yaml file set by command line --service-config-vars option.

Override this fixture to change the way path to config_vars.yaml is provided.

Definition at line 163 of file config.py.

◆ service_config_yaml()

dict service_config_yaml ( _service_config_hooked )

Returns the static config values after the USERVER_CONFIG_HOOKS were applied (if any).

Prefer using pytest_userver.plugins.config.service_config

Definition at line 259 of file config.py.

◆ service_daemon_instance()

service_daemon_instance	(	ensure_daemon_started,
		service_daemon_scope,
		builtin_client_deps,
		auto_client_deps,
		extra_client_deps )

Calls ensure_daemon_started on service_daemon_scope to actually start the service.

Makes sure that all the dependencies are prepared before the service starts.

See also: extra_client_deps; auto_client_deps; builtin_client_deps

Definition at line 315 of file service.py.

◆ service_daemon_scope()

service_daemon_scope	(	create_daemon_scope,
		daemon_scoped_mark,
		service_env,
		service_http_ping_url,
		service_config_path_temp,
		service_binary,
		service_binary_launcher,
		service_health_check,
		service_start_timeout )

Prepares the start of the service daemon.

Configures the health checking via the service_health_check fixture.

See also: service_daemon_instance

Definition at line 184 of file service.py.

◆ service_env()

dict[str, str] service_env ( )

Override this to pass extra environment variables to the service.

@pytest.fixture(scope='session')
def service_env(redis_sentinels):
    secdist_config = {
        'redis_settings': {
            'taxi-tmp': {
                'password': '',
                'database_index': 0,
                'sentinels': redis_sentinels,
                'shards': [{'name': 'test_master0'}],
            },
        },
    }
 
    return {'SECDIST_CONFIG': json.dumps(secdist_config)}

Definition at line 52 of file service.py.

◆ service_health_check()

Callable[..., Awaitable[bool]] \| None service_health_check	(		service_http_ping_url,
			service_non_http_health_checks )

Returns the health check function used by service_daemon_scope to detect that the service has started and is ready to accept requests.

Returns None when service_http_ping_url is set, in which case create_daemon_scope uses the ping URL directly. Otherwise returns a checker based on the service_non_http_health_checks info.

Override this fixture to change the way the service readiness is detected.

Definition at line 132 of file service.py.

◆ service_http_ping_url()

str \| None service_http_ping_url	(		service_config,
			service_baseurl )

Returns the service HTTP ping URL that is used by the testsuite to detect that the service is ready to work.

Returns None if there's no such URL.

By default, attempts to find server.handlers.Ping component by "handler-ping" name in static config. Override this fixture to change the behavior.

Definition at line 63 of file service.py.

◆ service_non_http_health_checks()

net.HealthChecks service_non_http_health_checks ( service_config )

Returns a health checks info.

By default, returns pytest_userver.utils.net.get_health_checks_info().

Override this fixture to change the way testsuite detects the tested service being alive.

Definition at line 82 of file service.py.

◆ service_port()

int service_port	(	pytestconfig,
		_original_service_config,
		choose_free_port )

Returns the main listener port number of the service set by command line --service-port option.

If no port is specified in the command line option, keeps the original port specified in the static config.

Override this fixture to change the way the main listener port number is detected by the testsuite.

Definition at line 96 of file base.py.

◆ service_secdist_path()

pathlib.Path | None service_secdist_path ( pytestconfig )

Returns the path to secure_data.json file set by command line --service-secdist option.

Override this fixture to change the way path to secure_data.json is provided.

Definition at line 177 of file config.py.

◆ service_source_dir()

pathlib.Path service_source_dir ( pytestconfig )

Returns the path to the service source directory that is set by command line --service-source-dir option.

Override this fixture to change the way the path to the service source directory is detected by testsuite.

Definition at line 55 of file base.py.

◆ service_start_timeout()

float service_start_timeout ( )

Returns service start timeout in seconds.

Override this fixture to change the service start timeout.

Definition at line 100 of file service.py.

◆ service_tmpdir()

service_tmpdir	(		service_binary,
			tmp_path_factory )

Returns the path for temporary files.

The path is the same for the whole session and files are not removed (at least by this fixture) between tests.

Definition at line 205 of file config.py.

◆ sql_coverage()

Coverage sql_coverage ( sql_files )

Returns data about the current coverage of statements.

See SQL coverage tests for more info.

Definition at line 61 of file sql_coverage.py.

◆ sql_statement_hook()

sql_statement_hook	(		testpoint,
			sql_coverage )

Hook that accepts requests from the testpoint with information on PostgreSQL statements coverage.

See SQL coverage tests for more info.

Definition at line 73 of file sql_coverage.py.

◆ sqlite_db()

sqlite_db	(	_list_dbpath_components,
		service_client,
		testpoint )

Removes all the data from SQLite tables at the end of the test to guarantee tests isolation.

Definition at line 37 of file sqlite.py.

◆ substitute_config_vars()

Callable[[Any, dict], Any] substitute_config_vars ( service_env )

A function that takes config_yaml, config_vars and applies all substitutions just like the service would.

Useful when patching the service config. It's a good idea to pass a component's config instead of the whole config_yaml to avoid unnecessary work.

Warning: The returned YAML is a clone, mutating it will not modify the actual config while in a config hook!

Definition at line 329 of file config.py.

◆ userver_cache_control()

CacheControl userver_cache_control	(	_userver_cache_control_context,
		_userver_cache_fixtures,
		request )

Userver cache control handler.

To install per cache handler use USERVER_CACHE_CONTROL_HOOKS variable in your pytest plugin:

USERVER_CACHE_CONTROL_HOOKS = {
    'my-cache-name': 'my_cache_cc',
}
 
@pytest.fixture
def my_cache_cc(my_cache_context):
    def cache_control(request, state):
        new_state = my_cache_context.get_state()
        if state == new_state:
            # Cache is already up to date, no need to update
            request.exclude()
        else:
            # Request incremental update, if you cache supports it
            request.incremental()
        return new_state
    return cache_control

Definition at line 186 of file caches.py.

◆ userver_client_cleanup()

Callable[[client.Client], AsyncGenerator] userver_client_cleanup	(	request,
		service_logs_update_position,
		servicelogs_register_flusher,
		_dynamic_config_defaults_storage,
		_check_config_marks,
		dynamic_config )

Contains the pre-test and post-test setup that depends on service_client.

Feel free to override, but in that case make sure to call the original userver_client_cleanup fixture instance.

Definition at line 51 of file service_client.py.

◆ userver_config_deadlock_detector()

ServiceConfigPatch userver_config_deadlock_detector ( str userver_deadlock_detector_mode )

Returns a function that adjusts the static configuration file for testsuite.

Sets the deadlock_detector parameter of the coro_pool component to the value of userver_deadlock_detector_mode fixture.

Definition at line 770 of file config.py.

◆ userver_config_dynconf_cache()

userver_config_dynconf_cache ( service_tmpdir )

Returns a function that adjusts the static configuration file for the testsuite.

Sets dynamic-config.fs-cache-path to a file that is reset after the tests to avoid leaking dynamic config values between test sessions.

Definition at line 216 of file dynamic_config.py.

◆ userver_config_dynconf_fallback()

userver_config_dynconf_fallback ( config_service_defaults )

Returns a function that adjusts the static configuration file for the testsuite.

Removes dynamic-config.defaults-path. Updates dynamic-config.defaults with config_service_defaults.

Definition at line 244 of file dynamic_config.py.

◆ userver_config_dynconf_url()

userver_config_dynconf_url ( mockserver_info )

Returns a function that adjusts the static configuration file for the testsuite.

Sets the dynamic-config-client.config-url to the value of mockserver configs-service, so that the mock_configs_service fixture could work.

Definition at line 280 of file dynamic_config.py.

◆ userver_config_grpc_endpoint()

userver_config_grpc_endpoint	(	pytestconfig,
		grpc_service_port_fallback,
		substitute_config_vars,
		request,
		choose_free_port )

Returns a function that adjusts the static config for testsuite.

if the original service config specifies grpc-server.port, and that port is taken, then adjusts it to a free port;
if the original service config specifies grpc-server.unix-socket-path, then adjusts it to a tmp path (see grpc_socket_path);
in service runner mode, uses the original grpc port from config or grpc_service_port_fallback.

Override this fixture to change the way grpc-server endpoint config is patched for tests.

Definition at line 145 of file client.py.

◆ userver_config_grpc_mockserver()

userver_config_grpc_mockserver ( )

Returns a function that adjusts the static config for testsuite.

Finds grpc-client-middleware-pipeline in config_yaml and enables grpc-client-middleware-testsuite.

Definition at line 107 of file mockserver.py.

◆ userver_config_http_client()

ServiceConfigPatch userver_config_http_client ( allowed_url_prefixes )

Returns a function that adjusts the static configuration file for testsuite.

Sets increased timeout and limits allowed URLs for http-client-core component.

Definition at line 549 of file config.py.

◆ userver_config_http_server()

ServiceConfigPatch userver_config_http_server	(		service_port,
			monitor_port )

Returns a function that adjusts the static configuration file for testsuite.

Sets the server.listener.port to listen on service_port fixture value; sets the server.listener-monitor.port to listen on monitor_port fixture value.

Definition at line 481 of file config.py.

◆ userver_config_logging()

ServiceConfigPatch userver_config_logging	(		userver_log_level,
			_service_logfile_path )

Returns a function that adjusts the static configuration file for testsuite.

Sets the logging.loggers.default to log to @stderr with level set from --service-log-level pytest configuration option.

Definition at line 602 of file config.py.

◆ userver_config_logging_otlp()

ServiceConfigPatch userver_config_logging_otlp ( )

Returns a function that adjusts the static configuration file for testsuite.

Sets the otlp-logger.load-enabled to false to disable OTLP logging and leave the default file logger.

Definition at line 633 of file config.py.

◆ userver_config_scylla()

userver_config_scylla	(		scylla_connection_info,
			_scylla_tcp_ready )

Definition at line 117 of file scylla.py.

◆ userver_config_secdist()

ServiceConfigPatch userver_config_secdist ( service_secdist_path )

Returns a function that adjusts the static configuration file for testsuite.

Sets the default-secdist-provider.config to the value of service_secdist_path fixture.

Definition at line 712 of file config.py.

◆ userver_config_substitutions()

ServiceConfigPatch userver_config_substitutions ( _service_config_substitution_vars )

Replaces substitution vars in all strings within config_vars using string.Template.substitute.

Substitution vars can be used as a shorthand for writing a full-fledged config hook in many common cases.

Unlike normal config_vars, substitution vars can also apply to a part of a string. For example, for config_vars entry

frobnicator-url: $mockserver/frobnicator

a possible patching result is as follows:

frobnicator-url: http://127.0.0.1:1234/frobnicator

Currently, the following substitution vars are supported:

mockserver - mockserver url
grpc_mockserver - grpc mockserver endpoint

Definition at line 434 of file config.py.

◆ userver_config_testpoint()

userver_config_testpoint ( mockserver_info )

Returns a function that adjusts the static configuration file for the testsuite.

Sets the tests-control.skip-unregistered-testpoints to True.

Definition at line 51 of file testpoint.py.

◆ userver_config_testsuite()

ServiceConfigPatch userver_config_testsuite	(		pytestconfig,
			mockserver_info )

Returns a function that adjusts the static configuration file for testsuite.

Sets up testsuite-support component, which:

increases timeouts for userver drivers
disables periodic cache updates
enables testsuite tasks

Sets the testsuite-enabled in config_vars.yaml to True; sets the tests-control.testpoint-url to mockserver URL.

Definition at line 651 of file config.py.

◆ userver_config_ydb()

userver_config_ydb ( ydb_service_settings )

Returns a function that adjusts the static configuration file for testsuite.

For all ydb.databases, sets endpoint and database to the local test YDB instance.

Definition at line 338 of file ydbsupport.py.

◆ userver_deadlock_detector_mode()

str userver_deadlock_detector_mode ( )

Returns Deadlock detector mode for testsuite.

Override this fixture to modify the deadlock detector settings. By default, it operates in detect-only mode. For a full list of available options, refer to the coro_pool.deadlock_detector parameter in the components.ManagerControllerComponent.

Definition at line 788 of file config.py.

◆ userver_default_log_level()

str userver_default_log_level ( )

Default log level to use in userver if no command line option was provided.

Returns 'debug'.

Definition at line 577 of file config.py.

◆ userver_dumps_root()

pathlib.Path userver_dumps_root ( tmp_path_factory )

The directory which the service will use for cache dumps.

Dumps of individual components will be stored in {userver_dumps_root}/{component-name}/{datetime}-v{format-version}

See also: userver::dump::Dumper

Definition at line 11 of file dumps.py.

◆ userver_log_capture()

logcapture.CaptureServer userver_log_capture	(		pytestconfig,
			userver_log_level )

Session-scoped TCP server that receives service logs in TSKV format.

The server is started once per testsuite session. Static config is patched so that the service can send logs when log capture is enabled for a test.

In most tests use service_client.capture_logs() instead of this fixture directly. See testsuite_logs_capture.

Example — filter captured logs after a request:

async def test_select(service_client: pytest_userver.client.Client):
    async with service_client.capture_logs(log_level='INFO') as capture:
        response: testsuite.utils.http.ClientResponse = await service_client.get('/logcapture')
        assert response.status == 200
 
    records = capture.select(
        text='Message to capture',
        link=response.headers['x-yarequestid'],
    )
    assert len(records) == 1, capture.select()

Example — subscribe to log events as they arrive:

async def test_subscribe(service_client, mockserver):
    async with service_client.capture_logs() as capture:
 
        @capture.subscribe(
            text='Message to capture',
            trace_id=mockserver.trace_id,
        )
        def log_event(link, **other):
            pass
 
        response = await service_client.get(
            '/logcapture',
            headers={'x-yatraceid': mockserver.trace_id},
        )
        assert response.status == 200
 
        call = await log_event.wait_call()
        assert call['link'] == response.headers['x-yarequestid']

Definition at line 37 of file log_capture.py.

◆ userver_log_level()

str userver_log_level	(		pytestconfig,
			userver_default_log_level )

Returns –service-log-level value if provided, otherwise returns userver_default_log_level() value from fixture.

Definition at line 589 of file config.py.

◆ userver_mongo_config()

userver_mongo_config ( mongo_connection_info )

Returns a function that adjusts the static configuration file for the testsuite.

Sets the dbconnection to the testsuite started MongoDB credentials if the dbconnection starts with mongodb://. Additionally increases MongoDB connection timeouts to 30 seconds.

Definition at line 20 of file mongo.py.

◆ userver_pg_config()

userver_pg_config ( pgsql_local )

Returns a function that adjusts the static configuration file for the testsuite.

Sets the dbconnection to the testsuite started PostgreSQL credentials if there's only one dbconnection in static config.

Definition at line 72 of file postgresql.py.

◆ userver_pg_ntrx()

RegisteredNtrx userver_pg_ntrx ( testpoint )

The fixture maintains single query fault injection state using RegisteredNtrx class.

See also: pytest_userver.plugins.postgresql.RegisteredNtrx

@pytest.mark.skip('This test is broken. Fix it, please')
async def test_ntrx_fail(service_client, pgsql, userver_pg_ntrx):
    response = await service_client.delete('/v1/key-value?key=foo')
    assert response.status == 200
 
    response = await service_client.post('/v1/key-value?key=foo&value=bar')
    assert response.status == 201
 
    with userver_pg_ntrx.mock_failure('sample_select_value'):
        response = await service_client.get('/v1/key-value?key=foo')
        assert response.status == 500
 
    response = await service_client.get('/v1/key-value?key=foo')
    assert response.status == 200
    assert response.content == b'bar'
 

Definition at line 148 of file postgresql.py.

◆ userver_pg_trx()

sql.RegisteredTrx userver_pg_trx ( testpoint )

The fixture maintains transaction fault injection state using RegisteredTrx class.

See also: pytest_userver.sql.RegisteredTrx

@pytest.mark.skip('This test is broken. Fix it, please')
async def test_trx_fail(service_client, pgsql, userver_pg_trx):
    response = await service_client.delete('/v1/key-value?key=foo')
    assert response.status == 200
 
    userver_pg_trx.enable_failure('sample_transaction_insert_key_value')
 
    response = await service_client.post('/v1/key-value?key=foo&value=bar')
    assert response.status == 500
 
    response = await service_client.get('/v1/key-value?key=foo')
    assert response.status == 404

Definition at line 125 of file postgresql.py.

◆ userver_sqlite_config()

userver_sqlite_config ( tmp_path_factory )

Returns a function that adjusts the SQLite static configuration file for the testsuite.

Sets the db-path to be an in memory DB with shared cache.

Definition at line 9 of file sqlite.py.

◆ userver_ydb_trx()

sql.RegisteredTrx userver_ydb_trx ( testpoint )

The fixture maintains transaction fault injection state using RegisteredTrx class.

See also: pytest_userver.sql.RegisteredTrx

@pytest.mark.skip('This test is broken. Fix it, please')
async def test_trx_fail(service_client, pgsql, userver_pg_trx):
    response = await service_client.delete('/v1/key-value?key=foo')
    assert response.status == 200
 
    userver_pg_trx.enable_failure('sample_transaction_insert_key_value')
 
    response = await service_client.post('/v1/key-value?key=foo&value=bar')
    assert response.status == 500
 
    response = await service_client.get('/v1/key-value?key=foo')
    assert response.status == 404

Definition at line 315 of file ydbsupport.py.

◆ websocket_client()

websocket_client	(		service_client,
			service_port )

Fixture that provides access to userver based websocket service.

Usage example:

async def test_echo(websocket_client):
    async with websocket_client.get('chat') as chat:
        await chat.send('hello')
        response = await chat.recv()
        assert response == 'hello'

You can pass extra kwargs to get, they will be forwarded to websockets.connect.

Definition at line 103 of file service_client.py.

◆ ydb()

client.YdbClient ydb	(		_ydb_client,
			_ydb_init )

YDB client fixture.

Definition at line 27 of file ydbsupport.py.

◆ ydb_cli()

pathlib.Path ydb_cli ( )

Path to YDB CLI executable.

Override this fixture to change the way YDB CLI is discovered.

Definition at line 226 of file ydbsupport.py.

◆ ydb_migration_dir()

pathlib.Path ydb_migration_dir ( service_source_dir )

Directory with migration files.

Definition at line 136 of file ydbsupport.py.

◆ yql_statement_hook()

yql_statement_hook	(		testpoint,
			sql_coverage )

Hook that accepts requests from the testpoint with information on YDB statements coverage.

See SQL coverage tests for more info.

Definition at line 90 of file sql_coverage.py.

Detailed Description

Namespaces

Functions

Function Documentation

◆ allowed_url_prefixes()

◆ allowed_url_prefixes_extra()

◆ auto_client_deps()

◆ build_dir()

◆ builtin_client_deps()

◆ cache_invalidation_state()

◆ choose_free_port()

◆ cleanup_userver_dumps()

◆ config_fallback_path()

◆ config_service_defaults()

◆ congestion_control_fake_mode()

◆ daemon_scoped_mark()

◆ db_dump_schema_path()

◆ dynamic_config()

◆ dynamic_config_fallback_patch()

◆ extra_client_deps()

◆ goose_binary_path()

◆ grpc_channel()

◆ grpc_mockserver()

◆ grpc_mockserver_endpoint()

◆ grpc_mockserver_session()

◆ grpc_service_endpoint()

◆ grpc_service_port_fallback()

◆ grpc_service_timeout()

◆ grpc_socket_path()

◆ kafka_components()

◆ kafka_secdist()

◆ mock_configs_service()

◆ monitor_baseurl()

◆ monitor_client()

◆ monitor_port()

◆ on_uncovered()

◆ read_latest_dump()

◆ scylla()

◆ scylla_connection_info()

◆ scylla_wait_timeout()

◆ service_baseurl()

◆ service_binary()

◆ service_binary_launcher()

◆ service_client()

◆ service_config()

◆ service_config_path()

◆ service_config_path_temp()

◆ service_config_vars()

◆ service_config_vars_path()

◆ service_config_yaml()

◆ service_daemon_instance()

◆ service_daemon_scope()

◆ service_env()

◆ service_health_check()

◆ service_http_ping_url()

◆ service_non_http_health_checks()

◆ service_port()

◆ service_secdist_path()

◆ service_source_dir()

◆ service_start_timeout()

◆ service_tmpdir()

◆ sql_coverage()

◆ sql_statement_hook()

◆ sqlite_db()

◆ substitute_config_vars()

◆ userver_cache_control()

◆ userver_client_cleanup()

◆ userver_config_deadlock_detector()

◆ userver_config_dynconf_cache()

◆ userver_config_dynconf_fallback()

◆ userver_config_dynconf_url()

◆ userver_config_grpc_endpoint()

◆ userver_config_grpc_mockserver()

◆ userver_config_http_client()

◆ userver_config_http_server()

◆ userver_config_logging()

◆ userver_config_logging_otlp()

◆ userver_config_scylla()

◆ userver_config_secdist()

◆ userver_config_substitutions()