Harden #301/#212/#295 regression tests + make AttributeDict generic

progressbar/utils.py

@@ -517,7 +517,7 @@ def excepthook(
         self.flush()
 
 
-class AttributeDict(dict):
+class AttributeDict(dict[str, T], typing.Generic[T]):
     """
     A dict that can be accessed with .attribute.
 
@@ -563,13 +563,13 @@ class AttributeDict(dict):
     AttributeError: No such attribute: spam
     """
 
-    def __getattr__(self, name: str) -> typing.Any:
+    def __getattr__(self, name: str) -> T:
         if name in self:
             return self[name]
         else:
             raise AttributeError(f'No such attribute: {name}')
 
-    def __setattr__(self, name: str, value: typing.Any) -> None:
+    def __setattr__(self, name: str, value: T) -> None:
         self[name] = value
 
     def __delattr__(self, name: str) -> None:

tests/test_stream.py

@@ -1,7 +1,9 @@
 import io
 import logging
 import os
+import subprocess
 import sys
+import textwrap
 
 import pytest
 
@@ -485,3 +487,69 @@ def flush(self) -> None:
     written = wrapper.write('hello\n')
     assert written == 6
     assert target.flushes >= 1
+
+
+def test_redirect_stdout_unwrapped_after_keyboard_interrupt() -> None:
+    # #212: when redirect_stdout wraps sys.stdout and iteration is abandoned
+    # by a KeyboardInterrupt, the bar must unwrap stdout again. Runs in a
+    # subprocess because stream wrapping mutates process-global sys.stdout.
+    child = textwrap.dedent(
+        """
+        import sys
+        import progressbar
+        from progressbar.utils import WrappingIO
+
+        try:
+            for i in progressbar.progressbar(range(100), redirect_stdout=True):
+                print('text', i)
+                if i == 50:
+                    raise KeyboardInterrupt
+        except KeyboardInterrupt:
+            pass
+
+        sys.exit(2 if isinstance(sys.stdout, WrappingIO) else 0)
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, '-c', child], capture_output=True, text=True
+    )
+    assert result.returncode == 0, (
+        f'stdout left wrapped after interrupt; rc={result.returncode}\n'
+        f'stderr={result.stderr[-500:]}'
+    )
+
+
+def test_wrap_logging_deduplicates_shared_handler(monkeypatch) -> None:
+    # A handler attached to more than one logger must be wrapped only once.
+    # _iter_loggers yields the root logger then named loggers, so a handler
+    # shared by both is seen twice and the second visit is skipped.
+    for _ in range(5):
+        progressbar.streams.unwrap(stderr=True, stdout=True)
+    progressbar.streams.unwrap_logging()
+
+    stream = io.StringIO()
+    monkeypatch.setattr(sys, 'stderr', stream)
+    monkeypatch.setattr(progressbar.streams, 'original_stderr', stream)
+    monkeypatch.setattr(progressbar.streams, 'stderr', stream)
+
+    root = logging.getLogger()
+    named = logging.getLogger('progressbar-test-shared-handler')
+    named.handlers = []
+    named.propagate = False
+    handler = logging.StreamHandler(sys.stderr)
+    named.addHandler(handler)
+    root.addHandler(handler)  # same handler object on two loggers
+
+    progressbar.streams.wrap_stderr()
+    try:
+        progressbar.streams.wrap_logging()
+        # Recorded exactly once despite being reachable via two loggers.
+        shared = [
+            h for h, _ in progressbar.streams.logging_handlers if h is handler
+        ]
+        assert len(shared) == 1
+    finally:
+        progressbar.streams.unwrap_logging()
+        progressbar.streams.unwrap(stderr=True)
+        root.removeHandler(handler)
+        named.handlers = []

tests/test_terminal.py

@@ -222,3 +222,18 @@ def test_redirect_blank_line_off_by_default(monkeypatch) -> None:
 
     monkeypatch.setattr(utils.streams, 'needs_clear', lambda: True)
     assert '\n' not in _redirect_update_output(redirect_blank_line=False)
+
+
+def test_redirect_blank_line_position(monkeypatch) -> None:
+    # #295: the blank line must sit immediately AFTER the clear sequence and
+    # BEFORE the bar redraw, not merely 'somewhere in the output'.
+    from progressbar import utils
+
+    monkeypatch.setattr(utils.streams, 'needs_clear', lambda: True)
+    output = _redirect_update_output(redirect_blank_line=True)
+    clear = '\r' + ' ' * 40 + '\r'
+    assert (clear + '\n') in output, repr(output)
+    assert output.index(clear + '\n') < output.index('50%'), repr(output)
+
+    output_off = _redirect_update_output(redirect_blank_line=False)
+    assert (clear + '\n') not in output_off, repr(output_off)

tests/test_utils.py

@@ -175,6 +175,20 @@ def test_attribute_dict_set_get_del() -> None:
         del attrs.spam
 
 
+def test_attribute_dict_generic_value_type() -> None:
+    # The generic upgrade lets homogeneous, explicitly-typed instances flow the
+    # real value type. Runtime contract asserted here; the static benefit
+    # (reveal_type -> int, mis-typed assignment flagged) is checked by pyright.
+    attrs: utils.AttributeDict[int] = utils.AttributeDict()
+    attrs.count = 5
+    assert attrs.count == 5
+    assert attrs['count'] == 5
+
+    mixed = utils.AttributeDict(a=1, b='x')
+    assert mixed.a == 1
+    assert mixed.b == 'x'
+
+
 def test_stream_wrapper_unwrap_restores_excepthook() -> None:
     # Regression: C7 - unwrap_stdout/unwrap_stderr left the custom
     # excepthook installed forever.

tests/test_with.py

@@ -33,3 +33,33 @@ def test_context_manager_and_iterable_no_duplicate() -> None:
     # The completed bar must be rendered exactly once; the bug finished the
     # bar twice (StopIteration and then __exit__), drawing it a second time.
     assert fd.getvalue().count('100% (10 of 10)') == 1, repr(fd.getvalue())
+
+
+def test_context_manager_and_iterable_reporter_widgets_no_duplicate() -> None:
+    # Regression #301 with the reporter's exact widget set and a generator:
+    # using the bar as BOTH context manager and iterable must render the
+    # completed bar exactly once.
+    from progressbar.widgets import (
+        AnimatedMarker,
+        GranularBar,
+        SimpleProgress,
+        Timer,
+    )
+
+    fd = io.StringIO()
+    widgets = [
+        AnimatedMarker(),
+        ' ',
+        SimpleProgress(),
+        ' ',
+        GranularBar(),
+        ' ',
+        Timer(),
+    ]
+    with progressbar.ProgressBar(
+        max_value=10, fd=fd, is_terminal=True, term_width=60, widgets=widgets
+    ) as bar:
+        for _ in bar(i for i in range(10)):
+            pass
+
+    assert fd.getvalue().count('10 of 10') == 1, repr(fd.getvalue())