Yandex.Taxi Testsuite

https://github.com/yandex/yandex-taxi-testsuite
Admin
testsuite: microservices testing framework
Tokens:13,988
Snippets:110
Trust Score:8.5
License:MIT
Update:1 year ago
Show doc for...
Tokens:
Raw
### Python Example Web Service

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-postgres/README.rst

This snippet shows a basic aiohttp web application that handles GET requests, demonstrating a simple 'Hello, World' type service.

```Python
import asyncio
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

app = web.Application()
app.router.add_get('/', handle)
app.router.add_get('/{name}', handle)

if __name__ == '__main__':
    web.run_app(app, port=8080)
```

--------------------------------

### Example Service Implementation (server.py)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mysql/README.rst

This Python file demonstrates a basic aiohttp web service that could serve as the chat storage backend. It defines a simple GET endpoint.

```Python
import aiohttp.web

async def handle(request):
    return aiohttp.web.Response(text="Hello, world")

app = aiohttp.web.Application()
app.router.add_get('/', handle)

if __name__ == '__main__':
    aiohttp.web.run_app(app)
```

--------------------------------

### Example Web Service Implementation (Python)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mongo/README.rst

This Python code defines a simple web application using aiohttp, handling basic GET requests. It serves as the core service for the chat storage backend example, demonstrating how to set up routes and handle requests.

```Python
# server.py
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

app = web.Application()
app.router.add_get('/', handle)
app.router.add_get('/{name}', handle)

if __name__ == '__main__':
    web.run_app(app, port=8080)
```

--------------------------------

### Pytest Service Integration Tests

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-postgres/README.rst

This snippet contains integration tests for the example aiohttp service, verifying its GET endpoints and response content using the aiohttp test client.

```Python
async def test_hello(cli):
    resp = await cli.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert 'Hello, Anonymous' in text

async def test_hello_name(cli):
    resp = await cli.get('/World')
    assert resp.status == 200
    text = await resp.text()
    assert 'Hello, World' in text
```

--------------------------------

### Run tests for example services

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Commands to navigate to the examples directory and run tests using `make`.

```sh
cd docs/examples && make
```

--------------------------------

### Run all testsuite examples

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/README.rst

Commands to execute all integration tests for the chat application examples, both locally using system pytest and within Docker containers for a consistent environment.

```Shell
make runtests
```

```Shell
make docker-runtests
```

--------------------------------

### Pytest Configuration and Fixtures (Python)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mongo/README.rst

This conftest.py file provides essential pytest fixtures for testing the aiohttp web application. It includes a client fixture for making requests to the app and a base test case class, simplifying test setup and execution.

```Python
# tests/conftest.py
import pytest
from aiohttp.test_utils import TestClient, AioHTTPTestCase
from server import app

@pytest.fixture
async def cli(aiohttp_client):
    return await aiohttp_client(app)

class MyTestCase(AioHTTPTestCase):
    async def get_application(self):
        return app
```

--------------------------------

### Pytest Conftest Fixtures (tests/conftest.py)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mysql/README.rst

This conftest.py file provides pytest fixtures for testing the example service. It sets up an aiohttp test client using the application defined in server.py.

```Python
import pytest

@pytest.fixture
def client(aiohttp_client):
    from server import app
    return aiohttp_client(app)
```

--------------------------------

### Service Test Case (tests/test_service.py)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mysql/README.rst

This Python file contains a pytest asynchronous test case for the example service. It uses the 'client' fixture to make a GET request and asserts the response status and content.

```Python
async def test_hello(client):
    resp = await client.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert 'Hello, world' in text
```

--------------------------------

### Python: Custom Data Setup for redis_store using pytest.mark.redis_store

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/redis.rst

This example shows how to use the `pytest.mark.redis_store` decorator to pre-populate the `redis_store` fixture with specific data before a test runs. It supports providing a list of Redis commands or specifying a file to load data from.

```python
import pytest

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.redis.pytest_plugin',
]

@pytest.mark.redis_store(
    ['set', 'foo', 'bar'],
    ['hset', 'baz', 'quux', 'bat'],
)
def test_redis_marker_store(redis_store):
    assert redis_store.get('foo') == b'bar'
    assert redis_store.hgetall('baz') == {b'quux': b'bat'}

@pytest.mark.redis_store(file='use_redis_store_file')
def test_redis_store_file(redis_store):
    assert redis_store.get('foo') == b'store'
```

--------------------------------

### Service Integration Tests (Python)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-mongo/README.rst

This file contains example integration tests for the aiohttp web service, utilizing the pytest fixtures defined in conftest.py. It verifies the functionality of the 'hello world' and named greeting endpoints, ensuring the service behaves as expected.

```Python
# tests/test_service.py
async def test_hello_world(cli):
    resp = await cli.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert 'Hello, Anonymous' in text

async def test_hello_name(cli):
    resp = await cli.get('/testuser')
    assert resp.status == 200
    text = await resp.text()
    assert 'Hello, testuser' in text
```

--------------------------------

### Run specific testsuite examples

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/README.rst

Commands to execute integration tests for a particular component of the chat application (backend or a specific storage service), both locally and within Dockerized environments.

```Shell
make runtests-chat-backend
```

```Shell
make runtests-chat-storage-mongo
```

```Shell
make runtests-chat-storage-mysql
```

