logging.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960
  1. # mypy: allow-untyped-defs
  2. """Access and control log capturing."""
  3. from __future__ import annotations
  4. from collections.abc import Generator
  5. from collections.abc import Mapping
  6. from collections.abc import Set as AbstractSet
  7. from contextlib import contextmanager
  8. from contextlib import nullcontext
  9. from datetime import datetime
  10. from datetime import timedelta
  11. from datetime import timezone
  12. import io
  13. from io import StringIO
  14. import logging
  15. from logging import LogRecord
  16. import os
  17. from pathlib import Path
  18. import re
  19. from types import TracebackType
  20. from typing import final
  21. from typing import Generic
  22. from typing import Literal
  23. from typing import TYPE_CHECKING
  24. from typing import TypeVar
  25. from _pytest import nodes
  26. from _pytest._io import TerminalWriter
  27. from _pytest.capture import CaptureManager
  28. from _pytest.config import _strtobool
  29. from _pytest.config import Config
  30. from _pytest.config import create_terminal_writer
  31. from _pytest.config import hookimpl
  32. from _pytest.config import UsageError
  33. from _pytest.config.argparsing import Parser
  34. from _pytest.deprecated import check_ispytest
  35. from _pytest.fixtures import fixture
  36. from _pytest.fixtures import FixtureRequest
  37. from _pytest.main import Session
  38. from _pytest.stash import StashKey
  39. from _pytest.terminal import TerminalReporter
  40. if TYPE_CHECKING:
  41. logging_StreamHandler = logging.StreamHandler[StringIO]
  42. else:
  43. logging_StreamHandler = logging.StreamHandler
  44. DEFAULT_LOG_FORMAT = "%(levelname)-8s %(name)s:%(filename)s:%(lineno)d %(message)s"
  45. DEFAULT_LOG_DATE_FORMAT = "%H:%M:%S"
  46. _ANSI_ESCAPE_SEQ = re.compile(r"\x1b\[[\d;]+m")
  47. caplog_handler_key = StashKey["LogCaptureHandler"]()
  48. caplog_records_key = StashKey[dict[str, list[logging.LogRecord]]]()
  49. def _remove_ansi_escape_sequences(text: str) -> str:
  50. return _ANSI_ESCAPE_SEQ.sub("", text)
  51. class DatetimeFormatter(logging.Formatter):
  52. """A logging formatter which formats record with
  53. :func:`datetime.datetime.strftime` formatter instead of
  54. :func:`time.strftime` in case of microseconds in format string.
  55. """
  56. def formatTime(self, record: LogRecord, datefmt: str | None = None) -> str:
  57. if datefmt and "%f" in datefmt:
  58. ct = self.converter(record.created)
  59. tz = timezone(timedelta(seconds=ct.tm_gmtoff), ct.tm_zone)
  60. # Construct `datetime.datetime` object from `struct_time`
  61. # and msecs information from `record`
  62. # Using int() instead of round() to avoid it exceeding 1_000_000 and causing a ValueError (#11861).
  63. dt = datetime(*ct[0:6], microsecond=int(record.msecs * 1000), tzinfo=tz)
  64. return dt.strftime(datefmt)
  65. # Use `logging.Formatter` for non-microsecond formats
  66. return super().formatTime(record, datefmt)
  67. class ColoredLevelFormatter(DatetimeFormatter):
  68. """A logging formatter which colorizes the %(levelname)..s part of the
  69. log format passed to __init__."""
  70. LOGLEVEL_COLOROPTS: Mapping[int, AbstractSet[str]] = {
  71. logging.CRITICAL: {"red"},
  72. logging.ERROR: {"red", "bold"},
  73. logging.WARNING: {"yellow"},
  74. logging.WARN: {"yellow"},
  75. logging.INFO: {"green"},
  76. logging.DEBUG: {"purple"},
  77. logging.NOTSET: set(),
  78. }
  79. LEVELNAME_FMT_REGEX = re.compile(r"%\(levelname\)([+-.]?\d*(?:\.\d+)?s)")
  80. def __init__(self, terminalwriter: TerminalWriter, *args, **kwargs) -> None:
  81. super().__init__(*args, **kwargs)
  82. self._terminalwriter = terminalwriter
  83. self._original_fmt = self._style._fmt
  84. self._level_to_fmt_mapping: dict[int, str] = {}
  85. for level, color_opts in self.LOGLEVEL_COLOROPTS.items():
  86. self.add_color_level(level, *color_opts)
  87. def add_color_level(self, level: int, *color_opts: str) -> None:
  88. """Add or update color opts for a log level.
  89. :param level:
  90. Log level to apply a style to, e.g. ``logging.INFO``.
  91. :param color_opts:
  92. ANSI escape sequence color options. Capitalized colors indicates
  93. background color, i.e. ``'green', 'Yellow', 'bold'`` will give bold
  94. green text on yellow background.
  95. .. warning::
  96. This is an experimental API.
  97. """
  98. assert self._fmt is not None
  99. levelname_fmt_match = self.LEVELNAME_FMT_REGEX.search(self._fmt)
  100. if not levelname_fmt_match:
  101. return
  102. levelname_fmt = levelname_fmt_match.group()
  103. formatted_levelname = levelname_fmt % {"levelname": logging.getLevelName(level)}
  104. # add ANSI escape sequences around the formatted levelname
  105. color_kwargs = {name: True for name in color_opts}
  106. colorized_formatted_levelname = self._terminalwriter.markup(
  107. formatted_levelname, **color_kwargs
  108. )
  109. self._level_to_fmt_mapping[level] = self.LEVELNAME_FMT_REGEX.sub(
  110. colorized_formatted_levelname, self._fmt
  111. )
  112. def format(self, record: logging.LogRecord) -> str:
  113. fmt = self._level_to_fmt_mapping.get(record.levelno, self._original_fmt)
  114. self._style._fmt = fmt
  115. return super().format(record)
  116. class PercentStyleMultiline(logging.PercentStyle):
  117. """A logging style with special support for multiline messages.
  118. If the message of a record consists of multiple lines, this style
  119. formats the message as if each line were logged separately.
  120. """
  121. def __init__(self, fmt: str, auto_indent: int | str | bool | None) -> None:
  122. super().__init__(fmt)
  123. self._auto_indent = self._get_auto_indent(auto_indent)
  124. @staticmethod
  125. def _get_auto_indent(auto_indent_option: int | str | bool | None) -> int:
  126. """Determine the current auto indentation setting.
  127. Specify auto indent behavior (on/off/fixed) by passing in
  128. extra={"auto_indent": [value]} to the call to logging.log() or
  129. using a --log-auto-indent [value] command line or the
  130. log_auto_indent [value] config option.
  131. Default behavior is auto-indent off.
  132. Using the string "True" or "on" or the boolean True as the value
  133. turns auto indent on, using the string "False" or "off" or the
  134. boolean False or the int 0 turns it off, and specifying a
  135. positive integer fixes the indentation position to the value
  136. specified.
  137. Any other values for the option are invalid, and will silently be
  138. converted to the default.
  139. :param None|bool|int|str auto_indent_option:
  140. User specified option for indentation from command line, config
  141. or extra kwarg. Accepts int, bool or str. str option accepts the
  142. same range of values as boolean config options, as well as
  143. positive integers represented in str form.
  144. :returns:
  145. Indentation value, which can be
  146. -1 (automatically determine indentation) or
  147. 0 (auto-indent turned off) or
  148. >0 (explicitly set indentation position).
  149. """
  150. if auto_indent_option is None:
  151. return 0
  152. elif isinstance(auto_indent_option, bool):
  153. if auto_indent_option:
  154. return -1
  155. else:
  156. return 0
  157. elif isinstance(auto_indent_option, int):
  158. return int(auto_indent_option)
  159. elif isinstance(auto_indent_option, str):
  160. try:
  161. return int(auto_indent_option)
  162. except ValueError:
  163. pass
  164. try:
  165. if _strtobool(auto_indent_option):
  166. return -1
  167. except ValueError:
  168. return 0
  169. return 0
  170. def format(self, record: logging.LogRecord) -> str:
  171. if "\n" in record.message:
  172. if hasattr(record, "auto_indent"):
  173. # Passed in from the "extra={}" kwarg on the call to logging.log().
  174. auto_indent = self._get_auto_indent(record.auto_indent)
  175. else:
  176. auto_indent = self._auto_indent
  177. if auto_indent:
  178. lines = record.message.splitlines()
  179. formatted = self._fmt % {**record.__dict__, "message": lines[0]}
  180. if auto_indent < 0:
  181. indentation = _remove_ansi_escape_sequences(formatted).find(
  182. lines[0]
  183. )
  184. else:
  185. # Optimizes logging by allowing a fixed indentation.
  186. indentation = auto_indent
  187. lines[0] = formatted
  188. return ("\n" + " " * indentation).join(lines)
  189. return self._fmt % record.__dict__
  190. def get_option_ini(config: Config, *names: str):
  191. for name in names:
  192. ret = config.getoption(name) # 'default' arg won't work as expected
  193. if ret is None:
  194. ret = config.getini(name)
  195. if ret:
  196. return ret
  197. def pytest_addoption(parser: Parser) -> None:
  198. """Add options to control log capturing."""
  199. group = parser.getgroup("logging")
  200. def add_option_ini(option, dest, default=None, type=None, **kwargs):
  201. parser.addini(
  202. dest, default=default, type=type, help="Default value for " + option
  203. )
  204. group.addoption(option, dest=dest, **kwargs)
  205. add_option_ini(
  206. "--log-level",
  207. dest="log_level",
  208. default=None,
  209. metavar="LEVEL",
  210. help=(
  211. "Level of messages to catch/display."
  212. " Not set by default, so it depends on the root/parent log handler's"
  213. ' effective level, where it is "WARNING" by default.'
  214. ),
  215. )
  216. add_option_ini(
  217. "--log-format",
  218. dest="log_format",
  219. default=DEFAULT_LOG_FORMAT,
  220. help="Log format used by the logging module",
  221. )
  222. add_option_ini(
  223. "--log-date-format",
  224. dest="log_date_format",
  225. default=DEFAULT_LOG_DATE_FORMAT,
  226. help="Log date format used by the logging module",
  227. )
  228. parser.addini(
  229. "log_cli",
  230. default=False,
  231. type="bool",
  232. help='Enable log display during test run (also known as "live logging")',
  233. )
  234. add_option_ini(
  235. "--log-cli-level", dest="log_cli_level", default=None, help="CLI logging level"
  236. )
  237. add_option_ini(
  238. "--log-cli-format",
  239. dest="log_cli_format",
  240. default=None,
  241. help="Log format used by the logging module",
  242. )
  243. add_option_ini(
  244. "--log-cli-date-format",
  245. dest="log_cli_date_format",
  246. default=None,
  247. help="Log date format used by the logging module",
  248. )
  249. add_option_ini(
  250. "--log-file",
  251. dest="log_file",
  252. default=None,
  253. help="Path to a file when logging will be written to",
  254. )
  255. add_option_ini(
  256. "--log-file-mode",
  257. dest="log_file_mode",
  258. default="w",
  259. choices=["w", "a"],
  260. help="Log file open mode",
  261. )
  262. add_option_ini(
  263. "--log-file-level",
  264. dest="log_file_level",
  265. default=None,
  266. help="Log file logging level",
  267. )
  268. add_option_ini(
  269. "--log-file-format",
  270. dest="log_file_format",
  271. default=None,
  272. help="Log format used by the logging module",
  273. )
  274. add_option_ini(
  275. "--log-file-date-format",
  276. dest="log_file_date_format",
  277. default=None,
  278. help="Log date format used by the logging module",
  279. )
  280. add_option_ini(
  281. "--log-auto-indent",
  282. dest="log_auto_indent",
  283. default=None,
  284. help="Auto-indent multiline messages passed to the logging module. Accepts true|on, false|off or an integer.",
  285. )
  286. group.addoption(
  287. "--log-disable",
  288. action="append",
  289. default=[],
  290. dest="logger_disable",
  291. help="Disable a logger by name. Can be passed multiple times.",
  292. )
  293. _HandlerType = TypeVar("_HandlerType", bound=logging.Handler)
  294. # Not using @contextmanager for performance reasons.
  295. class catching_logs(Generic[_HandlerType]):
  296. """Context manager that prepares the whole logging machinery properly."""
  297. __slots__ = ("handler", "level", "orig_level")
  298. def __init__(self, handler: _HandlerType, level: int | None = None) -> None:
  299. self.handler = handler
  300. self.level = level
  301. def __enter__(self) -> _HandlerType:
  302. root_logger = logging.getLogger()
  303. if self.level is not None:
  304. self.handler.setLevel(self.level)
  305. root_logger.addHandler(self.handler)
  306. if self.level is not None:
  307. self.orig_level = root_logger.level
  308. root_logger.setLevel(min(self.orig_level, self.level))
  309. return self.handler
  310. def __exit__(
  311. self,
  312. exc_type: type[BaseException] | None,
  313. exc_val: BaseException | None,
  314. exc_tb: TracebackType | None,
  315. ) -> None:
  316. root_logger = logging.getLogger()
  317. if self.level is not None:
  318. root_logger.setLevel(self.orig_level)
  319. root_logger.removeHandler(self.handler)
  320. class LogCaptureHandler(logging_StreamHandler):
  321. """A logging handler that stores log records and the log text."""
  322. def __init__(self) -> None:
  323. """Create a new log handler."""
  324. super().__init__(StringIO())
  325. self.records: list[logging.LogRecord] = []
  326. def emit(self, record: logging.LogRecord) -> None:
  327. """Keep the log records in a list in addition to the log text."""
  328. self.records.append(record)
  329. super().emit(record)
  330. def reset(self) -> None:
  331. self.records = []
  332. self.stream = StringIO()
  333. def clear(self) -> None:
  334. self.records.clear()
  335. self.stream = StringIO()
  336. def handleError(self, record: logging.LogRecord) -> None:
  337. if logging.raiseExceptions:
  338. # Fail the test if the log message is bad (emit failed).
  339. # The default behavior of logging is to print "Logging error"
  340. # to stderr with the call stack and some extra details.
  341. # pytest wants to make such mistakes visible during testing.
  342. raise # noqa: PLE0704
  343. @final
  344. class LogCaptureFixture:
  345. """Provides access and control of log capturing."""
  346. def __init__(self, item: nodes.Node, *, _ispytest: bool = False) -> None:
  347. check_ispytest(_ispytest)
  348. self._item = item
  349. self._initial_handler_level: int | None = None
  350. # Dict of log name -> log level.
  351. self._initial_logger_levels: dict[str | None, int] = {}
  352. self._initial_disabled_logging_level: int | None = None
  353. def _finalize(self) -> None:
  354. """Finalize the fixture.
  355. This restores the log levels and the disabled logging levels changed by :meth:`set_level`.
  356. """
  357. # Restore log levels.
  358. if self._initial_handler_level is not None:
  359. self.handler.setLevel(self._initial_handler_level)
  360. for logger_name, level in self._initial_logger_levels.items():
  361. logger = logging.getLogger(logger_name)
  362. logger.setLevel(level)
  363. # Disable logging at the original disabled logging level.
  364. if self._initial_disabled_logging_level is not None:
  365. logging.disable(self._initial_disabled_logging_level)
  366. self._initial_disabled_logging_level = None
  367. @property
  368. def handler(self) -> LogCaptureHandler:
  369. """Get the logging handler used by the fixture."""
  370. return self._item.stash[caplog_handler_key]
  371. def get_records(
  372. self, when: Literal["setup", "call", "teardown"]
  373. ) -> list[logging.LogRecord]:
  374. """Get the logging records for one of the possible test phases.
  375. :param when:
  376. Which test phase to obtain the records from.
  377. Valid values are: "setup", "call" and "teardown".
  378. :returns: The list of captured records at the given stage.
  379. .. versionadded:: 3.4
  380. """
  381. return self._item.stash[caplog_records_key].get(when, [])
  382. @property
  383. def text(self) -> str:
  384. """The formatted log text."""
  385. return _remove_ansi_escape_sequences(self.handler.stream.getvalue())
  386. @property
  387. def records(self) -> list[logging.LogRecord]:
  388. """The list of log records."""
  389. return self.handler.records
  390. @property
  391. def record_tuples(self) -> list[tuple[str, int, str]]:
  392. """A list of a stripped down version of log records intended
  393. for use in assertion comparison.
  394. The format of the tuple is:
  395. (logger_name, log_level, message)
  396. """
  397. return [(r.name, r.levelno, r.getMessage()) for r in self.records]
  398. @property
  399. def messages(self) -> list[str]:
  400. """A list of format-interpolated log messages.
  401. Unlike 'records', which contains the format string and parameters for
  402. interpolation, log messages in this list are all interpolated.
  403. Unlike 'text', which contains the output from the handler, log
  404. messages in this list are unadorned with levels, timestamps, etc,
  405. making exact comparisons more reliable.
  406. Note that traceback or stack info (from :func:`logging.exception` or
  407. the `exc_info` or `stack_info` arguments to the logging functions) is
  408. not included, as this is added by the formatter in the handler.
  409. .. versionadded:: 3.7
  410. """
  411. return [r.getMessage() for r in self.records]
  412. def clear(self) -> None:
  413. """Reset the list of log records and the captured log text."""
  414. self.handler.clear()
  415. def _force_enable_logging(
  416. self, level: int | str, logger_obj: logging.Logger
  417. ) -> int:
  418. """Enable the desired logging level if the global level was disabled via ``logging.disabled``.
  419. Only enables logging levels greater than or equal to the requested ``level``.
  420. Does nothing if the desired ``level`` wasn't disabled.
  421. :param level:
  422. The logger level caplog should capture.
  423. All logging is enabled if a non-standard logging level string is supplied.
  424. Valid level strings are in :data:`logging._nameToLevel`.
  425. :param logger_obj: The logger object to check.
  426. :return: The original disabled logging level.
  427. """
  428. original_disable_level: int = logger_obj.manager.disable
  429. if isinstance(level, str):
  430. # Try to translate the level string to an int for `logging.disable()`
  431. level = logging.getLevelName(level)
  432. if not isinstance(level, int):
  433. # The level provided was not valid, so just un-disable all logging.
  434. logging.disable(logging.NOTSET)
  435. elif not logger_obj.isEnabledFor(level):
  436. # Each level is `10` away from other levels.
  437. # https://docs.python.org/3/library/logging.html#logging-levels
  438. disable_level = max(level - 10, logging.NOTSET)
  439. logging.disable(disable_level)
  440. return original_disable_level
  441. def set_level(self, level: int | str, logger: str | None = None) -> None:
  442. """Set the threshold level of a logger for the duration of a test.
  443. Logging messages which are less severe than this level will not be captured.
  444. .. versionchanged:: 3.4
  445. The levels of the loggers changed by this function will be
  446. restored to their initial values at the end of the test.
  447. Will enable the requested logging level if it was disabled via :func:`logging.disable`.
  448. :param level: The level.
  449. :param logger: The logger to update. If not given, the root logger.
  450. """
  451. logger_obj = logging.getLogger(logger)
  452. # Save the original log-level to restore it during teardown.
  453. self._initial_logger_levels.setdefault(logger, logger_obj.level)
  454. logger_obj.setLevel(level)
  455. if self._initial_handler_level is None:
  456. self._initial_handler_level = self.handler.level
  457. self.handler.setLevel(level)
  458. initial_disabled_logging_level = self._force_enable_logging(level, logger_obj)
  459. if self._initial_disabled_logging_level is None:
  460. self._initial_disabled_logging_level = initial_disabled_logging_level
  461. @contextmanager
  462. def at_level(self, level: int | str, logger: str | None = None) -> Generator[None]:
  463. """Context manager that sets the level for capturing of logs. After
  464. the end of the 'with' statement the level is restored to its original
  465. value.
  466. Will enable the requested logging level if it was disabled via :func:`logging.disable`.
  467. :param level: The level.
  468. :param logger: The logger to update. If not given, the root logger.
  469. """
  470. logger_obj = logging.getLogger(logger)
  471. orig_level = logger_obj.level
  472. logger_obj.setLevel(level)
  473. handler_orig_level = self.handler.level
  474. self.handler.setLevel(level)
  475. original_disable_level = self._force_enable_logging(level, logger_obj)
  476. try:
  477. yield
  478. finally:
  479. logger_obj.setLevel(orig_level)
  480. self.handler.setLevel(handler_orig_level)
  481. logging.disable(original_disable_level)
  482. @contextmanager
  483. def filtering(self, filter_: logging.Filter) -> Generator[None]:
  484. """Context manager that temporarily adds the given filter to the caplog's
  485. :meth:`handler` for the 'with' statement block, and removes that filter at the
  486. end of the block.
  487. :param filter_: A custom :class:`logging.Filter` object.
  488. .. versionadded:: 7.5
  489. """
  490. self.handler.addFilter(filter_)
  491. try:
  492. yield
  493. finally:
  494. self.handler.removeFilter(filter_)
  495. @fixture
  496. def caplog(request: FixtureRequest) -> Generator[LogCaptureFixture]:
  497. """Access and control log capturing.
  498. Captured logs are available through the following properties/methods::
  499. * caplog.messages -> list of format-interpolated log messages
  500. * caplog.text -> string containing formatted log output
  501. * caplog.records -> list of logging.LogRecord instances
  502. * caplog.record_tuples -> list of (logger_name, level, message) tuples
  503. * caplog.clear() -> clear captured records and formatted log output string
  504. """
  505. result = LogCaptureFixture(request.node, _ispytest=True)
  506. yield result
  507. result._finalize()
  508. def get_log_level_for_setting(config: Config, *setting_names: str) -> int | None:
  509. for setting_name in setting_names:
  510. log_level = config.getoption(setting_name)
  511. if log_level is None:
  512. log_level = config.getini(setting_name)
  513. if log_level:
  514. break
  515. else:
  516. return None
  517. if isinstance(log_level, str):
  518. log_level = log_level.upper()
  519. try:
  520. return int(getattr(logging, log_level, log_level))
  521. except ValueError as e:
  522. # Python logging does not recognise this as a logging level
  523. raise UsageError(
  524. f"'{log_level}' is not recognized as a logging level name for "
  525. f"'{setting_name}'. Please consider passing the "
  526. "logging level num instead."
  527. ) from e
  528. # run after terminalreporter/capturemanager are configured
  529. @hookimpl(trylast=True)
  530. def pytest_configure(config: Config) -> None:
  531. config.pluginmanager.register(LoggingPlugin(config), "logging-plugin")
  532. class LoggingPlugin:
  533. """Attaches to the logging module and captures log messages for each test."""
  534. def __init__(self, config: Config) -> None:
  535. """Create a new plugin to capture log messages.
  536. The formatter can be safely shared across all handlers so
  537. create a single one for the entire test session here.
  538. """
  539. self._config = config
  540. # Report logging.
  541. self.formatter = self._create_formatter(
  542. get_option_ini(config, "log_format"),
  543. get_option_ini(config, "log_date_format"),
  544. get_option_ini(config, "log_auto_indent"),
  545. )
  546. self.log_level = get_log_level_for_setting(config, "log_level")
  547. self.caplog_handler = LogCaptureHandler()
  548. self.caplog_handler.setFormatter(self.formatter)
  549. self.report_handler = LogCaptureHandler()
  550. self.report_handler.setFormatter(self.formatter)
  551. # File logging.
  552. self.log_file_level = get_log_level_for_setting(
  553. config, "log_file_level", "log_level"
  554. )
  555. log_file = get_option_ini(config, "log_file") or os.devnull
  556. if log_file != os.devnull:
  557. directory = os.path.dirname(os.path.abspath(log_file))
  558. if not os.path.isdir(directory):
  559. os.makedirs(directory)
  560. self.log_file_mode = get_option_ini(config, "log_file_mode") or "w"
  561. self.log_file_handler = _FileHandler(
  562. log_file, mode=self.log_file_mode, encoding="UTF-8"
  563. )
  564. log_file_format = get_option_ini(config, "log_file_format", "log_format")
  565. log_file_date_format = get_option_ini(
  566. config, "log_file_date_format", "log_date_format"
  567. )
  568. log_file_formatter = DatetimeFormatter(
  569. log_file_format, datefmt=log_file_date_format
  570. )
  571. self.log_file_handler.setFormatter(log_file_formatter)
  572. # CLI/live logging.
  573. self.log_cli_level = get_log_level_for_setting(
  574. config, "log_cli_level", "log_level"
  575. )
  576. if self._log_cli_enabled():
  577. terminal_reporter = config.pluginmanager.get_plugin("terminalreporter")
  578. # Guaranteed by `_log_cli_enabled()`.
  579. assert terminal_reporter is not None
  580. capture_manager = config.pluginmanager.get_plugin("capturemanager")
  581. # if capturemanager plugin is disabled, live logging still works.
  582. self.log_cli_handler: (
  583. _LiveLoggingStreamHandler | _LiveLoggingNullHandler
  584. ) = _LiveLoggingStreamHandler(terminal_reporter, capture_manager)
  585. else:
  586. self.log_cli_handler = _LiveLoggingNullHandler()
  587. log_cli_formatter = self._create_formatter(
  588. get_option_ini(config, "log_cli_format", "log_format"),
  589. get_option_ini(config, "log_cli_date_format", "log_date_format"),
  590. get_option_ini(config, "log_auto_indent"),
  591. )
  592. self.log_cli_handler.setFormatter(log_cli_formatter)
  593. self._disable_loggers(loggers_to_disable=config.option.logger_disable)
  594. def _disable_loggers(self, loggers_to_disable: list[str]) -> None:
  595. if not loggers_to_disable:
  596. return
  597. for name in loggers_to_disable:
  598. logger = logging.getLogger(name)
  599. logger.disabled = True
  600. def _create_formatter(self, log_format, log_date_format, auto_indent):
  601. # Color option doesn't exist if terminal plugin is disabled.
  602. color = getattr(self._config.option, "color", "no")
  603. if color != "no" and ColoredLevelFormatter.LEVELNAME_FMT_REGEX.search(
  604. log_format
  605. ):
  606. formatter: logging.Formatter = ColoredLevelFormatter(
  607. create_terminal_writer(self._config), log_format, log_date_format
  608. )
  609. else:
  610. formatter = DatetimeFormatter(log_format, log_date_format)
  611. formatter._style = PercentStyleMultiline(
  612. formatter._style._fmt, auto_indent=auto_indent
  613. )
  614. return formatter
  615. def set_log_path(self, fname: str) -> None:
  616. """Set the filename parameter for Logging.FileHandler().
  617. Creates parent directory if it does not exist.
  618. .. warning::
  619. This is an experimental API.
  620. """
  621. fpath = Path(fname)
  622. if not fpath.is_absolute():
  623. fpath = self._config.rootpath / fpath
  624. if not fpath.parent.exists():
  625. fpath.parent.mkdir(exist_ok=True, parents=True)
  626. # https://github.com/python/mypy/issues/11193
  627. stream: io.TextIOWrapper = fpath.open(mode=self.log_file_mode, encoding="UTF-8") # type: ignore[assignment]
  628. old_stream = self.log_file_handler.setStream(stream)
  629. if old_stream:
  630. old_stream.close()
  631. def _log_cli_enabled(self) -> bool:
  632. """Return whether live logging is enabled."""
  633. enabled = self._config.getoption(
  634. "--log-cli-level"
  635. ) is not None or self._config.getini("log_cli")
  636. if not enabled:
  637. return False
  638. terminal_reporter = self._config.pluginmanager.get_plugin("terminalreporter")
  639. if terminal_reporter is None:
  640. # terminal reporter is disabled e.g. by pytest-xdist.
  641. return False
  642. return True
  643. @hookimpl(wrapper=True, tryfirst=True)
  644. def pytest_sessionstart(self) -> Generator[None]:
  645. self.log_cli_handler.set_when("sessionstart")
  646. with catching_logs(self.log_cli_handler, level=self.log_cli_level):
  647. with catching_logs(self.log_file_handler, level=self.log_file_level):
  648. return (yield)
  649. @hookimpl(wrapper=True, tryfirst=True)
  650. def pytest_collection(self) -> Generator[None]:
  651. self.log_cli_handler.set_when("collection")
  652. with catching_logs(self.log_cli_handler, level=self.log_cli_level):
  653. with catching_logs(self.log_file_handler, level=self.log_file_level):
  654. return (yield)
  655. @hookimpl(wrapper=True)
  656. def pytest_runtestloop(self, session: Session) -> Generator[None, object, object]:
  657. if session.config.option.collectonly:
  658. return (yield)
  659. if self._log_cli_enabled() and self._config.get_verbosity() < 1:
  660. # The verbose flag is needed to avoid messy test progress output.
  661. self._config.option.verbose = 1
  662. with catching_logs(self.log_cli_handler, level=self.log_cli_level):
  663. with catching_logs(self.log_file_handler, level=self.log_file_level):
  664. return (yield) # Run all the tests.
  665. @hookimpl
  666. def pytest_runtest_logstart(self) -> None:
  667. self.log_cli_handler.reset()
  668. self.log_cli_handler.set_when("start")
  669. @hookimpl
  670. def pytest_runtest_logreport(self) -> None:
  671. self.log_cli_handler.set_when("logreport")
  672. @contextmanager
  673. def _runtest_for(self, item: nodes.Item, when: str) -> Generator[None]:
  674. """Implement the internals of the pytest_runtest_xxx() hooks."""
  675. with (
  676. catching_logs(
  677. self.caplog_handler,
  678. level=self.log_level,
  679. ) as caplog_handler,
  680. catching_logs(
  681. self.report_handler,
  682. level=self.log_level,
  683. ) as report_handler,
  684. ):
  685. caplog_handler.reset()
  686. report_handler.reset()
  687. item.stash[caplog_records_key][when] = caplog_handler.records
  688. item.stash[caplog_handler_key] = caplog_handler
  689. try:
  690. yield
  691. finally:
  692. log = report_handler.stream.getvalue().strip()
  693. item.add_report_section(when, "log", log)
  694. @hookimpl(wrapper=True)
  695. def pytest_runtest_setup(self, item: nodes.Item) -> Generator[None]:
  696. self.log_cli_handler.set_when("setup")
  697. empty: dict[str, list[logging.LogRecord]] = {}
  698. item.stash[caplog_records_key] = empty
  699. with self._runtest_for(item, "setup"):
  700. yield
  701. @hookimpl(wrapper=True)
  702. def pytest_runtest_call(self, item: nodes.Item) -> Generator[None]:
  703. self.log_cli_handler.set_when("call")
  704. with self._runtest_for(item, "call"):
  705. yield
  706. @hookimpl(wrapper=True)
  707. def pytest_runtest_teardown(self, item: nodes.Item) -> Generator[None]:
  708. self.log_cli_handler.set_when("teardown")
  709. try:
  710. with self._runtest_for(item, "teardown"):
  711. yield
  712. finally:
  713. del item.stash[caplog_records_key]
  714. del item.stash[caplog_handler_key]
  715. @hookimpl
  716. def pytest_runtest_logfinish(self) -> None:
  717. self.log_cli_handler.set_when("finish")
  718. @hookimpl(wrapper=True, tryfirst=True)
  719. def pytest_sessionfinish(self) -> Generator[None]:
  720. self.log_cli_handler.set_when("sessionfinish")
  721. with catching_logs(self.log_cli_handler, level=self.log_cli_level):
  722. with catching_logs(self.log_file_handler, level=self.log_file_level):
  723. return (yield)
  724. @hookimpl
  725. def pytest_unconfigure(self) -> None:
  726. # Close the FileHandler explicitly.
  727. # (logging.shutdown might have lost the weakref?!)
  728. self.log_file_handler.close()
  729. class _FileHandler(logging.FileHandler):
  730. """A logging FileHandler with pytest tweaks."""
  731. def handleError(self, record: logging.LogRecord) -> None:
  732. # Handled by LogCaptureHandler.
  733. pass
  734. class _LiveLoggingStreamHandler(logging_StreamHandler):
  735. """A logging StreamHandler used by the live logging feature: it will
  736. write a newline before the first log message in each test.
  737. During live logging we must also explicitly disable stdout/stderr
  738. capturing otherwise it will get captured and won't appear in the
  739. terminal.
  740. """
  741. # Officially stream needs to be a IO[str], but TerminalReporter
  742. # isn't. So force it.
  743. stream: TerminalReporter = None # type: ignore
  744. def __init__(
  745. self,
  746. terminal_reporter: TerminalReporter,
  747. capture_manager: CaptureManager | None,
  748. ) -> None:
  749. super().__init__(stream=terminal_reporter) # type: ignore[arg-type]
  750. self.capture_manager = capture_manager
  751. self.reset()
  752. self.set_when(None)
  753. self._test_outcome_written = False
  754. def reset(self) -> None:
  755. """Reset the handler; should be called before the start of each test."""
  756. self._first_record_emitted = False
  757. def set_when(self, when: str | None) -> None:
  758. """Prepare for the given test phase (setup/call/teardown)."""
  759. self._when = when
  760. self._section_name_shown = False
  761. if when == "start":
  762. self._test_outcome_written = False
  763. def emit(self, record: logging.LogRecord) -> None:
  764. ctx_manager = (
  765. self.capture_manager.global_and_fixture_disabled()
  766. if self.capture_manager
  767. else nullcontext()
  768. )
  769. with ctx_manager:
  770. if not self._first_record_emitted:
  771. self.stream.write("\n")
  772. self._first_record_emitted = True
  773. elif self._when in ("teardown", "finish"):
  774. if not self._test_outcome_written:
  775. self._test_outcome_written = True
  776. self.stream.write("\n")
  777. if not self._section_name_shown and self._when:
  778. self.stream.section("live log " + self._when, sep="-", bold=True)
  779. self._section_name_shown = True
  780. super().emit(record)
  781. def handleError(self, record: logging.LogRecord) -> None:
  782. # Handled by LogCaptureHandler.
  783. pass
  784. class _LiveLoggingNullHandler(logging.NullHandler):
  785. """A logging handler used when live logging is disabled."""
  786. def reset(self) -> None:
  787. pass
  788. def set_when(self, when: str) -> None:
  789. pass
  790. def handleError(self, record: logging.LogRecord) -> None:
  791. # Handled by LogCaptureHandler.
  792. pass