diff --git a/craft_cli/printer.py b/craft_cli/printer.py index 59b8a79b..b0cdfdc8 100644 --- a/craft_cli/printer.py +++ b/craft_cli/printer.py @@ -29,6 +29,7 @@ import time import weakref from collections.abc import Callable +from contextlib import suppress from dataclasses import dataclass, field from datetime import datetime from functools import lru_cache @@ -52,6 +53,12 @@ ANSI_SHOW_CURSOR = "\x1b[?25h" +def _safe_print(*args: Any, **kwargs: Any) -> None: + """Print to a stream, ignoring BrokenPipeError from downstream consumers.""" + with suppress(BrokenPipeError): + print(*args, **kwargs) + + @dataclass class _MessageInfo: """Comprehensive information for a message that may go to screen and log.""" @@ -233,7 +240,7 @@ def __init__(self, log_filepath: pathlib.Path) -> None: if not TESTMODE: self.spinner.start() if _supports_ansi_escape_sequences() and _stream_is_terminal(sys.stderr): - print(ANSI_HIDE_CURSOR, end="", file=sys.stderr, flush=True) + _safe_print(ANSI_HIDE_CURSOR, end="", file=sys.stderr, flush=True) weakref.finalize(self, self.stop) def set_terminal_prefix(self, prefix: str) -> None: @@ -293,11 +300,11 @@ def _write_line_terminal( ): # If the last message's stream is different from this new one, # send a carriage return to the original stream only. - print("\r", flush=True, file=self.prv_msg.stream, end="") + _safe_print("\r", flush=True, file=self.prv_msg.stream, end="") previous_line_end = "" if self.prv_msg and previous_line_end == "\n": previous_line_end = "" - print(flush=True, file=self.prv_msg.stream) + _safe_print(flush=True, file=self.prv_msg.stream) # fill with spaces until the very end, on one hand to clear a possible previous message, # but also to always have the cursor at the very end @@ -320,11 +327,11 @@ def _write_line_terminal( line = _format_term_line( previous_line_end, text, spintext, ephemeral=message.ephemeral ) - print(line, end="", flush=True, file=message.stream) + _safe_print(line, end="", flush=True, file=message.stream) if message.end_line: # finish the just shown line, as we need a clean terminal for some external thing - print(flush=True, file=message.stream) + _safe_print(flush=True, file=message.stream) self.unfinished_stream = None else: self.unfinished_stream = message.stream @@ -340,7 +347,7 @@ def _write_line_captured(self, message: _MessageInfo) -> None: else: text = message.text - print(text, file=message.stream) + _safe_print(text, file=message.stream) def _write_bar_terminal(self, message: _MessageInfo) -> None: """Write a progress bar to the screen.""" @@ -362,7 +369,7 @@ def _write_bar_terminal(self, message: _MessageInfo) -> None: else: # complete the previous line, leaving that message ok maybe_cr = "" - print(flush=True, file=self.prv_msg.stream) + _safe_print(flush=True, file=self.prv_msg.stream) if ( message.bar_progress is None or message.bar_total is None @@ -401,7 +408,7 @@ def _write_bar_terminal(self, message: _MessageInfo) -> None: text = text[: terminal_width - 1] # space for cursor line = f"{maybe_cr}{text}" - print(line, end="", flush=True, file=message.stream) + _safe_print(line, end="", flush=True, file=message.stream) self.unfinished_stream = message.stream def _write_bar_captured(self, message: _MessageInfo) -> None: @@ -516,7 +523,7 @@ def stop(self) -> None: if self.spinner.is_alive(): self.spinner.stop() if _supports_ansi_escape_sequences() and _stream_is_terminal(sys.stderr): - print(ANSI_SHOW_CURSOR, end="", file=sys.stderr, flush=True) + _safe_print(ANSI_SHOW_CURSOR, end="", file=sys.stderr, flush=True) if self.unfinished_stream is not None and not self.unfinished_stream.closed: # With unfinished_stream set, the prv_msg object is valid. if self.prv_msg is not None and self.prv_msg.ephemeral: @@ -524,11 +531,11 @@ def stop(self) -> None: # request must clean and reset the line. cleaner = " " * (_get_terminal_width() - 1) line = "\r" + cleaner + "\r" - print(line, end="", flush=True, file=self.prv_msg.stream) + _safe_print(line, end="", flush=True, file=self.prv_msg.stream) else: # The last printed message is permanent. Leave the cursor on # the next clean line. - print(flush=True, file=self.unfinished_stream) + _safe_print(flush=True, file=self.unfinished_stream) self.log.close() self.stopped = True diff --git a/tests/unit/test_printer.py b/tests/unit/test_printer.py index d32c61aa..b1dcee23 100644 --- a/tests/unit/test_printer.py +++ b/tests/unit/test_printer.py @@ -25,6 +25,7 @@ from datetime import datetime from io import StringIO from pathlib import Path +from typing import cast import pytest from craft_cli import printer as printermod @@ -1208,6 +1209,74 @@ def test_stop_streams_unfinished_err(capsys, log_filepath): assert err == "\n" +def test_show_captured_broken_pipe_does_not_raise(log_filepath, monkeypatch): + """BrokenPipeError from a captured stream should not crash show().""" + monkeypatch.setattr(printermod, "TESTMODE", True) + + class BrokenPipeStream: + closed = False + + def isatty(self): + return False + + def write(self, _text): + raise BrokenPipeError + + def flush(self): + raise BrokenPipeError + + stream = cast("StringIO", BrokenPipeStream()) + printer = Printer(log_filepath) + printer.show(stream, "test text") + + +def test_stop_broken_pipe_does_not_raise(log_filepath, monkeypatch): + """BrokenPipeError from an unfinished stream should not crash stop().""" + monkeypatch.setattr(printermod, "TESTMODE", True) + + class BrokenPipeStream: + closed = False + + def isatty(self): + return False + + def write(self, _text): + raise BrokenPipeError + + def flush(self): + raise BrokenPipeError + + stream = cast("StringIO", BrokenPipeStream()) + printer = Printer(log_filepath) + printer.unfinished_stream = stream + printer.prv_msg = _MessageInfo(stream, "test") + printer.stop() + + +def test_stop_ephemeral_broken_pipe_does_not_raise(log_filepath, monkeypatch): + """BrokenPipeError from an unfinished ephemeral stream should not crash stop().""" + monkeypatch.setattr(printermod, "TESTMODE", True) + monkeypatch.setattr(printermod, "_get_terminal_width", lambda: 10) + + class BrokenPipeStream: + closed = False + + def isatty(self): + return False + + def write(self, _text): + raise BrokenPipeError + + def flush(self): + raise BrokenPipeError + + stream = cast("StringIO", BrokenPipeStream()) + printer = Printer(log_filepath) + printer.unfinished_stream = stream + printer.prv_msg = _MessageInfo(stream, "test", ephemeral=True) + printer.stop() + + def test_stop_spinner_ok(log_filepath): """Stop the spinner.""" printer = Printer(log_filepath)