```Shell
make runtests-chat-storage-postgres
```

```Shell
make docker-runtests-chat-backend
```

```Shell
make docker-runtests-chat-storage-mongo
```

```Shell
make docker-runtests-chat-storage-mysql
```

```Shell
make docker-runtests-chat-storage-postgres
```

--------------------------------

### Install ClickHouse on Debian using apt

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/clickhouse.rst

This snippet provides the bash command to install `clickhouse-common-static` on Debian-based systems. It first updates the package list and then proceeds with the installation, as recommended for setting up ClickHouse.

```bash
sudo apt update && apt install clickhouse-common-static
```

--------------------------------

### Install Yandex Taxi Testsuite with pip

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Instructions for installing the Yandex Taxi Testsuite and its optional dependencies for various databases (MongoDB, PostgreSQL, Redis, MySQL, ClickHouse, RabbitMQ, Kafka) using pip.

```sh
pip3 install yandex-taxi-testsuite

# testsuite with mongodb support
pip3 install yandex-taxi-testsuite[mongodb]

# testsuite with postgresql support
pip3 install yandex-taxi-testsuite[postgresql]
pip3 install yandex-taxi-testsuite[postgresql-binary]

# testsuite with redis support
pip3 install yandex-taxi-testsuite[redis]

# testsuite with mysql support
pip3 install yandex-taxi-testsuite[mysql]

# testsuite with clickhouse support
pip3 install yandex-taxi-testsuite[clickhouse]

# testsuite with rabbitmq support
pip3 install yandex-taxi-testsuite[rabbitmq]

# testsuite with kafka support
pip3 install yandex-taxi-testsuite[kafka]
```

--------------------------------

### Run Pytest with Envinfo Plugin Example

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/other_plugins.rst

Example command line arguments for running pytest with verbose output and exit-on-first-fail, demonstrating how `envinfo` details might be collected. The `--envinfo-no-git` flag can be used to disable Git status collection if performance is an issue.

```shell
python3 -m pytest -vv -x
```

--------------------------------

### CaptureServer Class API Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicelogs.rst

API documentation for the `CaptureServer` class, outlining its constructor and methods for starting, waiting for clients, capturing logs, setting default log levels, and retrieving socket information.

```APIDOC
CaptureServer()
  members: __init__, start, wait_for_client, capture, default_log_level, getsocknames
```

--------------------------------

### Python Kafka Producer and Consumer Chain Example

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/kafka.rst

This example demonstrates a basic asynchronous workflow for sending a message to a Kafka topic using `kafka_producer` and then consuming it back using `kafka_consumer`, verifying the message's topic, key, and value.

```python
async def test_kafka_producer_consumer_chain(kafka_producer, kafka_consumer):
    TOPIC = 'Test-topic-chain'
    KEY = 'test-key'
    MESSAGE = 'test-message'

    await kafka_producer.send(TOPIC, KEY, MESSAGE)

    consumed_message = await kafka_consumer.receive_one([TOPIC])

    assert consumed_message.topic == TOPIC
    assert consumed_message.key == KEY
    assert consumed_message.value == MESSAGE
```

--------------------------------

### Run chat backend with various storage options using Docker Compose

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/README.rst

Commands to launch the chat backend service integrated with different database types (PostgreSQL, MongoDB, MySQL), orchestrated by Docker Compose. These commands set up the necessary containers for the application to run.

```Shell
make run-chat-storage-postgres
```

```Shell
make run-chat-storage-mongo
```

```Shell
make run-chat-storage-mysql
```

--------------------------------

### Expected console output after Docker application startup

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/README.rst

Illustrates the console output seen upon successful startup of the chat application services via Docker Compose, indicating the application is running and accessible at a specific URL.

```Shell
chat-postgres_1 | ======== Running on http://0.0.0.0:8081 ========
chat-postgres_1 | (Press CTRL+C to quit)
```

--------------------------------

### Pytest Conftest for Aiohttp Client

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/examples/chat-storage-postgres/README.rst

This snippet defines a pytest fixture in `conftest.py` to provide an aiohttp test client, simplifying HTTP requests in tests.

```Python
import pytest
from aiohttp.test_utils import TestClient, loop_context
from server import app

@pytest.fixture
async def cli(aiohttp_client):
    return await aiohttp_client(app)
```

--------------------------------

### Install pre-commit hooks for linters

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Command to install pre-commit hooks that automatically run ruff and other linters before committing changes.

```sh
make install-pre-commit-hooks
```

--------------------------------

### Python AsyncCallQueue Wrapper Example

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/utils.rst

Demonstrates how to use the `acallqueue` decorator to wrap an asynchronous function, allowing its calls to be tracked. It shows how to check if calls were made, the number of calls, and retrieve the arguments of the next call, which would be `{'a': 1, 'b': 2}` for the example call.

```python
@callinfo.acallqueue
async def func(a, b):
    return a + b

await func(1, 2)

func.has_calls
func.times_called
func.next_call()
```

--------------------------------

### Service Configuration Example for Mockserver Redirection

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Illustrates a typical URL format used to configure a service to point its HTTP calls to the mockserver instead of its original endpoint. This is crucial for isolating and testing service interactions.

```text
http://mockserver-address/service-name/path
```

--------------------------------

