Quality audit PR 1/6: correctness batch + compatibility guardrails

progressbar/bar.py

@@ -1106,13 +1106,23 @@ def _needs_update(self):
             # limited by the min_poll_interval check above)
             return self.value != self._last_drawn_value
 
-        # Update if value increment is not large enough to
-        # add more bars to progressbar (according to current
-        # terminal width)
-        with contextlib.suppress(Exception):
+        # Update if the value increment is large enough to add more bars
+        # to the progressbar (according to the current terminal width).
+        # While the state is incomplete -- nothing drawn yet, no usable
+        # terminal width, no (nonzero) max value -- there is no width
+        # threshold to compute and no redraw is due; those guards mirror
+        # what a `suppress(Exception)` used to swallow here. Anything else
+        # failing in this math is a real bug and should propagate instead
+        # of silently stopping redraws.
+        if (
+            self.value is not None
+            and self._last_drawn_value is not None
+            and self.term_width
+            and self.max_value
+        ):
             divisor: float = self.max_value / self.term_width  # type: ignore
-            value_divisor = self.value // divisor  # type: ignore
-            pvalue_divisor = self._last_drawn_value // divisor  # type: ignore
+            value_divisor = self.value // divisor
+            pvalue_divisor = self._last_drawn_value // divisor
             if value_divisor != pvalue_divisor:
                 return True
         # No need to redraw yet
