diff --git a/progressbar/utils.py b/progressbar/utils.py index affb15c..062d2a8 100644 --- a/progressbar/utils.py +++ b/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: diff --git a/tests/test_stream.py b/tests/test_stream.py index 30bdaa9..1fe9758 100644 --- a/tests/test_stream.py +++ b/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 = [] diff --git a/tests/test_terminal.py b/tests/test_terminal.py index 77815a4..5412ec7 100644 --- a/tests/test_terminal.py +++ b/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) diff --git a/tests/test_utils.py b/tests/test_utils.py index 375d907..55ec8e4 100644 --- a/tests/test_utils.py +++ b/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. diff --git a/tests/test_with.py b/tests/test_with.py index 79307c9..97c24df 100644 --- a/tests/test_with.py +++ b/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())