### Python Mockserver Handler Example

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Illustrates a basic Python handler function for the mockserver, showing how to access request body and method parameters.

```python
def handler(*, body_json, method):
    assert method == 'POST'
    assert body_json == {...}
    reutrn {...}
```

--------------------------------

### HTTP Request for Mockserver Error Emulation

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Shows an example HTTP GET request with the 'X-Testsuite-Supported-Errors' header. This header is used to signal to the mockserver which types of errors (network, timeout) the client can handle for emulation.

```text
GET /path HTTP/1.1
X-Testsuite-Supported-Errors: network,timeout
...
```

--------------------------------

### Python: Basic Usage of redis_cluster_store Fixture for Redis Cluster

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/redis.rst

This example illustrates how to utilize the `redis_cluster_store` fixture to connect and perform operations on a Redis cluster. The fixture provides an interface similar to `redis.RedisCluster()` for setting and retrieving data.

```python
import pytest

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.redis.pytest_plugin',
]


def test_redis_basic(redis_cluster_store):
    redis_cluster_store.set('somekey', 'somedata')
    assert redis_cluster_store.get('somekey') == b'somedata'
```

--------------------------------

### C++ Service-Side Testpoint Call

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/testpoint.rst

Example C++ code demonstrating how a service calls a testpoint. It first checks if testpoints are enabled, then prepares a JSON data payload, calls the testpoint with a specific name and data, and optionally processes any response received.

```cpp
Response View::Handle(Request&& request) {
  // Testpoint must be enabled for test run
  if (testpoint::IsEnabled()) {
    json::Value data;
    data["foo"] = "bar";
    const auto response = testpoint::Call("foo", data);
    if (respone) {
       // process response if required
    }
  }
}
```

--------------------------------

### Python: Custom Data Setup for redis_cluster_store using pytest.mark.redis_cluster_store

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/redis.rst

This snippet demonstrates how to use the `pytest.mark.redis_cluster_store` decorator to initialize the `redis_cluster_store` fixture with custom data for a test. It allows specifying Redis commands directly or loading data from a file.

```python
import pytest

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.redis.pytest_plugin',
]

@pytest.mark.redis_cluster_store(
    ['set', 'foo', 'bar'],
    ['hset', 'baz', 'quux', 'bat'],
)
def test_redis_marker_store(redis_cluster_store):
    assert redis_cluster_store.get('foo') == b'bar'
    assert redis_cluster_store.hgetall('baz') == {b'quux': b'bat'}

@pytest.mark.redis_cluster_store(file='use_redis_store_file')
def test_redis_store_file(redis_cluster_store):
    assert redis_cluster_store.get('foo') == b'store'
```

--------------------------------

### API Reference: Testsuite Service Client Classes

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicerunner.rst

Documentation for client classes within `testsuite.daemons.service_client`. This includes `Client` and `AiohttpClient`, detailing their constructors (`__init__`) and various HTTP methods (`post`, `put`, `patch`, `get`, `delete`, `options`, `request`) for interacting with services.

```APIDOC
Module: testsuite.daemons.service_client

Classes:
  Client():
    Members:
      __init__
      post
      put
      patch
      get
      delete
      options
      request

  AiohttpClient():
    Members:
      __init__
      post
      put
      patch
      get
      delete
      options
      request
```

--------------------------------

### Testpoint HTTP Interface Request Example

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/testpoint.rst

Example of the HTTP POST request to the mockserver's `/testpoint` endpoint. This interface is used by services to interact with testpoints, sending a JSON payload containing the testpoint name and associated data.

```APIDOC
POST /testpoint HTTP/1.0
Host: mockerver
Content-Type: appication/json

{"name": "foo", "data": {"foo": "bar"}}
```

--------------------------------

### APIDOC: testsuite.databases.mysql.pytest_plugin.mysql_local Fixture

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

API documentation for the `mysql_local` pytest fixture provided by `testsuite`, typically used for local MySQL schema discovery and setup.

```APIDOC
testsuite.databases.mysql.pytest_plugin.mysql_local()
```

--------------------------------

### Python: Basic Usage of redis_store Fixture for Redis with Sentinels

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/redis.rst

This snippet demonstrates how to use the `redis_store` fixture provided by the testsuite to interact with a non-cluster Redis instance configured with sentinels. It shows basic `set` and `get` operations, mimicking the `redis.StrictRedis()` interface.

```python
import pytest

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.redis.pytest_plugin',
]


def test_redis_basic(redis_store):
    redis_store.set('somekey', 'somedata')
    assert redis_store.get('somekey') == b'somedata'
```

--------------------------------

### Python: Access MySQL Database in testsuite

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

This example illustrates how to interact with a MySQL database instance provided by `testsuite`'s `mysql` fixture. It shows how to obtain a cursor for a specific database (e.g., 'chat_messages'), execute a parameterized SQL query, and fetch a single record.

```python
def test_read_from_database(mysql):
    cursor = mysql['chat_messages'].cursor()
    cursor.execute(
        'SELECT username, text FROM messages WHERE id = %s', (data['id'],),
    )
    record = cursor.fetchone()
    assert record == ('foo', 'bar')
```

--------------------------------

### Python Testcase Testpoint Handler

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/testpoint.rst

Example Python testcase code showing how to define a testpoint handler using the `testpoint` fixture. This snippet demonstrates setting up an asynchronous handler for a specific testpoint name ('foo') and asserting its invocation count after a client interaction.

