Github   Telegram
No Matches
Functional service tests (testsuite)

Getting started

Userver has built-in support for functional service tests using Yandex.Taxi Testsuite. Testsuite is based on pytest and allows developers to test thier services in isolated environment. It starts service binary with minimal database and all external services mocked then allows developer to call service handlers and test their result.

Supported features:

  • Database startup (Mongo, Postgresql, Clickhouse, ...)
  • Per-test database state
  • Service startup
  • Mocksever to mock external service handlers
  • Mock service time, utils::datetime::Now()
  • Testpoint
  • Cache invalidation
  • Logs capture
  • Service runner

CMake integration

With userver_testsuite_add() function you can easily add testsuite support to your project. Its main purpose is:

  • Setup Python environment virtualenv or use an existing one.
  • Create runner script that setups PYTHONPATH and passes extra arguments to pytest.
  • Registers ctest target.
  • Adds a start-* target that starts the service and databases with testsuite configs and waits for keyboard interruption to stop the service.

cmake/UserverTestsuite.cmake library is automatically addded to CMake path after userever enviroment setup. Add the following line to use it:


Then create testsuite target:



  • SERVICE_TARGET, required CMake name of the target service to test. Used as suffix for testsuite- and start- CMake target names.
  • WORKING_DIRECTORY, pytest working directory. Default is ${CMAKE_CURRENT_SOURCE_DIR}.
  • PYTEST_ARGS, list of extra arguments passed to pytest.
  • PYTHONPATH, list of directories to be prepended to PYTHONPATH.
  • REQUIREMENTS, list of reqirements.txt files used to populate virtualenv.
  • PYTHON_BINARY, path to existing Python binary.
  • VIRTUALENV_ARGS, list of extra arguments passed to virtualenv.

Python environment

You may want to create new virtual environment with its own set of packages. Or reuse existing one. That could be done this way:

  • If REQUIREMENTS is given then new virtualenv is created
  • Else if PYTHON_BINARY is specified it's used
  • Otherwise value of ${TESTSUITE_VENV_PYTHON} is used

Basic requirements.txt file may look like this:


Creating per-testsuite virtual environment is a recommended way to go. It creates virtualenv that could be found in current binary directory:


Run with ctest

userver_testsuite_add() registers a ctest target with name testsuite-${SERVICE_TARGET}. Run all project tests with ctest command or use filters to run specific tests:

ctest -V -R testsuite-my-project # SERVICE_TARGET argument is used

Direct run

To run pytest directly userver_testsuite_add() creates a testsuite runner script that could be found in corresponding binary directory. This may be useful to run a single testcase, to start the testsuite with gdb or to start the testsuite with extra pytest arguments:


You can use it to manually start testsuite with extra pytest arguments, e.g.:

./build/tests/runtests-testsuite-my-project -vvx ./tests -k test_foo

Please refer to testuite and pytest documentation for available options. Run it with --help argument to see the short options description.

./build/tests/runtests-testsuite-my-project ./tests --help


By default internal pytest_userver plugin is included in python path. It provides basic testsuite support for userver service. To use it add it to your pytest_plugins in root

pytest_plugins = ['pytest_userver.plugins']

It requires extra PYTEST_ARGS to be passed:


Userver testsuite support

Userver has built-in support for testsuite.

In order to use it you need to register correspoding components:


Add components to components list:


Add testsuite components to config.yaml:

load-enabled: $testsuite-enabled
path: /tests/{action}
method: POST
task_processor: main-task-processor
testpoint-timeout: 10s
testpoint-url: mockserver/testpoint
throttling_enabled: false
Please note that the testsuite support must be disabled in production enviroment. Testsuite sets the testsuite-enabled variable into true when runs the service. In the example above this variable controls whether or tests-control component is loaded.


Service config generation

pytest_userver uses config.yaml and config_vars.yaml passed to pytest to generate services configs. pytest_userver provides a way to modify this configs before startings service. You can decalre USERVER_CONFIG_HOOKS variable in your pytest-plugin it is list of functions or pytest-fixtures that are run before config is written to disk. Example usage:

USERVER_CONFIG_HOOKS = ['userver_config_configs_client']
def userver_config_configs_client(mockserver_info):
def do_patch(config_yaml, config_vars):
components = config_yaml['components_manager']['components']
] = mockserver_info.base_url.rstrip('/')
return do_patch

Service client

Fixture service_client is used to access service being tested:

async def test_ping(service_client):
response = await service_client.get('/ping')
assert response.status == 200

When tests-control component is enabled service_client is pytest_userver.client.Client instance. Which supports special testsuite related methods.

On first call to service_client service state is implicitly updated, e.g.: caches, mocked time, etc.

Service environment variables

Use this fixture to provide extra environment variables for your service:

def service_env():
return {'SECDIST_CONFIG': json.dumps(SECDIST_CONFIG)}

