userver: /data/code/userver/testsuite/pytest_plugins/pytest_userver/plugins/logging.py Source File
⚠️ This is the documentation for an old userver version. Click here to switch to the latest version.
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages Concepts
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