```python
async def test_handler(client, testpoint):
    @testpoint('foo')
    def foo_handler(data):
        pass

    response = await client.post(...)
    assert response.status_code == 200
    assert foo_handler.times_called
```

--------------------------------

### Python Example: Accessing ClickHouse Database in Test

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/clickhouse.rst

This Python test function demonstrates how to interact with a ClickHouse database using the `clickhouse` pytest fixture. It shows how to obtain a connection to a specific database (`chat_messages`), execute a simple SQL query, and assert the returned results, illustrating basic data retrieval.

```python
def test_read_from_database(clickhouse):
    connection = clickhouse['chat_messages']
    result = connection.execute('SELECT * FROM system.numbers LIMIT 5')
    assert result == [(0,), (1,), (2,), (3,), (4,)]
```

--------------------------------

### Reading Data from PostgreSQL in Test Function

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/postgresql.rst

This Python example illustrates how to access and query a PostgreSQL database within a test function using the `pgsql` fixture. It demonstrates obtaining a cursor for a specific database and executing a `SELECT` query to retrieve data.

```python
def test_read_from_database(pgsql):
    cursor = pgsql['chat_messages'].cursor()
    cursor.execute(
        'SELECT username, text FROM messages WHERE id = %s', (data['id'],),
    )
    record = cursor.fetchone()
    assert record == ('foo', 'bar')
```

--------------------------------

### Get Source Directory for Session-Scoped Static Files in Python Testsuite

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/staticfiles.rst

This session-scoped fixture returns the source directory related to a given path, useful for accessing data files in session fixtures. It demonstrates how to read a file located alongside the test module.

```python
@pytest.fixture(scope='session')
def my_data(get_source_directory):
    return (
        testsuite_get_source_directory(__file__)
        .joinpath('foo.txt')
        .read_text()
    )
```

--------------------------------

### Python: Access MongoDB Collection via `mongodb` Fixture

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mongo.rst

This Python example illustrates how to use the `mongodb` fixture to access a specific collection and perform a `find_one` operation. It demonstrates interacting with the MongoDB database within a pytest test function, returning a `pymongo.collection.Collection` instance.

```python
def test_foo(mongodb):
    doc = mongodb.collection_name.find_one({'_id': 123})
    assert doc == {...}
```

--------------------------------

### C++ HTTP Client Error Handling for Testsuite Errors

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Provides a C++ example of an HTTP client function that processes mockserver's 599 status code and 'X-Testsuite-Error' header. It demonstrates how to map these testsuite-specific errors to internal exceptions like NetworkError or TimeoutError.

```cpp
Response HttpClient::request(const Request& request) {
  auto response = PerformRequest(request);
  if (response.status_code == 599 &&
     request.hasHeader("X-Testsuite-Error")) {
    const auto &testsuite_error =
        respones.getHeader("X-Testsuite-Error");
    if (testsuite_error == "network")
      throw NetworkError();
    if (testsuite_error == "timeout")
      throw TimeoutError()
    throw std::runtime_error("Unhandled error code");
   }
   return response;
}
```

--------------------------------

### Get Source Path for Static Files in Python Testsuite

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/staticfiles.rst

This fixture returns the source path related to a given file path, typically used to locate static files within the testsuite. It helps in constructing paths to resources relative to the test file.

```python
def test_foo(testsuite_get_source_path):
    path = testsuite_get_source_path(__file__)
    ...
```

--------------------------------

### Build Yandex Taxi Testsuite documentation

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Command to build the project's documentation.

```sh
make build-docs
```

--------------------------------

### testsuite.plugins.common API Reference for Static File Fixtures

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/staticfiles.rst

API documentation for fixtures and functions within the `testsuite.plugins.common` module, providing tools for loading, accessing, and resolving paths for static files in tests.

```APIDOC
testsuite.plugins.common:
  Functions/Fixtures:
    load():
      Description: Returns LoadFixture instance.
      Returns: LoadFixture
      Class: LoadFixture
        Members: __call__
    load_binary():
      Description: Returns LoadBinaryFixture instance.
      Returns: LoadBinaryFixture
      Class: LoadBinaryFixture
        Members: __call__
    load_json():
      Description: Returns LoadJsonFixture instance.
      Returns: LoadJsonFixture
      Class: LoadJsonFixture
        Members: __call__
    load_yaml():
      Description: Returns LoadYamlFixture instance.
      Returns: LoadYamlFixture
      Class: LoadYamlFixture
        Members: __call__
    open_file():
      Description: Returns OpenFileFixture instance.
      Returns: OpenFileFixture
      Class: OpenFileFixture
        Members: __call__
    get_file_path():
      Description: Returns GetFilePathFixture instance.
      Returns: GetFilePathFixture
      Class: GetFilePathFixture
        Members: __call__
    get_directory_path():
      Description: Returns GetDirectoryPathFixture instance.
      Returns: GetDirectoryPathFixture
      Class: GetDirectoryPathFixture
        Members: __call__
    get_search_pathes():
      Description: Returns GetSearchPathesFixture instance.
      Returns: GetSearchPathesFixture
      Class: GetSearchPathesFixture
        Members: __call__
    static_dir():
      Description: Fixture for accessing the static directory.
    initial_data_path():
      Description: Fixture for accessing initial data paths.
    testsuite_get_source_path(path: str) -> pathlib.Path:
      Description: Returns source path related to given path.
      Parameters:
        path: The path to get the source path for.
      Returns: pathlib.Path
    testsuite_get_source_directory(path: str) -> pathlib.Path:
      Description: Session scope fixture. Returns source directory related to given path.
      Parameters:
        path: The path to get the source directory for.
      Returns: pathlib.Path
```

