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.
| 759 | |
| 760 | |
| 761 | class _LiveLoggingStreamHandler(logging.StreamHandler): |
| 762 | """A logging StreamHandler used by the live logging feature: it will |
| 763 | write a newline before the first log message in each test. |
| 764 | |
| 765 | During live logging we must also explicitly disable stdout/stderr |
| 766 | capturing otherwise it will get captured and won't appear in the |
| 767 | terminal. |
| 768 | """ |
| 769 | |
| 770 | # Officially stream needs to be a IO[str], but TerminalReporter |
| 771 | # isn't. So force it. |
| 772 | stream: TerminalReporter = None # type: ignore |
| 773 | |
| 774 | def __init__( |
| 775 | self, |
| 776 | terminal_reporter: TerminalReporter, |
| 777 | capture_manager: Optional[CaptureManager], |
| 778 | ) -> None: |
| 779 | super().__init__(stream=terminal_reporter) # type: ignore[arg-type] |
| 780 | self.capture_manager = capture_manager |
| 781 | self.reset() |
| 782 | self.set_when(None) |
| 783 | self._test_outcome_written = False |
| 784 | |
| 785 | def reset(self) -> None: |
| 786 | """Reset the handler; should be called before the start of each test.""" |
| 787 | self._first_record_emitted = False |
| 788 | |
| 789 | def set_when(self, when: Optional[str]) -> None: |
| 790 | """Prepare for the given test phase (setup/call/teardown).""" |
| 791 | self._when = when |
| 792 | self._section_name_shown = False |
| 793 | if when == "start": |
| 794 | self._test_outcome_written = False |
| 795 | |
| 796 | def emit(self, record: logging.LogRecord) -> None: |
| 797 | ctx_manager = ( |
| 798 | self.capture_manager.global_and_fixture_disabled() |
| 799 | if self.capture_manager |
| 800 | else nullcontext() |
| 801 | ) |
| 802 | with ctx_manager: |
| 803 | if not self._first_record_emitted: |
| 804 | self.stream.write("\n") |
| 805 | self._first_record_emitted = True |
| 806 | elif self._when in ("teardown", "finish"): |
| 807 | if not self._test_outcome_written: |
| 808 | self._test_outcome_written = True |
| 809 | self.stream.write("\n") |
| 810 | if not self._section_name_shown and self._when: |
| 811 | self.stream.section("live log " + self._when, sep="-", bold=True) |
| 812 | self._section_name_shown = True |
| 813 | super().emit(record) |
| 814 | |
| 815 | def handleError(self, record: logging.LogRecord) -> None: |
| 816 | # Handled by LogCaptureHandler. |
| 817 | pass |
| 818 |
no outgoing calls