userver: /home/antonyzhilin/arcadia/taxi/uservices/userver/testsuite/pytest_plugins/pytest_userver/plugins/log_capture.py Source File
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
log_capture.py
1"""
2Capture and work with logs.
3"""
4
5# pylint: disable=redefined-outer-name
6import logging
7
8import pytest
9
10from testsuite import logcapture
11from testsuite.logcapture import __tracebackhide__ # noqa
12
13from ..utils import tskv
14
15USERVER_CONFIG_HOOKS = ['_userver_config_logs_capture']
16DEFAULT_PORT = 2211
17
18logger = logging.getLogger(__name__)
19
20
21def pytest_addoption(parser):
22 group = parser.getgroup('logs-capture')
23 group.addoption(
24 '--logs-capture-port',
25 type=int,
26 default=0,
27 help='Port to bind logs-capture server to.',
28 )
29 group.addoption(
30 '--logs-capture-host',
31 default='localhost',
32 help='Host to bind logs-capture server to.',
33 )
34
35
36@pytest.fixture(scope='session')
37async def userver_log_capture(pytestconfig, userver_log_level):
38 host = pytestconfig.option.logs_capture_host
39 port = pytestconfig.option.logs_capture_port
40 if pytestconfig.option.service_wait or pytestconfig.option.service_disable:
41 port = port or DEFAULT_PORT
42
43 server = logcapture.CaptureServer(
44 log_level=logcapture.LogLevel.from_string(userver_log_level),
45 parse_line=_tskv_parse_line,
46 )
47 async with server.start(host=host, port=port):
48 yield server
49
50
51@pytest.fixture(scope='session')
52def _userver_config_logs_capture(userver_log_capture):
53 socknames = userver_log_capture.getsocknames()
54 assert socknames
55 sockname = socknames[0]
56
57 def patch_config(config, _config_vars) -> None:
58 logging_config = config['components_manager']['components']['logging']
59 default_logger = logging_config['loggers']['default']
60 # Other formats are not yet supported by log-capture.
61 default_logger['format'] = 'tskv'
62 default_logger['testsuite-capture'] = {
63 'host': sockname[0],
64 'port': sockname[1],
65 }
66
67 return patch_config
68
69
70def _tskv_parse_line(rawline: bytes):
71 line = rawline.decode('utf-8')
72 return tskv.parse_line(line)