14logger = logging.getLogger(__name__)
18 def __init__(self, path: pathlib.Path):
22 def update_position(self):
25 except FileNotFoundError:
34 eof_handler: typing.Optional[typing.Callable[[], bool]] =
None,
35 limit_position: bool =
False,
38 max_position = self.
path.stat().st_size
42 for line, position
in _raw_line_reader(
45 eof_handler=eof_handler,
48 line = line.decode(encoding=
'utf-8', errors=
'backslashreplace')
51 if not line.startswith(
'tskv\t'):
53 if not line.endswith(
'\n'):
57 if max_position
is not None and position >= max_position:
62 def __init__(self, *, colorize_factory, delay: float = 0.05):
68 def register_logfile(self, path: pathlib.Path):
75 def join(self, timeout: float = 10):
77 for thread
in self.
_threads.values():
78 thread.join(timeout=timeout)
80 def _logreader_thread(self, path: pathlib.Path):
82 while not path.exists():
87 for line
in logfile.readlines(eof_handler=self.
_eof_handler):
88 line = line.rstrip(
'\r\n')
89 line = colorizer(line)
93 def _write_logline(self, line: str):
94 print(line, file=sys.stderr)
96 def _eof_handler(self) -> bool:
106 def __init__(self, *, colorize_factory, config):
112 def pytest_sessionstart(self, session):
113 if _is_live_logs_enabled(self.
_config):
120 def pytest_sessionfinish(self, session):
124 def pytest_runtest_setup(self, item):
128 @pytest.hookimpl(wrapper=True, tryfirst=True)
129 def pytest_runtest_makereport(self, item, call):
135 def update_position(self):
136 for logfile
in self.
_logs.values():
137 logfile.update_position()
139 def register_flusher(self, func):
140 loop = asyncio.get_running_loop()
143 def register_logfile(self, path: pathlib.Path, title: str):
144 logger.info(
'Watching service log file: %s', path)
146 self.
_logs[path, title] = logfile
150 def _userver_report_attach(self, report):
152 for (_, title), logfile
in self.
_logs.items():
155 def _userver_report_attach_log(self, logfile: LogFile, report, title):
158 for line
in logfile.readlines(limit_position=
True):
159 line = line.rstrip(
'\r\n')
160 line = colorizer(line)
164 value = log.getvalue()
166 report.sections.append((f
'Captured {title} {report.when}', value))
168 def _run_flushers(self):
171 loop.run_until_complete(flusher())
173 logger.exception(
'failed to run logging flushers:')
176@pytest.fixture(scope='session')
177def service_logfile_path(
179 service_tmpdir: pathlib.Path,
180) -> typing.Optional[pathlib.Path]:
182 Holds optional service logfile path. You may want to override this
185 By default returns value of --service-logs-file option or creates
188 if pytestconfig.option.service_logs_file:
189 return pytestconfig.option.service_logs_file
190 return service_tmpdir /
'service.log'
193@pytest.fixture(scope='session')
194def _service_logfile_path(
195 userver_register_logfile,
196 service_logfile_path: typing.Optional[pathlib.Path],
197) -> typing.Optional[pathlib.Path]:
198 if not service_logfile_path:
200 userver_register_logfile(
201 service_logfile_path,
205 return service_logfile_path
208@pytest.fixture(scope='session')
209def userver_register_logfile(_userver_logging_plugin: UserverLoggingPlugin):
211 Register logfile. Registered logfile is monitored in case of test failure
212 and its contents is attached to pytest report.
214 :param path: pathlib.Path corresponding to log file
215 :param title: title to be used in pytest report
216 :param truncate: file is truncated if True
219 def register_logfile(
220 path: pathlib.Path, *, title: str, truncate: bool = False,
225 def do_truncate(path):
226 with path.open(
'wb+')
as fp:
229 def register_logfile(
233 truncate: bool =
False,
237 _userver_logging_plugin.register_logfile(path, title)
240 return register_logfile
243@pytest.fixture(scope='session')
244def _userver_logging_plugin(pytestconfig) -> UserverLoggingPlugin:
245 return pytestconfig.pluginmanager.get_plugin(
'userver_logging')
248def pytest_configure(config):
249 pretty_logs = config.option.service_logs_pretty
250 colors_enabled = _should_enable_color(config)
251 verbose = pretty_logs ==
'verbose'
253 def colorize_factory():
257 colors_enabled=colors_enabled,
259 return colorizer.colorize_line
261 def handle_line(line):
267 colorize_factory=colorize_factory,
270 config.pluginmanager.register(plugin,
'userver_logging')
273def pytest_report_header(config):
275 if config.option.service_logs_file:
276 headers.append(f
'servicelogs: {config.option.service_logs_file}')
280def pytest_terminal_summary(terminalreporter, config) -> None:
281 logfile = config.option.service_logs_file
283 terminalreporter.ensure_newline()
284 terminalreporter.section(
'Service logs', sep=
'-', blue=
True, bold=
True)
285 terminalreporter.write_sep(
'-', f
'service log file: {logfile}')
288def _should_enable_color(pytestconfig) -> bool:
289 option = getattr(pytestconfig.option,
'color',
'no')
293 return sys.stderr.isatty()
300 eof_handler: typing.Optional[typing.Callable[[], bool]] =
None,
302 with path.open(
'rb')
as fp:
303 position = fp.seek(position)
308 line = partial + line
310 if line.endswith(b
'\n'):
311 position += len(line)
321def _is_live_logs_enabled(config):
322 if not config.option.service_live_logs_disable:
324 config.option.capture ==
'no' and config.option.showcapture
in (
'all',
'log'),