config.py

"""
Work with the configuration files of the service in testsuite.
"""
 
# pylint: disable=redefined-outer-name
import copy
import dataclasses
import itertools
import logging
import os
import pathlib
import string
import subprocess
import types
from typing import Any
from collections.abc import Callable
from collections.abc import Mapping
from typing import TypeAlias
 
import pytest
import yaml
 
# flake8: noqa E266

USERVER_CONFIG_HOOKS = [
    'userver_config_substitutions',
    'userver_config_http_server',
    'userver_config_http_client',
    'userver_config_logging',
    'userver_config_logging_otlp',
    'userver_config_testsuite',
    'userver_config_secdist',
    'userver_config_testsuite_middleware',
    'userver_config_deadlock_detector',
]
 
ServiceConfigPatch: TypeAlias = Callable[[dict, dict], None]
 
 
# @cond
 
 
logger = logging.getLogger(__name__)
 
 
class _UserverConfigPlugin:
    def __init__(self):
        self._config_hooks = []
 
    @property
    def userver_config_hooks(self):
        return self._config_hooks
 
    def pytest_plugin_registered(self, plugin, manager):
        if not isinstance(plugin, types.ModuleType):
            return
        uhooks = getattr(plugin, 'USERVER_CONFIG_HOOKS', None)
        if uhooks is not None:
            self._config_hooks.extend(uhooks)
 
 
@dataclasses.dataclass(frozen=True)
class _UserverConfig:
    config_yaml: dict
    config_vars: dict
 
 
def pytest_configure(config):
    config.pluginmanager.register(_UserverConfigPlugin(), 'userver_config')
 
 
def pytest_addoption(parser) -> None:
    group = parser.getgroup('userver-config')
    group.addoption(
        '--service-log-level',
        type=str.lower,
        choices=['trace', 'debug', 'info', 'warning', 'error', 'critical'],
    )
    group.addoption(
        '--service-config',
        type=pathlib.Path,
        help='Path to service.yaml file.',
    )
    group.addoption(
        '--service-config-vars',
        type=pathlib.Path,
        help='Path to config_vars.yaml file.',
    )
    group.addoption(
        '--service-secdist',
        type=pathlib.Path,
        help='Path to secure_data.json file.',
    )
    group.addoption(
        '--config-fallback',
        type=pathlib.Path,
        help='Path to dynamic config fallback file.',
    )
    group.addoption(
        '--dump-config',
        action='store_true',
        help='Dump config from binary before running tests',
    )
 
 
# @endcond
 
 
@pytest.fixture(scope='session')
def service_config_path(pytestconfig, service_binary) -> pathlib.Path:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    if pytestconfig.option.dump_config:
        subprocess.run([
            service_binary,
            '--dump-config',
            pytestconfig.option.service_config,
        ])
    return pytestconfig.option.service_config
 
 
@pytest.fixture(scope='session')
def db_dump_schema_path(service_binary, service_tmpdir) -> pathlib.Path:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
 
    """
    path = service_tmpdir.joinpath('schemas')
    os.mkdir(path)
    subprocess.run([
        service_binary,
        '--dump-db-schema',
        path / '0_db_schema.sql',
    ])
    return path
 
 
@pytest.fixture(scope='session')
def service_config_vars_path(pytestconfig) -> pathlib.Path | None:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    return pytestconfig.option.service_config_vars
 
 
@pytest.fixture(scope='session')
def service_secdist_path(pytestconfig) -> pathlib.Path | None:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    return pytestconfig.option.service_secdist
 
 
@pytest.fixture(scope='session')
def config_fallback_path(pytestconfig) -> pathlib.Path:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    return pytestconfig.option.config_fallback
 
 
@pytest.fixture(scope='session')
def 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.
 
    @ingroup userver_testsuite_fixtures
    """
    return tmp_path_factory.mktemp(
        pathlib.Path(service_binary).name,
        numbered=False,
    )
 
 
