Spawning and accessing service¶

Testsuite starts process with a service being testsed. Service must implement ping-url that returns 200 OK when service is up.

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

Fixtures¶

service_client_options¶

Fixture that contains service client options.

service_client_default_headers¶

Defines default http headers to be used by service client.

register_daemon_scope¶

testsuite.daemons.pytest_plugin.register_daemon_scope(name, spawn)[source]¶

Context manager that registers service process session.

Yields daemon scope instance.

Parameters:: name – service name
Spawn spawn:: spawner function

ensure_daemon_started¶

testsuite.daemons.pytest_plugin.ensure_daemon_started(name)¶: Fixture that starts requested service.

create_service_client¶

testsuite.daemons.pytest_plugin.create_service_client()¶: Returns CreateServiceClientFixture instance.

class testsuite.daemons.pytest_plugin.CreateServiceClientFixture[source]¶

Creates service client instance.

Example:

def my_client(create_service_client):
    return create_service_client('http://localhost:9999/')

__call__(base_url: str, *, client_class=<class 'testsuite.daemons.service_client.Client'>, **kwargs)[source]¶

Parameters:

base_url – base url for http client
client_class – client class to use

Returns:

client_class instance

create_daemon_scope¶

testsuite.daemons.pytest_plugin.create_daemon_scope()¶: Returns CreateDaemonScope instance.

class testsuite.daemons.pytest_plugin.CreateDaemonScope[source]¶

Create daemon scope for daemon with command to start.

__call__(*, args: Sequence[str], check_url: str = None, ping_url: str = None, name: Optional[str] = None, base_command: Optional[Sequence] = None, env: Optional[Dict[str, str]] = None, poll_retries: int = 2000, ping_request_timeout: float = 1.0, ping_response_codes: Tuple[int] = (200,), health_check: Optional[Callable[[...], Awaitable[bool]]] = None, subprocess_options: Optional[Dict[str, Any]] = None, setup_service: Optional[Callable[[Popen], None]] = None, shutdown_signal: Optional[int] = None, stdout_handler=None, stderr_handler=None) → AsyncContextManager[_DaemonScope][source]¶

Parameters:

args – command arguments
base_command – Arguments to be prepended to args.
env – Environment variables dictionary.
poll_retries – Number of tries for service health check
ping_url – service health check url, service is considered up when 200 received.
ping_request_timeout – Timeout for ping_url request
ping_response_codes – HTTP resopnse codes tuple meaning that service is up and running.
health_check – Async function to check service is running.
subprocess_options – Custom subprocess options.
setup_service – Function to be called right after service is started.
shutdown_signal – Signal used to stop running services.

Returns:

Returns internal daemon scope instance to be used with ensure_daemon_started fixture.

service_spawner¶

Fixture that creates service spawner.

testsuite.daemons.pytest_plugin.service_spawner(args, *, ...)¶

Classes¶

class testsuite.daemons.service_client.Client[source]¶

__init__(base_url: str, *, session: ClientSession, ssl_context: Optional[SSLContext] = None, span_id_header: Optional[str] = None, headers: Optional[Dict[str, str]] = None, timeout: float = 120.0)¶

Parameters:

base_url – Base client url
session – aiohttp.ClientSession instance
headers – default request headers dictionary
timeout – http client default timeout

coroutine delete(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP DELETE request.

coroutine get(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP GET request.

coroutine options(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP OPTIONS request.

coroutine patch(path, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP PATCH request.

coroutine post(path: str, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP POST request.

coroutine put(path, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP PUT request.

coroutine request(http_method: str, path: str, **kwargs) → TResponse¶: Perform HTTP http_method request.

class testsuite.daemons.service_client.AiohttpClient[source]¶

__init__(base_url: str, *, session: ClientSession, ssl_context: Optional[SSLContext] = None, span_id_header: Optional[str] = None, headers: Optional[Dict[str, str]] = None, timeout: float = 120.0)¶

Parameters:

base_url – Base client url
session – aiohttp.ClientSession instance
headers – default request headers dictionary
timeout – http client default timeout

coroutine delete(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP DELETE request.

coroutine get(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP GET request.

coroutine options(path: str, headers: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, **kwargs) → TResponse¶: Perform HTTP OPTIONS request.

coroutine patch(path, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP PATCH request.

coroutine post(path: str, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP POST request.

coroutine put(path, json: Optional[Union[int, float, str, list, dict]] = None, data: Any = None, params: Optional[Dict[str, str]] = None, bearer: Optional[str] = None, x_real_ip: Optional[str] = None, headers: Optional[Dict[str, str]] = None, **kwargs) → TResponse¶: Perform HTTP PUT request.

coroutine request(http_method: str, path: str, **kwargs) → TResponse¶: Perform HTTP http_method request.

Spawning and accessing service¶

Fixtures¶

service_client_options¶

service_client_default_headers¶

register_daemon_scope¶

ensure_daemon_started¶

create_service_client¶

create_daemon_scope¶

service_spawner¶

Classes¶

Navigation

Related Topics