log_capture.py

"""
Capture and work with logs.
"""
 
# pylint: disable=redefined-outer-name
import logging
 
import pytest
 
from testsuite import logcapture
from testsuite.logcapture import __tracebackhide__  # noqa
 
from ..utils import tskv
 
USERVER_CONFIG_HOOKS = ['_userver_config_logs_capture']
DEFAULT_PORT = 2211
 
logger = logging.getLogger(__name__)
 
 
def pytest_addoption(parser) -> None:
    group = parser.getgroup('logs-capture')
    group.addoption(
        '--logs-capture-port',
        type=int,
        default=0,
        help='Port to bind logs-capture server to.',
    )
    group.addoption(
        '--logs-capture-host',
        default='localhost',
        help='Host to bind logs-capture server to.',
    )
 
 
@pytest.fixture(scope='session')
async def userver_log_capture(pytestconfig, userver_log_level) -> logcapture.CaptureServer:
    """
    Session-scoped TCP server that receives service logs in TSKV format.
 
    The server is started once per testsuite session. Static config is patched
    so that the service can send logs when log capture is enabled for a test.
 
    In most tests use
    @ref pytest_userver.plugins.service_client.service_client "service_client".capture_logs()
    instead of this fixture directly. See @ref testsuite_logs_capture.
 
    Example — filter captured logs after a request:
 
    @snippet samples/testsuite-support/tests/test_logcapture.py select
 
    Example — subscribe to log events as they arrive:
 
    @snippet samples/testsuite-support/tests/test_logcapture.py subscribe
 
    @ingroup userver_testsuite_fixtures
    """
    host = pytestconfig.option.logs_capture_host
    port = pytestconfig.option.logs_capture_port
    if pytestconfig.option.service_wait or pytestconfig.option.service_disable:
        port = port or DEFAULT_PORT
 
    server = logcapture.CaptureServer(
        log_level=logcapture.LogLevel.from_string(userver_log_level),
        parse_line=_tskv_parse_line,
    )
    async with server.start(host=host, port=port):
        yield server
 
 
@pytest.fixture(scope='session')
def _userver_config_logs_capture(userver_log_capture):
    socknames = userver_log_capture.getsocknames()
    assert socknames
    sockname = socknames[0]
 
    def patch_config(config: dict, _config_vars: dict) -> None:
        logging_config = config['components_manager']['components']['logging']
        default_logger = logging_config['loggers']['default']
        # Other formats are not yet supported by log-capture.
        default_logger['format'] = 'tskv'
        default_logger['testsuite-capture'] = {
            'host': sockname[0],
            'port': sockname[1],
        }
 
    return patch_config
 
 
def _tskv_parse_line(rawline: bytes) -> tskv.TskvRow:
    line = rawline.decode(encoding='utf-8', errors='replace')
    return tskv.parse_line(line)