--------------------------------

### APIDOC: Mockserver Command Line Options

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Documentation for various command-line options available to configure the mockserver's behavior, including error handling, host/port binding for HTTP and HTTPS, and Unix socket support.

```APIDOC
--mockserver-nofail: If no mockserver handler is installed for a path, mockserver returns HTTP 500 error and fails current test. This behavior can be switched off.
--mockserver-host HOST: Explicitly set HTTP mockserver hostname to bind to. Default is localhost.
--mockserver-port PORT: Explicitly set HTTP mockserver port. Default is 0 (bind to random port).
--mockserver-ssl-host HOST: Explicitly set HTTPs mockserver hostname to bind to. Default is localhost.
--mockserver-ssl-port PORT: Explicitly set HTTPs mockserver port. Default is 0 (bind to random port).
--mockserver-unix-socket PATH: Bind mockserver to a Unix domain socket. --mockserver-host and --mockserver-port options will be ignored.
```

--------------------------------

### Python: Using create_tcp_mockserver Context Manager

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/tcp_mockserver.rst

Illustrates how to instantiate and use the TCP mockserver within an async context manager for testing purposes.

```python
async with create_tcp_mockserver(host='localhost', port=0) as mockserver:
    yield mockserver
```

--------------------------------

### APIDOC: create_tcp_mockserver Function Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/tcp_mockserver.rst

API documentation for the `create_tcp_mockserver` function, which creates a TCP mockserver instance. It allows binding to a specific host and port or using an existing socket, and forwards additional keyword arguments to `asyncio.loop.create_server`.

```APIDOC
Function: create_tcp_mockserver(*, host='localhost', port=0, sock=None, **kwargs)
Parameters:
  host (str): hostname to bind to, default is 'localhost'
  port (int): port to bind to, default 0 binds to random port
  sock (socket): socket to use instead of (hostname, port) pair
  kwargs: extra parameters passed to asyncio.loop.create_server
Returns: Mockserver - An instance of Mockserver.
```

--------------------------------

### Run specific tests directly with pytest

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Steps to set up the virtual environment, activate it, and then run pytest for a specific test path, such as `tests/plugins/mockserver`.

```sh
make setup-dev-venv
. .venv-dev/bin/activate
pytest -vv tests/plugins/mockserver
```

--------------------------------

### API Reference: Testsuite Pytest Plugin Fixtures

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicerunner.rst

Documentation for fixtures and functions available in `testsuite.daemons.pytest_plugin`. It includes fixtures for client options and default headers, and functions like `register_daemon_scope`, `ensure_daemon_started`, `create_service_client`, `create_daemon_scope`, and `service_spawner_factory`, along with their associated return types and callable methods.

```APIDOC
Module: testsuite.daemons.pytest_plugin

Fixtures:
  service_client_options: Fixture that contains service client options.
  service_client_default_headers: Defines default http headers to be used by service client.

Functions:
  register_daemon_scope(name, spawn)
  ensure_daemon_started(name)

  create_service_client:
    Returns: CreateServiceClientFixture instance.
    Class: CreateServiceClientFixture()
      Methods: __call__

  create_daemon_scope:
    Returns: CreateDaemonScope instance.
    Class: CreateDaemonScope()
      Methods: __call__

  service_spawner_factory:
    Returns: ServiceSpawnerFactory instance.
    Class: ServiceSpawnerFactory()
      Methods: __call__
```

--------------------------------

### Set up development virtual environment

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Command to create a developer virtual environment for the testsuite, located in `.venv-dev`.

```sh
make setup-dev-venv
```

--------------------------------

### APIDOC: testsuite.databases.mysql.classes.ConnectionInfo Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

API documentation for the `ConnectionInfo` class, which holds detailed connection parameters for a MySQL instance, such as port, hostname, user, password, database name, and replace flag.

```APIDOC
testsuite.databases.mysql.classes.ConnectionInfo()
  port
  hostname
  user
  password
  dbname
  replace
```

--------------------------------

### APIDOC: `testsuite.mockserver.classes.MockserverInfo` Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Documentation for the `MockserverInfo` class, which encapsulates basic connection details and other relevant information about the mockserver instance, accessible via the `mockserver_info` fixture.

```APIDOC
testsuite.mockserver.classes.MockserverInfo()
```

--------------------------------

### PostgreSQL Testsuite API Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/postgresql.rst

Detailed API documentation for the PostgreSQL integration within the `testsuite` framework, including environment variables for configuration, functions for schema discovery, pytest fixtures for database management, test marks for data fixture overrides, and key classes for connection and control.

