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