@pytest.fixture(scope='session')
def service_config_path_temp(
    service_tmpdir,
    service_config,
    service_config_yaml,
    service_config_vars,
) -> pathlib.Path:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    dst_path = service_tmpdir / 'config.yaml'
 
    service_config_yaml = dict(service_config_yaml)
    if not service_config_vars:
        service_config_yaml.pop('config_vars', None)
    else:
        config_vars_path = service_tmpdir / 'config_vars.yaml'
        config_vars_text = yaml.dump(service_config_vars)
        logger.debug(
            'userver fixture "service_config_path_temp" writes the patched static config vars to "%s":\n%s',
            config_vars_path,
            config_vars_text,
        )
        config_vars_path.write_text(config_vars_text)
        service_config_yaml['config_vars'] = str(config_vars_path)
 
    logger.debug(
        'userver fixture "service_config_path_temp" writes the patched static config to "%s" equivalent to:\n%s',
        dst_path,
        yaml.dump(service_config),
    )
    dst_path.write_text(yaml.dump(service_config_yaml))
 
    return dst_path
 
 
@pytest.fixture(scope='session')
def service_config_yaml(_service_config_hooked) -> dict:
    """
    Returns the static config values after the USERVER_CONFIG_HOOKS were
    applied (if any). Prefer using
    pytest_userver.plugins.config.service_config
 
    @ingroup userver_testsuite_fixtures
    """
    return _service_config_hooked.config_yaml
 
 
@pytest.fixture(scope='session')
def service_config_vars(_service_config_hooked) -> dict:
    """
    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
 
    @ingroup userver_testsuite_fixtures
    """
    return _service_config_hooked.config_vars
 
 
def _substitute_values(config, service_config_vars: dict, service_env) -> None:
    if isinstance(config, dict):
        for key, value in config.items():
            if not isinstance(value, str):
                _substitute_values(value, service_config_vars, service_env)
                continue
 
            if not value.startswith('$'):
                continue
 
            new_value = service_config_vars.get(value[1:])
            if new_value is not None:
                config[key] = new_value
                continue
 
            env = config.get(f'{key}#env')
            if env:
                if service_env:
                    new_value = service_env.get(env)
                if not new_value:
                    new_value = os.environ.get(env)
                if new_value:
                    config[key] = new_value
                    continue
 
            fallback = config.get(f'{key}#fallback')
            if fallback:
                config[key] = fallback
                continue
 
            config[key] = None
 
    if isinstance(config, list):
        for i, value in enumerate(config):
            if not isinstance(value, str):
                _substitute_values(value, service_config_vars, service_env)
                continue
 
            if not value.startswith('$'):
                continue
 
            new_value = service_config_vars.get(value[1:])
            if new_value is not None:
                config[i] = new_value
 
 
@pytest.fixture(scope='session')
def substitute_config_vars(service_env) -> Callable[[Any, dict], Any]:
    """
    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!
 
    @ingroup userver_testsuite_fixtures
    """
 
    def substitute(config_yaml, config_vars, /):
        if config_yaml is not None and not isinstance(config_yaml, dict) and not isinstance(config_yaml, list):
            raise TypeError(
                f'{substitute_config_vars.__name__} can only be meaningfully '
                'called with dict and list nodes of config_yaml, while given: '
                f'{config_yaml!r}. Pass a containing object instead.',
            )
 
        config = copy.deepcopy(config_yaml)
        _substitute_values(config, config_vars, service_env)
        return config
 
    return substitute
 
 
@pytest.fixture(scope='session')
def service_config(
    service_config_yaml,
    service_config_vars,
    substitute_config_vars,
) -> dict:
    """
    Returns the static config values after the USERVER_CONFIG_HOOKS were
    applied (if any) and with all the '$', environment and fallback variables
    substituted.
 
    @ingroup userver_testsuite_fixtures
    """
    config = substitute_config_vars(service_config_yaml, service_config_vars)
    config.pop('config_vars', None)
    return config
 
 
@pytest.fixture(scope='session')
def _original_service_config(
    service_config_path,
    service_config_vars_path,
) -> _UserverConfig:
    config_vars: dict
    config_yaml: dict
 
    assert service_config_path is not None, 'Please specify proper path to the static config file, not None'
 
    with open(service_config_path, mode='rt') as fp:
        config_yaml = yaml.safe_load(fp)
 
    if service_config_vars_path:
        with open(service_config_vars_path, mode='rt') as fp:
            config_vars = yaml.safe_load(fp)
    else:
        config_vars = {}
 
    return _UserverConfig(config_yaml=config_yaml, config_vars=config_vars)
 
 
