A logging StreamHandler used by the live logging feature: it will write a newline before the first log message in each test. During live logging we must also explicitly disable stdout/stderr capturing otherwise it will get captured and won't appear in the terminal.
| 903 | |
| 904 | |
| 905 | class _LiveLoggingStreamHandler(logging_StreamHandler): |
| 906 | """A logging StreamHandler used by the live logging feature: it will |
| 907 | write a newline before the first log message in each test. |
| 908 | |
| 909 | During live logging we must also explicitly disable stdout/stderr |
| 910 | capturing otherwise it will get captured and won't appear in the |
| 911 | terminal. |
| 912 | """ |
| 913 | |
| 914 | # Officially stream needs to be a IO[str], but TerminalReporter |
| 915 | # isn't. So force it. |
| 916 | stream: TerminalReporter = None # type: ignore |
| 917 | |
| 918 | def __init__( |
| 919 | self, |
| 920 | terminal_reporter: TerminalReporter, |
| 921 | capture_manager: CaptureManager | None, |
| 922 | ) -> None: |
| 923 | super().__init__(stream=terminal_reporter) # type: ignore[arg-type] |
| 924 | self.capture_manager = capture_manager |
| 925 | self.reset() |
| 926 | self.set_when(None) |
| 927 | self._test_outcome_written = False |
| 928 | |
| 929 | def reset(self) -> None: |
| 930 | """Reset the handler; should be called before the start of each test.""" |
| 931 | self._first_record_emitted = False |
| 932 | |
| 933 | def set_when(self, when: str | None) -> None: |
| 934 | """Prepare for the given test phase (setup/call/teardown).""" |
| 935 | self._when = when |
| 936 | self._section_name_shown = False |
| 937 | if when == "start": |
| 938 | self._test_outcome_written = False |
| 939 | |
| 940 | def emit(self, record: logging.LogRecord) -> None: |
| 941 | ctx_manager = ( |
| 942 | self.capture_manager.global_and_fixture_disabled() |
| 943 | if self.capture_manager |
| 944 | else nullcontext() |
| 945 | ) |
| 946 | with ctx_manager: |
| 947 | if not self._first_record_emitted: |
| 948 | self.stream.write("\n") |
| 949 | self._first_record_emitted = True |
| 950 | elif self._when in ("teardown", "finish"): |
| 951 | if not self._test_outcome_written: |
| 952 | self._test_outcome_written = True |
| 953 | self.stream.write("\n") |
| 954 | if not self._section_name_shown and self._when: |
| 955 | self.stream.section("live log " + self._when, sep="-", bold=True) |
| 956 | self._section_name_shown = True |
| 957 | super().emit(record) |
| 958 | |
| 959 | def handleError(self, record: logging.LogRecord) -> None: |
| 960 | # Handled by LogCaptureHandler. |
| 961 | pass |
| 962 |
no outgoing calls