userver: /home/antonyzhilin/arcadia/taxi/uservices/userver/testsuite/pytest_plugins/pytest_userver/plugins/logging.py Source File
Loading...
Searching...
No Matches
logging.py
1import asyncio
2import io
3import logging
4import pathlib
5import sys
6import threading
7import time
8import typing
9
10import pytest
11
12from pytest_userver.utils import colorize
13
14logger = logging.getLogger(__name__)
15
16
17class LogFile:
18 def __init__(self, path: pathlib.Path):
19 self.path = path
20 self.position = 0
21
22 def update_position(self):
23 try:
24 st = self.path.stat()
25 except FileNotFoundError:
26 pos = 0
27 else:
28 pos = st.st_size
29 self.position = pos
30 return pos
31
32 def readlines(
33 self,
34 eof_handler: typing.Optional[typing.Callable[[], bool]] = None,
35 limit_position: bool = False,
36 ):
37 if limit_position:
38 max_position = self.path.stat().st_size
39 else:
40 max_position = None
41 first_skipped = False
42 for line, position in _raw_line_reader(
43 self.path,
44 self.position,
45 eof_handler=eof_handler,
46 ):
47 # userver does not give any guarantees about log file encoding
48 line = line.decode(encoding='utf-8', errors='backslashreplace')
49 if not first_skipped:
50 first_skipped = True
51 if not line.startswith('tskv\t'):
52 continue
53 if not line.endswith('\n'):
54 continue
55 self.position = position
56 yield line
57 if max_position is not None and position >= max_position:
58 break
59
60
62 def __init__(self, *, colorize_factory, delay: float = 0.05):
63 self._colorize_factory = colorize_factory
64 self._threads = {}
65 self._exiting = False
66 self._delay = delay
67
68 def register_logfile(self, path: pathlib.Path):
69 if path in self._threads:
70 return
71 thread = threading.Thread(target=self._logreader_thread, args=(path,))
72 self._threads[path] = thread
73 thread.start()
74
75 def join(self, timeout: float = 10):
76 self._exiting = True
77 for thread in self._threads.values():
78 thread.join(timeout=timeout)
79
80 def _logreader_thread(self, path: pathlib.Path):
81 # wait for file to appear
82 while not path.exists():
83 if self._eof_handler():
84 return
85 colorizer = self._colorize_factory()
86 logfile = LogFile(path)
87 for line in logfile.readlines(eof_handler=self._eof_handler):
88 line = line.rstrip('\r\n')
89 line = colorizer(line)
90 if line:
91 self._write_logline(line)
92
93 def _write_logline(self, line: str):
94 print(line, file=sys.stderr)
95
96 def _eof_handler(self) -> bool:
97 if self._exiting:
98 return True
99 time.sleep(self._delay)
100 return False
101
102
104 _live_logs = None
105
106 def __init__(self, *, colorize_factory, config):
107 self._colorize_factory = colorize_factory
108 self._config = config
109 self._logs = {}
110 self._flushers = []
111
112 def pytest_sessionstart(self, session):
113 if _is_live_logs_enabled(self._config):
115 colorize_factory=self._colorize_factory,
116 )
117 else:
118 self._live_logs = None
119
120 def pytest_sessionfinish(self, session):
121 if self._live_logs:
122 self._live_logs.join()
123
124 def pytest_runtest_setup(self, item):
125 self._flushers.clear()
126 self.update_position()
127
128 @pytest.hookimpl(wrapper=True, tryfirst=True)
129 def pytest_runtest_makereport(self, item, call):
130 report = yield
131 if report.failed and not self._live_logs:
132 self._userver_report_attach(report)
133 return report
134
135 def update_position(self):
136 for logfile in self._logs.values():
137 logfile.update_position()
138
139 def register_flusher(self, func):
140 loop = asyncio.get_running_loop()
141 self._flushers.append((loop, func))
142
143 def register_logfile(self, path: pathlib.Path, title: str):
144 logger.info('Watching service log file: %s', path)
145 logfile = LogFile(path)
146 self._logs[path, title] = logfile
147 if self._live_logs:
148 self._live_logs.register_logfile(path)
149
150 def _userver_report_attach(self, report):
151 self._run_flushers()
152 for (_, title), logfile in self._logs.items():
153 self._userver_report_attach_log(logfile, report, title)
154
155 def _userver_report_attach_log(self, logfile: LogFile, report, title):
156 log = io.StringIO()
157 colorizer = self._colorize_factory()
158 for line in logfile.readlines(limit_position=True):
159 line = line.rstrip('\r\n')
160 line = colorizer(line)
161 if line:
162 log.write(line)
163 log.write('\n')
164 value = log.getvalue()
165 if value:
166 report.sections.append((f'Captured {title} {report.when}', value))
167
168 def _run_flushers(self):
169 try:
170 for loop, flusher in self._flushers:
171 loop.run_until_complete(flusher())
172 except Exception:
173 logger.exception('failed to run logging flushers:')
174
175
176@pytest.fixture(scope='session')
177def service_logfile_path(
178 pytestconfig,
179 service_tmpdir: pathlib.Path,
180) -> typing.Optional[pathlib.Path]:
181 """
182 Holds optional service logfile path. You may want to override this
183 in your service.
184
185 By default returns value of --service-logs-file option or creates
186 temporary file.
187 """
188 if pytestconfig.option.service_logs_file:
189 return pytestconfig.option.service_logs_file
190 return service_tmpdir / 'service.log'
191
192
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:
199 return None
200 userver_register_logfile(
201 service_logfile_path,
202 title='userver/log',
203 truncate=True,
204 )
205 return service_logfile_path
206
207
208@pytest.fixture(scope='session')
209def userver_register_logfile(_userver_logging_plugin: UserverLoggingPlugin):
210 """
211 Register logfile. Registered logfile is monitored in case of test failure
212 and its contents is attached to pytest report.
213
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
217
218 ```python
219 def register_logfile(
220 path: pathlib.Path, *, title: str, truncate: bool = False,
221 ) -> None:
222 ```
223 """
224
225 def do_truncate(path):
226 with path.open('wb+') as fp:
227 fp.truncate()
228
229 def register_logfile(
230 path: pathlib.Path,
231 *,
232 title: str,
233 truncate: bool = False,
234 ) -> None:
235 if truncate:
236 do_truncate(path)
237 _userver_logging_plugin.register_logfile(path, title)
238 return path
239
240 return register_logfile
241
242
243@pytest.fixture(scope='session')
244def _userver_logging_plugin(pytestconfig) -> UserverLoggingPlugin:
245 return pytestconfig.pluginmanager.get_plugin('userver_logging')
246
247
248def pytest_configure(config):
249 pretty_logs = config.option.service_logs_pretty
250 colors_enabled = _should_enable_color(config)
251 verbose = pretty_logs == 'verbose'
252
253 def colorize_factory():
254 if pretty_logs:
255 colorizer = colorize.Colorizer(
256 verbose=verbose,
257 colors_enabled=colors_enabled,
258 )
259 return colorizer.colorize_line
260
261 def handle_line(line):
262 return line
263
264 return handle_line
265
266 plugin = UserverLoggingPlugin(
267 colorize_factory=colorize_factory,
268 config=config,
269 )
270 config.pluginmanager.register(plugin, 'userver_logging')
271
272
273def pytest_report_header(config):
274 headers = []
275 if config.option.service_logs_file:
276 headers.append(f'servicelogs: {config.option.service_logs_file}')
277 return headers
278
279
280def pytest_terminal_summary(terminalreporter, config) -> None:
281 logfile = config.option.service_logs_file
282 if logfile:
283 terminalreporter.ensure_newline()
284 terminalreporter.section('Service logs', sep='-', blue=True, bold=True)
285 terminalreporter.write_sep('-', f'service log file: {logfile}')
286
287
288def _should_enable_color(pytestconfig) -> bool:
289 option = getattr(pytestconfig.option, 'color', 'no')
290 if option == 'yes':
291 return True
292 if option == 'auto':
293 return sys.stderr.isatty()
294 return False
295
296
297def _raw_line_reader(
298 path: pathlib.Path,
299 position: int = 0,
300 eof_handler: typing.Optional[typing.Callable[[], bool]] = None,
301) -> int:
302 with path.open('rb') as fp:
303 position = fp.seek(position)
304 while True:
305 partial = None
306 for line in fp:
307 if partial:
308 line = partial + line
309 partial = None
310 if line.endswith(b'\n'):
311 position += len(line)
312 yield line, position
313 else:
314 partial = line
315 if not eof_handler:
316 break
317 if eof_handler():
318 break
319
320
321def _is_live_logs_enabled(config):
322 if not config.option.service_live_logs_disable:
323 return bool(
324 config.option.capture == 'no' and config.option.showcapture in ('all', 'log'),
325 )
326 return False