@pytest.fixture(scope='session')
def _service_config_hooked(
    daemon_scoped_mark,
    pytestconfig,
    request,
    _original_service_config,
) -> _UserverConfig:
    config_yaml = copy.deepcopy(_original_service_config.config_yaml)
    config_vars = copy.deepcopy(_original_service_config.config_vars)
 
    plugin = pytestconfig.pluginmanager.get_plugin('userver_config')
    local_hooks = (daemon_scoped_mark or {}).get('config_hooks', ())
 
    for hook in itertools.chain(plugin.userver_config_hooks, local_hooks):
        if not callable(hook):
            hook_func = request.getfixturevalue(hook)
        else:
            hook_func = hook
        hook_func(config_yaml, config_vars)
 
    return _UserverConfig(config_yaml=config_yaml, config_vars=config_vars)
 
 
@pytest.fixture(scope='session')
def _service_config_substitution_vars(request, mockserver_info) -> Mapping[str, str]:
    substitution_vars = {
        'mockserver': mockserver_info.base_url.removesuffix('/'),
    }
    if request.config.pluginmanager.hasplugin('pytest_userver.plugins.grpc.mockserver'):
        grpc_mockserver_endpoint = request.getfixturevalue('grpc_mockserver_endpoint')
        substitution_vars['grpc_mockserver'] = grpc_mockserver_endpoint
    return substitution_vars
 
 
@pytest.fixture(scope='session')
def userver_config_substitutions(_service_config_substitution_vars) -> ServiceConfigPatch:
    """
    Replaces substitution vars in all strings within `config_vars` using
    [string.Template.substitute](https://docs.python.org/3/library/string.html#string.Template.substitute).
 
    Substitution vars can be used as a shorthand for writing a full-fledged @ref SERVICE_CONFIG_HOOKS "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
 
    @code{.yaml}
    frobnicator-url: $mockserver/frobnicator
    @endcode
 
    a possible patching result is as follows:
 
    @code{.yaml}
    frobnicator-url: http://127.0.0.1:1234/frobnicator
    @endcode
 
    Currently, the following substitution vars are supported:
 
    * `mockserver` - mockserver url
    * `grpc_mockserver` - grpc mockserver endpoint
 
    @ingroup userver_testsuite_fixtures
    """
 
    def _substitute(key, value, parent: list | dict) -> None:
        if isinstance(value, str):
            parent[key] = string.Template(value).safe_substitute(_service_config_substitution_vars)
        elif isinstance(value, dict):
            for child_key, child_value in value.items():
                _substitute(child_key, child_value, value)
        elif isinstance(value, list):
            for child_key, child_value in enumerate(value):
                _substitute(child_key, child_value, value)
 
    def patch_config(config_yaml, config_vars):
        for key, value in config_vars.items():
            _substitute(key, value, config_vars)
 
    return patch_config
 
 