@@ -1184,11 +1194,11 @@ def update(
         if self.start_time is None:
             self.start()
 
-        if (
-            value is not None
-            and value is not base.UnknownLength
-            and isinstance(value, (int, float))
-        ):
+        # `isinstance(value, (int, float))` already excludes both `None` and
+        # the `UnknownLength` sentinel (a class, not a numeric instance), so
+        # the earlier explicit `is not None`/`is not UnknownLength` clauses
+        # were redundant.
+        if isinstance(value, (int, float)):
             if self.max_value is base.UnknownLength:
                 # Can't compare against unknown lengths so just update
                 pass

progressbar/env.py

@@ -121,8 +121,11 @@ def is_ansi_terminal(
         # is going to return False if the instance has been overridden and
         # isatty has not been defined we have no way of knowing so we will not
         # use ansi.  ansi terminals will typically define one of the 2
-        # environment variables.
-        with contextlib.suppress(Exception):
+        # environment variables. Only the errors a stream legitimately
+        # produces are treated as "not a terminal": OSError (real I/O),
+        # ValueError (closed/detached file objects) and AttributeError
+        # (objects without isatty). Anything else is a bug and propagates.
+        with contextlib.suppress(OSError, ValueError, AttributeError):
             is_tty: bool = fd.isatty()
             # Try and match any of the huge amount of Linux/Unix ANSI consoles
             if is_tty and ANSI_TERM_RE.match(os.environ.get('TERM', '')):
@@ -155,12 +158,14 @@ def is_terminal(
         # Allow a environment variable override
         is_terminal = env_flag('PROGRESSBAR_IS_TERMINAL', None)
 
-    if is_terminal is None:  # pragma: no cover
-        # Bare except because a lot can go wrong on different systems. If we do
-        # get a TTY we know this is a valid terminal
+    if is_terminal is None:
+        # If we do get a TTY we know this is a valid terminal. Streams can
+        # legitimately fail with OSError (real I/O), ValueError (closed or
+        # detached file objects) or AttributeError (no isatty at all);
+        # anything else is a bug and propagates.
         try:
             is_terminal = fd.isatty()
-        except Exception:
+        except (OSError, ValueError, AttributeError):
             is_terminal = False
 
     return is_terminal

progressbar/fast.py

@@ -12,6 +12,11 @@
 #: present it replaces the pure-Python formatter below. Wired in a later task.
 _format_fast_line: typing.Callable[[FastProgressBar], str] | None = None
 
+#: Spinner frames cycled for unknown-length bars: bar, forward slash, dash,
+#: back slash. A plain (non-raw) literal so the escape is a single ``\`` and
+#: the string is exactly four characters.
+_SPINNER_FRAMES: str = '|/-\\'
+
 
 def _format_seconds(seconds: float) -> str:
     """Render elapsed/ETA seconds as H:MM:SS, matching the Timer widget."""
@@ -50,7 +55,7 @@ def _pure_format_fast_line(bar: FastProgressBar) -> str:
         return f'{prefix}{left}{barstr}{right}{suffix}'
 
     # Unknown length: spinner + count + elapsed (no bar/eta).
-    spinner = r'|/-\\'[int(elapsed * 4) % 4]
+    spinner = _SPINNER_FRAMES[int(elapsed * 4) % len(_SPINNER_FRAMES)]
     item_count = value - min_value + 1
     return (
         f'{prefix}{spinner} {item_count} Elapsed Time: {elapsed_text}{suffix}'

progressbar/multi.py

@@ -53,6 +53,18 @@ class SortKey(str, enum.Enum):
 
 
 class MultiBar(dict[str, bar.ProgressBar]):
+    """Render and manage multiple progressbars from background threads.
+
+    On a clean context-manager exit the multibar waits for its render
+    thread via :meth:`join`. By default (``join_timeout=None``) that wait
+    is unbounded, so a bar that never finishes blocks the program forever.
+    Pass ``join_timeout`` (seconds, or a :class:`datetime.timedelta`) to
+    bound that wait: once it elapses any still-unfinished bars are
+    abandoned and the render thread -- a daemon -- is left running so the
+    program can exit. The default preserves the historical wait-forever
+    behavior.
+    """
+
     fd: typing.TextIO
     _buffer: io.StringIO
 
@@ -73,6 +85,9 @@ class MultiBar(dict[str, bar.ProgressBar]):
     # updates
     update_interval: float
     remove_finished: float | None
+    #: Seconds to wait for the render thread on a clean context-manager
+    # exit before abandoning unfinished bars. `None` waits forever.
+    join_timeout: float | None
 
     #: The kwargs passed to the progressbar constructor
     progressbar_kwargs: dict[str, typing.Any]
@@ -104,6 +119,8 @@ def __init__(
         sort_key: str | SortKey = SortKey.CREATED,
         sort_reverse: bool = True,
         sort_keyfunc: SortKeyFunc | None = None,
+        *,
+        join_timeout: timedelta | float | None = None,
         **progressbar_kwargs: typing.Any,
     ):
         self.fd = fd
@@ -121,6 +138,9 @@ def __init__(
         self.remove_finished = python_utils.delta_to_seconds_or_none(
             remove_finished,
         )
+        self.join_timeout = python_utils.delta_to_seconds_or_none(
+            join_timeout,
+        )
 
         self.progressbar_kwargs = progressbar_kwargs
 
@@ -412,7 +432,14 @@ def __exit__(
         traceback: types.TracebackType | None,
     ) -> bool | None:
         if exc_type is None:
-            self.join()
+            # Bound the wait so a never-finishing bar cannot hang a clean
+            # exit; `join_timeout=None` keeps the historical forever-wait.
+            self.join(timeout=self.join_timeout)
+            if self._thread is not None:
+                # The timeout elapsed with bars unfinished: signal the
+                # render thread to shut down instead of leaving the daemon
+                # looping (and writing) until interpreter exit.
+                self.stop(timeout=self.update_interval)
         else:
             # Don't wait for unfinished progressbars when an exception is
             # propagating; that would block forever

progressbar/terminal/base.py

@@ -280,10 +280,12 @@ def hex(self) -> str:
 
     @property
     def to_ansi_16(self) -> int:
-        # Using int instead of round because it maps slightly better
-        red = int(self.red / 255)
-        green = int(self.green / 255)
-        blue = int(self.blue / 255)
+        # Threshold each channel at half intensity so a mid-range channel
+        # sets its bit. ``int(c / 255)`` was only ever 1 at exactly 255, which
+        # collapsed almost every colour (e.g. maroon 128,0,0) to black.
+        red = int(self.red >= 128)
+        green = int(self.green >= 128)
+        blue = int(self.blue >= 128)
         return (blue << 2) | (green << 1) | red
 
     @property
@@ -405,14 +407,20 @@ def ansi(self) -> types.Optional[str]:
         ):  # pragma: no branch
             return f'2;{self.rgb.red};{self.rgb.green};{self.rgb.blue}'
 
-        if self.xterm:  # pragma: no branch
+        # A true 16-colour terminal must not be handed a 256-colour index,
+        # so translate through to_ansi_16 there. Everywhere else prefer the
+        # registered xterm index (``is not None`` so index 0/Black counts):
+        # rendering an SGR at all means the caller decided colours are
+        # wanted (e.g. forced via ``enable_colors``), even when the global
+        # detection reported no support.
+        if env.COLOR_SUPPORT is env.ColorSupport.XTERM:
+            color = self.rgb.to_ansi_16
+        elif self.xterm is not None:
             color = self.xterm
         elif (
             env.COLOR_SUPPORT is env.ColorSupport.XTERM_256
         ):  # pragma: no branch
             color = self.rgb.to_ansi_256
-        elif env.COLOR_SUPPORT is env.ColorSupport.XTERM:  # pragma: no branch
-            color = self.rgb.to_ansi_16
         else:  # pragma: no branch
             return None
 
@@ -630,6 +638,18 @@ def __init__(self, color: Color, start_code: int, end_code: int) -> None:
         self._color = color
         super().__init__(start_code, end_code)
 
+    def __call__(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self,
+        text: str,
+        *args: typing.Any,
+    ) -> str:
+        if self._color.ansi is None:
+            # No usable color representation for this terminal (e.g. color
+            # support is NONE): leave the text unstyled instead of emitting
+            # a malformed escape code containing the literal string 'None'.
+            return text
+        return super().__call__(text, *args)
+
     @property
     def _start_template(self):
         return CSI.__call__(self, self._start_code, self._color.ansi)

progressbar/utils.py

@@ -30,6 +30,14 @@
 
 StringT = types.TypeVar('StringT', bound=types.StringTypes)
 
+# Precompiled ANSI CSI escape-sequence patterns (str and bytes). Compiled once
+# at import instead of per no_color() call, which runs for every widget on
+# every redraw.
+_ANSI_COLOR_RE: re.Pattern[str] = re.compile('\x1b\\[.*?[@-~]')
+_ANSI_COLOR_RE_BYTES: re.Pattern[bytes] = re.compile(
+    bytes(terminal.ESC, 'ascii') + b'\\[.*?[@-~]',
+)
+
 
 def deltas_to_seconds(
     *deltas: None | datetime.timedelta | float,
@@ -98,12 +106,11 @@ def no_color(value: StringT) -> StringT:
         # per-redraw render cost (len_color is called for every widget).
         if b'\x1b' not in value:
             return value  # type: ignore
-        pattern: bytes = bytes(terminal.ESC, 'ascii') + b'\\[.*?[@-~]'
-        return re.sub(pattern, b'', value)  # type: ignore
+        return _ANSI_COLOR_RE_BYTES.sub(b'', value)  # type: ignore
     elif isinstance(value, str):
         if '\x1b' not in value:
             return value  # type: ignore
-        return re.sub('\x1b\\[.*?[@-~]', '', value)  # type: ignore
+        return _ANSI_COLOR_RE.sub('', value)  # type: ignore
     else:
         raise TypeError(f'`value` must be a string or bytes, got {value!r}')