Extra client dependencies

Use this fixture to provide extra fixtures that your service depends on:

def client_deps(pgsql):


Mockserver allows to mock external HTTP handlers. It starts its own HTTP server that receives HTTP traffic from service being tested. And allows to install custom HTTP handlers within testsuite. In order to use it all HTTP clients must be pointed to mockserver address.

Mockserver usage example:

def mock_translations(mockserver, translations, mocked_time):
def mock(request):
return {
'content': translations,
'update_time': utils.timestring(,
return mock

Mock time

Userver provides a way to mock internal datetime value. It only works for datetime retrieved with utils::datetime::Now(), see Mocked time section for details.

From testsuite you can control it with mocked_time plugin.

Example usage:'2019-12-31T11:22:33Z')
async def test_now(service_client, mocked_time):
response = await service_client.get('/now')
assert response.status == 200
assert response.json() == {'now': '2019-12-31T11:22:33+00:00'}
# Change mocked time and sync state
await service_client.update_server_state()
response = await service_client.get('/now')
assert response.status == 200
assert response.json() == {'now': '2019-12-31T11:33:44+00:00'}

Example are available here:


Testpoints are used to send messages from the service to testcase and back. Typical use cases are:

  • Retrieve intermediate state of the service and test it
  • Inject errors into the service
  • Synchronize service and testcase execution

First of all you should include testpoint header:

It provides TESTPOINT() and family of TESTPOINT_CALLBACK() macroses that do nothing in production environment and only work when run under testsuite. Under testsuite they only make sense when corresponding testsuite handler is installed.

All testpoints has their own name that is used to call named testsuite handler. Second argument is formats::json::Value() instance that is only evaluated under testsuite.

TESTPOINT() usage sample:

TESTPOINT("simple-testpoint", [] {
builder["payload"] = "Hello, world!";
return builder.ExtractValue();

Then you can use testpoint from testcase:

async def test_basic(service_client, testpoint):
def simple_testpoint(data):
assert data == {'payload': 'Hello, world!'}
response = await service_client.get('/testpoint')
assert response.status == 200
assert simple_testpoint.times_called == 1

In order to eliminate unnecessary testpoint requests userver keeps track of testpoints that have testsuite handlers installed. Usually testpoint handlers are declared before first call to service_client which implicitly updates userver's list of testpoint. Sometimes it might be required to manually update server state. This can be achieved using service_client.update_server_state() method e.g.:

def injection_point(data):
return {'value': 'injected'}
await service_client.update_server_state()
assert injection_point.times_called == 0

Accessing testpoint userver is not aware of will raise an exception:

with pytest.raises(
assert injection_point.times_called == 0

Logs capture

Testsuite can be used to test logs written by service. To achieve this the testsuite starts a simple logs capture TCP server and tells the service to replicate logs to it on per test basis.

Example usage:

async def test_select(service_client):
async with service_client.capture_logs() as capture:
response = await service_client.get('/logcapture')
assert response.status == 200
records =
text='Message to catpure', link=response.headers['x-yarequestid'],
assert len(records) == 1

Example on logs capture usage could be found here:

Testsuite tasks

Testsuite tasks facility allows to register a custom function and call it by name from testsuite. It's useful for testing components that perform periodic job not related to its own HTTP handler.

You can use testsuite::TestsuiteTasks to register your own task:

auto& testsuite_tasks = userver::testsuite::GetTestsuiteTasks(context);
// Only register task for testsuite environment
if (testsuite_tasks.IsEnabled()) {
testsuite_tasks.RegisterTask("sample-task", [] {
TESTPOINT("sample-task/action", [] {
userver::formats::json::ValueBuilder builder;
return builder.ExtractValue();
} else {
// Proudction code goes here

After that you can call your task from testsuite code:

await service_client.run_task('sample-task')
assert task_action.times_called == 1

Or spawn the task asynchronously using context manager:

async with service_client.spawn_task('sample-task'):
await task_action.wait_call()

An example on testsuite tasks could be found here:


Testsuite provides access to userver metrics, see tutorial on configuration. It allows to:

  • retrieve service metrics with await monitor_client.get_metrics()
  • reset metrics using await service_client.reset_metrics()

Example usage:

async def test_reset(service_client, monitor_client):
# Reset service metrics
await service_client.reset_metrics()
# Retrieve metrics
metrics = await monitor_client.get_metrics()
assert metrics['sample-metrics']['foo'] == 0

Service runner

Testsuite provides a way to start standalone service with all mocks and database started. This can be done by adding --service-runner-mode flag to pytest, e.g.:

./build/tests/runtests-my-project ./tests -s --service-runner-mode

Please note that -s flag is required to disable console output capture.

pytest_userver provides default service runner testcase. In order to override it you have to add your own testcase with @pytest.mark.servicetest:

def test_service(service_client):