@pytest.fixture(scope='session')
def userver_config_http_server(service_port, monitor_port) -> ServiceConfigPatch:
    """
    Returns a function that adjusts the static configuration file for testsuite.
    Sets the `server.listener.port` to listen on
    @ref pytest_userver.plugins.base.service_port "service_port" fixture value;
    sets the `server.listener-monitor.port` to listen on
    @ref pytest_userver.plugins.base.monitor_port "monitor_port"
    fixture value.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def _patch_config(config_yaml, config_vars):
        components = config_yaml['components_manager']['components']
        if server := components.get('server'):
            for listener_name, new_port in [('listener', service_port), ('listener-monitor', monitor_port)]:
                if listener := server.get(listener_name):
                    ports = listener.get('ports')
                    if ports and 'port' in ports[0]:
                        ports[0]['port'] = new_port
                    else:
                        listener['port'] = new_port
 
    return _patch_config
 
 
@pytest.fixture(scope='session')
def allowed_url_prefixes_extra() -> list[str]:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    return []
 
 
@pytest.fixture(scope='session')
def allowed_url_prefixes(
    mockserver_info,
    mockserver_ssl_info,
    allowed_url_prefixes_extra,
) -> list[str]:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
    allowed_urls: list[str] = [mockserver_info.base_url]
 
    if mockserver_ssl_info:
        allowed_urls.append(mockserver_ssl_info.base_url)
 
    allowed_urls += allowed_url_prefixes_extra
 
    # Add WebSocket prefixes (ws:// and wss://) alongside HTTP prefixes
    allowed_urls.append(mockserver_info.ws_url('/'))
    if mockserver_ssl_info:
        allowed_urls.append(mockserver_ssl_info.ws_url('/'))
 
    return allowed_urls
 
 
@pytest.fixture(scope='session')
def userver_config_http_client(
    allowed_url_prefixes,
) -> ServiceConfigPatch:
    """
    Returns a function that adjusts the static configuration file for testsuite.
    Sets increased timeout and limits allowed URLs for `http-client-core` component.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def patch_config(config, config_vars) -> None:
        components: dict = config['components_manager']['components']
        if not {'http-client-core', 'testsuite-support'}.issubset(
            components.keys(),
        ):
            return
        if components['http-client-core'] is None:
            components['http-client-core'] = {}
        http_client_core = components['http-client-core']
        http_client_core['testsuite-enabled'] = True
        http_client_core['testsuite-timeout'] = '10s'
 
        http_client_core['testsuite-allowed-url-prefixes'] = allowed_url_prefixes
 
    return patch_config
 
 
@pytest.fixture(scope='session')
def userver_default_log_level() -> str:
    """
    Default log level to use in userver if no command line option was provided.
 
    Returns 'debug'.
 
    @ingroup userver_testsuite_fixtures
    """
    return 'debug'
 
 
@pytest.fixture(scope='session')
def userver_log_level(pytestconfig, userver_default_log_level) -> str:
    """
    Returns --service-log-level value if provided, otherwise returns
    userver_default_log_level() value from fixture.
 
    @ingroup userver_testsuite_fixtures
    """
    if pytestconfig.option.service_log_level:
        return pytestconfig.option.service_log_level
    return userver_default_log_level
 
 
@pytest.fixture(scope='session')
def userver_config_logging(userver_log_level, _service_logfile_path) -> ServiceConfigPatch:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
 
    if _service_logfile_path:
        default_file_path = str(_service_logfile_path)
    else:
        default_file_path = '@stderr'
 
    def _patch_config(config_yaml, config_vars):
        components = config_yaml['components_manager']['components']
        if 'logging' in components:
            loggers = components['logging'].setdefault('loggers', {})
            for logger in loggers.values():
                logger['file_path'] = '@null'
            loggers['default'] = {
                'file_path': default_file_path,
                'level': userver_log_level,
                'overflow_behavior': 'discard',
            }
        config_vars['logger_level'] = userver_log_level
 
    return _patch_config
 
 
@pytest.fixture(scope='session')
def userver_config_logging_otlp() -> ServiceConfigPatch:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def _patch_config(config_yaml, config_vars):
        components = config_yaml['components_manager']['components']
        if 'otlp-logger' in components:
            components['otlp-logger']['load-enabled'] = False
 
    return _patch_config
 
 