```APIDOC
Environment Variables:
  TESTSUITE_POSTGRESQL_PORT:
    Description: Use to override Postgresql server port.
    Default: 15433

Functions:
  testsuite.databases.pgsql.discover.find_schemas(root_name: str, paths: list[str | pathlib.Path]) -> dict[str, testsuite.databases.pgsql.discover.Schema]:
    Description: Discovers PostgreSQL schemas from specified paths.
    Parameters:
      root_name: The root name for the schemas.
      paths: A list of paths to search for schema definitions.

Fixtures:
  testsuite.databases.pgsql.pytest_plugin.pgsql() -> dict[str, testsuite.databases.pgsql.control.PgDatabaseWrapper]:
    Description: Provides access to PostgreSQL database wrappers, keyed by database name.
  testsuite.databases.pgsql.pytest_plugin.pgsql_cleanup_exclude_tables() -> frozenset[str]:
    Description: Redefine this fixture to exclude specific tables from cleanup after tests.
  testsuite.databases.pgsql.pytest_plugin.pgsql_local() -> dict[str, testsuite.databases.pgsql.control.PgDatabaseWrapper]:
    Description: A session-scoped fixture that creates and manages local PostgreSQL databases for testing.
  testsuite.databases.pgsql.pytest_plugin.pgsql_local_create(databases: list[testsuite.databases.pgsql.discover.Schema]) -> dict[str, testsuite.databases.pgsql.control.PgDatabaseWrapper]:
    Description: Helper fixture to create local PostgreSQL database instances from discovered schemas.
    Parameters:
      databases: A list of discovered database schema objects.

Marks:
  pytest.mark.pgsql(dbname: str, files: tuple[str, ...] = (), directories: tuple[str, ...] = (), queries: tuple[str, ...] = ()):
    Description: Use this mark to override or specify extra data fixtures in a per-test manner.
    Parameters:
      dbname: The name of the database to apply fixtures to.
      files: A tuple of filenames (e.g., SQL scripts) to apply to the database.
      directories: A tuple of directories containing SQL scripts to apply.
      queries: A tuple of raw SQL queries to apply directly to the database.

Classes:
  ServiceLocalConfig:
    Members:
      __getitem__: Access configuration items.
  testsuite.databases.pgsql.connection.PgConnectionInfo:
    Members:
      get_dsn: Returns the DSN string for connection.
      get_uri: Returns the URI string for connection.
      replace: Creates a new PgConnectionInfo instance with updated attributes.
  testsuite.databases.pgsql.control.PgDatabaseWrapper:
    Members:
      conn: The underlying psycopg2 connection object.
      conninfo: The connection information object.
      cursor: Returns a default cursor for the connection.
      dict_cursor: Returns a dictionary cursor for the connection.
```

--------------------------------

### Python: Basic Usage of redis_standalone_store Fixture for Standalone Redis

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/redis.rst

This snippet demonstrates the use of the `redis_standalone_store` fixture for interacting with a single, standalone Redis instance (without slaves or sentinels). It provides a `redis.StrictRedis()`-like interface for basic key-value operations.

```python
import pytest

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.redis.pytest_plugin',
]


def test_redis_basic(redis_standalone_store):
    redis_standalone_store.set('somekey', 'somedata')
    assert redis_standalone_store.get('somekey') == b'somedata'
```

--------------------------------

### API: `testsuite.databases.clickhouse.classes.ConnectionInfo` Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/clickhouse.rst

API documentation for the `ConnectionInfo` class within `testsuite.databases.clickhouse.classes`. This class encapsulates connection details for a ClickHouse instance, including host, TCP port, HTTP port, and database name.

```APIDOC
ConnectionInfo()
  :members: host, tcp_port, http_port, dbname
```

--------------------------------

### Python Fixtures for Testsuite Service Management

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicerunner.rst

This Python code defines `pytest` fixtures used within the testsuite framework. The `server_client` fixture ensures the service is running and provides a client instance, while `service_daemon` handles the service's lifecycle, including configuration generation and daemon scope management.

```python
import pytest

SERVICE_BASEURL = 'http://localhost:8080/'


@pytest.fixture
async def server_client(
        ensure_daemon_started,
        create_service_client,
        mockserver,
        service_daemon,
):
    # Start service if not started yet
    await ensure_daemon_started(service_daemon)
    # Create service client instance
    return create_service_client(SERVICE_BASEURL)


@pytest.fixture(scope='session')
async def service_daemon(create_daemon_scope):
    # Generate special config for testsuite.
    service_config_path = generate_service_config()

    async with create_daemon_scope(
            args=['path-to-service-binary', '--config', service_config_path],
            ping_url=SERVICE_BASEURL + 'ping',
    ) as scope:
        yield scope
```

--------------------------------

### APIDOC: `ConnectionInfo` Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/rabbitmq.rst

Documentation for the `ConnectionInfo` class, located in `testsuite.databases.rabbitmq.classes`. This class encapsulates the connection details for a RabbitMQ instance, specifically its host and TCP port.

```APIDOC
ConnectionInfo()
  members:
    host
    tcp_port
```

--------------------------------

### API: `testsuite.databases.clickhouse.pytest_plugin` Fixtures

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/clickhouse.rst

API documentation for key pytest fixtures provided by `testsuite.databases.clickhouse.pytest_plugin`. These fixtures include `clickhouse`, `clickhouse_local`, and `clickhouse_conn_info`, which facilitate ClickHouse integration and testing.

```APIDOC
clickhouse()
clickhouse_local()
clickhouse_conn_info()
```

--------------------------------

