fix(printer): ignore broken pipe on output streams

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,19 +523,19 @@ 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:
                 # If the last printed message is of 'ephemeral' type, the stop
                 # 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
 

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)