@pytest.fixture(scope='session')
def userver_config_testsuite(pytestconfig, mockserver_info) -> ServiceConfigPatch:
    """
    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.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def _set_postgresql_options(testsuite_support: dict) -> None:
        testsuite_support['testsuite-pg-execute-timeout'] = '35s'
        testsuite_support['testsuite-pg-statement-timeout'] = '30s'
        testsuite_support['testsuite-pg-readonly-master-expected'] = True
 
    def _set_redis_timeout(testsuite_support: dict) -> None:
        testsuite_support['testsuite-redis-timeout-connect'] = '40s'
        testsuite_support['testsuite-redis-timeout-single'] = '30s'
        testsuite_support['testsuite-redis-timeout-all'] = '30s'
 
    def _disable_cache_periodic_update(testsuite_support: dict) -> None:
        testsuite_support['testsuite-periodic-update-enabled'] = False
 
    def patch_config(config, config_vars) -> None:
        # Don't delay tests teardown unnecessarily.
        config['components_manager'].pop('graceful_shutdown_interval', None)
        config['components_manager'].pop('graceful_shutdown_continue_accepting_requests_interval', None)
        config['components_manager'].pop('graceful_shutdown_pending_requests_completion_interval', None)
        components: dict = config['components_manager']['components']
        if 'testsuite-support' not in components:
            return
        if components['testsuite-support'] is None:
            components['testsuite-support'] = {}
        testsuite_support = components['testsuite-support']
        testsuite_support['testsuite-increased-timeout'] = '30s'
        testsuite_support['testsuite-grpc-is-tls-enabled'] = False
        _set_postgresql_options(testsuite_support)
        _set_redis_timeout(testsuite_support)
        service_runner = pytestconfig.option.service_runner_mode
        if not service_runner:
            _disable_cache_periodic_update(testsuite_support)
        testsuite_support['testsuite-tasks-enabled'] = not service_runner
        testsuite_support['testsuite-periodic-dumps-enabled'] = '$userver-dumps-periodic'
        components['testsuite-support'] = testsuite_support
 
        config_vars['testsuite-enabled'] = True
        if 'tests-control' in components:
            components['tests-control']['testpoint-url'] = mockserver_info.url(
                'testpoint',
            )
 
    return patch_config
 
 
@pytest.fixture(scope='session')
def userver_config_secdist(service_secdist_path) -> ServiceConfigPatch:
    """
    Returns a function that adjusts the static configuration file for testsuite.
    Sets the `default-secdist-provider.config` to the value of
    @ref pytest_userver.plugins.config.service_secdist_path "service_secdist_path"
    fixture.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def _patch_config(config_yaml, config_vars):
        if not service_secdist_path:
            return
 
        components = config_yaml['components_manager']['components']
        if 'default-secdist-provider' not in components:
            return
 
        if not service_secdist_path.is_file():
            raise ValueError(
                f'"{service_secdist_path}" is not a file. Provide a '
                f'"--service-secdist" pytest option or override the '
                f'"service_secdist_path" fixture.',
            )
        components['default-secdist-provider']['config'] = str(
            service_secdist_path,
        )
 
    return _patch_config
 
 
@pytest.fixture(scope='session')
def userver_config_testsuite_middleware(userver_testsuite_middleware_enabled: bool) -> ServiceConfigPatch:
    def patch_config(config_yaml, config_vars):
        if not userver_testsuite_middleware_enabled:
            return
 
        components = config_yaml['components_manager']['components']
        if 'server' not in components:
            return
 
        pipeline_builder_name = components['server'].get(
            'middleware-pipeline-builder', 'default-server-middleware-pipeline-builder'
        )
        pipeline_builder = components.setdefault(pipeline_builder_name, {})
        middlewares = pipeline_builder.setdefault('append', [])
        middlewares.append('testsuite-exceptions-handling-middleware')
 
    return patch_config
 
 
@pytest.fixture(scope='session')
def userver_testsuite_middleware_enabled() -> bool:
    """Whether testsuite middleware is enabled."""
    return True
 
 
@pytest.fixture(scope='session')
def userver_config_deadlock_detector(userver_deadlock_detector_mode: str) -> ServiceConfigPatch:
    """
    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
    @ref pytest_userver.plugins.config.userver_deadlock_detector_mode "userver_deadlock_detector_mode"
    fixture.
 
    @ingroup userver_testsuite_fixtures
    """
 
    def patch_config(config_yaml, config_vars):
        coro_pool = config_yaml['components_manager'].setdefault('coro_pool', {})
        coro_pool['deadlock_detector'] = userver_deadlock_detector_mode
 
    return patch_config
 
 
@pytest.fixture(scope='session')
def userver_deadlock_detector_mode() -> str:
    """
    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`.
 
    @ingroup userver_testsuite_fixtures
    """
    return 'detect-only'