### APIDOC: Mockserver Class Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/tcp_mockserver.rst

API documentation for the `Mockserver` class, representing a TCP mockserver instance. This class provides methods for interacting with the mockserver (members are not detailed in this snippet, refer to full documentation).

```APIDOC
Class: Mockserver
Description: Represents a TCP mockserver instance.
Members: (Details not provided in this documentation snippet. Refer to full class documentation.)
```

--------------------------------

### APIDOC: testsuite.databases.mysql.control.ConnectionWrapper Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

API documentation for the `ConnectionWrapper` class, which encapsulates a MySQL connection and provides direct access to connection information and a database cursor.

```APIDOC
testsuite.databases.mysql.control.ConnectionWrapper()
  conninfo
  cursor
```

--------------------------------

### Python: Basic RabbitMQ Message Publishing and Consumption

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/rabbitmq.rst

This asynchronous test function demonstrates how to interact with RabbitMQ using the "rabbitmq" fixture. It shows the process of declaring an exchange and a queue, binding them, publishing a message, and then consuming the message to verify its content.

```python
async def test_rabbitmq_basic(rabbitmq):
    exchange = 'testsuite_exchange'
    queue = 'testsuite_queue'
    routing_key = 'testsuite_routing_key'

    channel = await rabbitmq.get_channel()

    async with channel:
        await channel.declare_exchange(
            exchange=exchange, exchange_type='fanout'
        )
        await channel.declare_queue(queue=queue)
        await channel.bind_queue(
            queue=queue, exchange=exchange, routing_key=routing_key
        )
        await channel.publish(
            exchange=exchange,
            routing_key=routing_key,
            body=b'hi from testsuite!',
        )

        messages = await channel.consume(queue=queue, count=1)
        assert messages == [b'hi from testsuite!']
```

--------------------------------

### Kafka Pytest Fixtures API Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/kafka.rst

Documents the available pytest fixtures provided by `testsuite.databases.kafka.pytest_plugin` for integrating Kafka into tests. These fixtures simplify interaction with Kafka producers, consumers, and custom topic configurations.

```APIDOC
kafka_producer():
  Description: A pytest fixture that provides an asynchronous Kafka producer instance.
  Returns: testsuite.databases.kafka.classes.KafkaProducer

kafka_consumer():
  Description: A pytest fixture that provides an asynchronous Kafka consumer instance.
  Returns: testsuite.databases.kafka.classes.KafkaConsumer

kafka_custom_topics():
  Description: A pytest fixture that allows overriding the default Kafka topic partition configuration.
  Returns: dict[str, int] (mapping topic names to partition counts)

kafka_local():
  Description: A pytest fixture that manages the lifecycle of a local Kafka service instance for testing.
  Returns: None
```

--------------------------------

### Auto-format Yandex Taxi Testsuite source code

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Command to automatically format the source code using `make`.

```sh
make venv-format
```

--------------------------------

### Python Pytest Plugin and Custom Kafka Topics Integration

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/kafka.rst

This snippet illustrates how to configure `pytest_plugins` to include the Kafka testsuite plugin and how to define custom Kafka topics with specific partition counts using a `kafka_custom_topics` fixture. This allows for more granular control over topic creation in tests.

```python
pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.kafka.pytest_plugin',
]

KAFKA_CUSTOM_TOPICS = {
    'Large-topic-1': 7,
    'Large-topic-2': 3,
}

@pytest.fixture(scope='session')
def kafka_custom_topics():
    return KAFKA_CUSTOM_TOPICS
```

--------------------------------

### APIDOC: testsuite.databases.mysql.pytest_plugin.mysql_conninfo Fixture

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

API documentation for the `mysql_conninfo` pytest fixture provided by `testsuite`, which provides connection information for the MySQL service.

```APIDOC
testsuite.databases.mysql.pytest_plugin.mysql_conninfo()
```

--------------------------------

### APIDOC: `mockserver_info` Pytest Fixture

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Documentation for the `mockserver_info` fixture, a session-scoped fixture that provides an instance of `testsuite.mockserver.classes.MockserverInfo`. It contains essential information about the running mockserver, such as its host and port.

```APIDOC
mockserver_info()
  Returns testsuite.mockserver.classes.MockserverInfo instance containing basic information about mockserver.
```

--------------------------------

### Run development tests using Makefile

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/README.rst

Command to run tests during development using the project's Makefile.

```sh
make venv-tests
```

--------------------------------

### Pytest Plugin and Schema Discovery for PostgreSQL

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/postgresql.rst

This Python snippet demonstrates how to configure `pytest_plugins` to enable PostgreSQL support in `testsuite`. It also shows how to use `discover.find_schemas` to locate and load database schemas from a specified path, preparing them for test execution.

```python
from testsuite.databases.pgsql import discover

pytest_plugins = [
    'testsuite.pytest_plugin',
    'testsuite.databases.pgsql.pytest_plugin',
]


@pytest.fixture(scope='session')
def pgsql_local(pgsql_local_create):
    tests_dir = pathlib.Path(__file__).parent
    sqldata_path = tests_dir.joinpath('../schemas/postgresql')
    databases = discover.find_schemas('service_name', [sqldata_path])
    return pgsql_local_create(list(databases.values()))
```

--------------------------------

