13logger = logging.getLogger(__name__)
17 def __init__(self, path: pathlib.Path):
21 def update_position(self):
24 except FileNotFoundError:
33 eof_handler: typing.Optional[typing.Callable[[], bool]] =
None,
36 for line
in _raw_line_reader(
40 line = line.decode(encoding=
'utf-8', errors=
'backslashreplace')
43 if not line.startswith(
'tskv '):
45 if not line.endswith(
'\n'):
52 def __init__(self, *, colorize_factory, delay: float = 0.05):
58 def register_logfile(self, path: pathlib.Path):
65 def join(self, timeout: float = 10):
67 for thread
in self.
_threads.values():
68 thread.join(timeout=timeout)
70 def _logreader_thread(self, path: pathlib.Path):
72 while not path.exists():
77 for line
in logfile.readlines(eof_handler=self.
_eof_handler):
78 line = line.rstrip(
'\r\n')
79 line = colorizer(line)
83 def _write_logline(self, line: str):
84 print(line, file=sys.stderr)
86 def _eof_handler(self) -> bool:
96 def __init__(self, *, colorize_factory, live_logs_enabled: bool =
False):
102 def pytest_sessionfinish(self, session):
106 def pytest_runtest_setup(self, item):
107 for logfile
in self.
_logs.values():
108 logfile.update_position()
110 @pytest.hookimpl(wrapper=True)
111 def pytest_runtest_teardown(self, item):
114 def register_logfile(self, path: pathlib.Path, title: str):
115 logger.info(
'Watching service log file: %s', path)
117 self.
_logs[path, title] = logfile
121 def _userver_log_dump(self, item, when):
127 if item.utestsuite_report.failed:
130 def _userver_report_attach(self, item, when):
131 for (_, title), logfile
in self.
_logs.items():
134 def _userver_report_attach_log(self, logfile: LogFile, item, when, title):
135 report = io.StringIO()
137 for line
in logfile.readlines():
138 line = line.rstrip(
'\r\n')
139 line = colorizer(line)
143 value = report.getvalue()
145 item.add_report_section(when, title, value)
148@pytest.fixture(scope='session')
149def service_logfile_path(
150 pytestconfig, service_tmpdir: pathlib.Path,
151) -> typing.Optional[pathlib.Path]:
153 Holds optional service logfile path. You may want to override this
156 By default returns value of --service-logs-file option or creates
159 if pytestconfig.option.service_logs_file:
160 return pytestconfig.option.service_logs_file
161 return service_tmpdir /
'service.log'
164@pytest.fixture(scope='session')
165def _service_logfile_path(
166 userver_register_logfile,
167 service_logfile_path: typing.Optional[pathlib.Path],
168) -> typing.Optional[pathlib.Path]:
169 if not service_logfile_path:
171 userver_register_logfile(
172 service_logfile_path, title=
'userver/log', truncate=
True,
174 return service_logfile_path
177@pytest.fixture(scope='session')
178def userver_register_logfile(_userver_logging_plugin: UserverLoggingPlugin):
180 Register logfile. Registered logfile is monitored in case of test failure
181 and its contents is attached to pytest report.
183 :param path: pathlib.Path corresponding to log file
184 :param title: title to be used in pytest report
185 :param truncate: file is truncated if True
188 def register_logfile(
189 path: pathlib.Path, *, title: str, truncate: bool = False,
194 def do_truncate(path):
195 with path.open(
'wb+')
as fp:
198 def register_logfile(
199 path: pathlib.Path, *, title: str, truncate: bool =
False,
203 _userver_logging_plugin.register_logfile(path, title)
206 return register_logfile
209@pytest.fixture(scope='session')
210def _userver_logging_plugin(pytestconfig) -> UserverLoggingPlugin:
211 return pytestconfig.pluginmanager.get_plugin(
'userver_logging')
214def pytest_configure(config):
215 live_logs_enabled = bool(
216 config.option.capture ==
'no'
217 and config.option.showcapture
in (
'all',
'log'),
220 pretty_logs = config.option.service_logs_pretty
221 colors_enabled = _should_enable_color(config)
222 verbose = pretty_logs ==
'verbose'
224 def colorize_factory():
226 colorizer = colorize.Colorizer(
227 verbose=verbose, colors_enabled=colors_enabled,
229 return colorizer.colorize_line
231 def handle_line(line):
237 colorize_factory=colorize_factory, live_logs_enabled=live_logs_enabled,
239 config.pluginmanager.register(plugin,
'userver_logging')
242def pytest_report_header(config):
244 if config.option.service_logs_file:
245 headers.append(f
'servicelogs: {config.option.service_logs_file}')
249def pytest_terminal_summary(terminalreporter, config) -> None:
250 logfile = config.option.service_logs_file
252 terminalreporter.ensure_newline()
253 terminalreporter.section(
'Service logs', sep=
'-', blue=
True, bold=
True)
254 terminalreporter.write_sep(
'-', f
'service log file: {logfile}')
257def _should_enable_color(pytestconfig) -> bool:
258 option = getattr(pytestconfig.option,
'color',
'no')
262 return sys.stderr.isatty()
269 eof_handler: typing.Optional[typing.Callable[[], bool]] =
None,
270 bufsize: int = 16384,
273 with path.open(
'rb')
as fp:
276 chunk = fp.read(bufsize)
284 lines = buf.splitlines(keepends=
True)
285 if not lines[-1].endswith(b
'\n'):