### APIDOC: `testsuite.mockserver.server.MockserverFixture` Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Documentation for the `MockserverFixture` class, which is returned by the `mockserver` fixture. This class provides methods for setting up and controlling mock HTTP responses within tests.

```APIDOC
testsuite.mockserver.server.MockserverFixture()
```

--------------------------------

### Registering Logfile with Custom Formatter (Python)

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicelogs.rst

Demonstrates how to register a logfile with the testsuite, providing a custom formatter factory to process log lines. The formatter can have its own context and transforms byte lines into strings.

```python
async def test_formatter(servicelogs_register_logfile):
    logfile = pathlib.Path(...)

    def formatter_factory():
        # formatter may have its own context
        def formatter(line: bytes) -> str | None:
            return line.encode('utf-8')
        return formatter

    servicelogs_register_logfile(
        logfile, title='My service logs',
        formatter_factory=formatter_factory
    )
```

```APIDOC
servicelogs_register_logfile(path: pathlib.Path, *, title: str, formatter_factory)
  Registers logfile into testsuite.
  path: path to logfile
  title: logfile name
  formatter_factory: Log formatter factory
```

--------------------------------

### APIDOC: `testsuite.databases.mongo.connection.ConnectionInfo` Class

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mongo.rst

API documentation for the `ConnectionInfo` class, which provides details about the MongoDB connection, including database name, host, port, and methods for retrieving the URI.

```APIDOC
testsuite.databases.mongo.connection.ConnectionInfo():
  Members:
    dbname
    host
    port
    retry_writes
    get_uri
```

--------------------------------

### Python: Pytest Session Fixture for Mockserver URL Construction

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Shows how to create a session-scoped pytest fixture (`service_args`) that utilizes `mockserver_info.url()` to dynamically build mockserver URLs for different services. This allows services under test to be configured to communicate with the mockserver during testing.

```python
@pytest.fixture(scope='session')
def service_args(mockserver_info):
    """Build service startup args."""
    return (
        'bin/service',
        '--service1-base-url',
        mockserver_info.url('service1'),
        '--service2-base-url',
        mockserver_info.url('service2'),
    )
```

--------------------------------

### Pytest OpenTracing Configuration for Mockserver

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Shows the 'pytest.ini' configuration options to enable OpenTracing integration with the mockserver. These settings specify whether tracing is enabled and which HTTP headers carry the trace and span IDs.

```text
mockserver-tracing-enabled = true
mockserver-trace-id-header = X-TraceId
mockserver-span-id-header = X-SpanId
```

--------------------------------

### APIDOC: Mockserver `pytest.ini` Configuration Options

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mockserver.rst

Details the configuration options that can be set in `pytest.ini` to control mockserver's tracing behavior, custom header names, SSL certificate paths, and HTTP proxy functionality.

```APIDOC
mockserver-tracing-enabled: Boolean flag. Controls how mockserver handles requests with trace-id header set.
  - When request trace-id header not from testsuite:
    - True: handle, if handler missing return http status 500.
    - False: handle, if handler missing raise HandlerNotFoundError.
  - When request trace-id header from other test:
    - True: do not handle, return http status 500.
    - False: handle, if handler missing raise HandlerNotFoundError.
mockserver-trace-id-header: Name of tracing http header, value changes from test to test and is constant within test. Default value is X-YaTraceId.
mockserver-span-id-header: Name of tracing http header, value is unique for each request. Default value is X-YaSpanId.
mockserver-ssl-cert-file: Path to SSL certificate file to set up mockserver_ssl.
mockserver-ssl-key-file: Path to SSL key file to set up mockserver_ssl.
mockserver-http-proxy-enabled: When enabled, mockserver acts as http proxy. Is disabled by default.
```

--------------------------------

### APIDOC: pytest.mark.mysql Mark

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/mysql.rst

API documentation for the `pytest.mark.mysql` mark, used to specify extra data fixtures for MySQL databases on a per-test basis. It allows defining the database name, and lists of files, directories, or queries to apply.

```APIDOC
pytest.mark.mysql(dbname, *, files=(), directories=(), queries=())
  dbname: Database name.
  files: List of filenames to apply to the database.
  directories: List of directories to apply to the database.
  queries: List of queries to apply to the database.
```

--------------------------------

### Capture Class API Reference

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/servicelogs.rst

API documentation for the `Capture` class, specifying its methods for selecting and subscribing to log capture events.

```APIDOC
Capture
  members: select, subscribe
```

--------------------------------

### APIDOC: Testsuite Advanced Matching Utilities

Source: https://github.com/yandex/yandex-taxi-testsuite/blob/develop/docs/utils.rst

API documentation for advanced matching utilities in `testsuite.matching`. This comprehensive section covers type matching with `IsInstance`, logical operations such as `And`, `Or`, and `Not` for combining matchers, and specialized matchers for dictionaries (`AnyDict`, `DictOf`, `PartialDict`, `recursive_partial_dict`) and lists (`AnyList`, `ListOf`, `unordered_list`). Additionally, it includes `Capture` for extracting matched values.

```APIDOC
Classes (testsuite.matching):
  - IsInstance
  - And
  - Or
  - Not
  - AnyDict
  - DictOf
  - PartialDict
  - AnyList
  - ListOf
  - Capture
Functions (testsuite.matching):
  - recursive_partial_dict
  - unordered_list
```
Tokens:
Raw