diff --git a/news/14238.bugfix.rst b/news/14238.bugfix.rst new file mode 100644 index 0000000000..a2f74c9aec --- /dev/null +++ b/news/14238.bugfix.rst @@ -0,0 +1,2 @@ +``inprocess-build-deps`` reports started/finished status when printing to a +non-interactive terminal properly. diff --git a/src/pip/_internal/build_env/installer.py b/src/pip/_internal/build_env/installer.py index 790ad5f3b7..899dbeafc8 100644 --- a/src/pip/_internal/build_env/installer.py +++ b/src/pip/_internal/build_env/installer.py @@ -10,7 +10,7 @@ from typing import TYPE_CHECKING from pip._internal.build_env.base import Prefix -from pip._internal.cli.spinners import open_rich_spinner, open_spinner +from pip._internal.cli.spinners import open_spinner from pip._internal.exceptions import ( BuildDependencyInstallError, DiagnosticPipError, @@ -143,7 +143,7 @@ def install( identify_requirement = ( f" for {for_req.name}" if for_req and for_req.name else "" ) - with open_spinner(f"Installing {kind}") as spinner: + with open_spinner(f"Installing {kind}", autostart=False) as spinner: call_subprocess( args, command_desc=f"installing {kind}{identify_requirement}", @@ -225,7 +225,7 @@ def install( # Hide the logs from the installation of build dependencies. # They will be shown only if an error occurs. capture_ctx: ContextManager[StringIO] = capture_logging() - spinner: ContextManager[None] = open_rich_spinner(f"Installing {kind}") + spinner: ContextManager[object] = open_spinner(f"Installing {kind}") else: # Otherwise, pass-through all logs (with a header). capture_ctx, spinner = nullcontext(StringIO()), nullcontext() diff --git a/src/pip/_internal/cli/progress_bars.py b/src/pip/_internal/cli/progress_bars.py index f276c7e2c4..35e8d0da9c 100644 --- a/src/pip/_internal/cli/progress_bars.py +++ b/src/pip/_internal/cli/progress_bars.py @@ -2,6 +2,7 @@ import functools import sys +import time from collections.abc import Callable, Generator, Iterable, Iterator from typing import TYPE_CHECKING, Literal, TypeVar @@ -19,7 +20,6 @@ TransferSpeedColumn, ) -from pip._internal.cli.spinners import RateLimiter from pip._internal.utils.logging import get_console, get_indentation if TYPE_CHECKING: @@ -30,6 +30,20 @@ BarType = Literal["on", "off", "raw"] +class RateLimiter: + def __init__(self, min_update_interval_seconds: float) -> None: + self._min_update_interval_seconds = min_update_interval_seconds + self._last_update: float = 0 + + def ready(self) -> bool: + now = time.time() + delta = now - self._last_update + return delta >= self._min_update_interval_seconds + + def reset(self) -> None: + self._last_update = time.time() + + def _rich_download_progress_bar( iterable: Iterable[bytes], *, diff --git a/src/pip/_internal/cli/spinners.py b/src/pip/_internal/cli/spinners.py index 58aad2853d..2aaca9c54b 100644 --- a/src/pip/_internal/cli/spinners.py +++ b/src/pip/_internal/cli/spinners.py @@ -3,233 +3,164 @@ import contextlib import itertools import logging -import sys -import time from collections.abc import Generator -from typing import IO, Final - -from pip._vendor.rich.console import ( - Console, - ConsoleOptions, - RenderableType, - RenderResult, -) +from threading import Event, Thread +from typing import Final, Protocol + +from pip._vendor.rich.console import Console from pip._vendor.rich.live import Live -from pip._vendor.rich.measure import Measurement from pip._vendor.rich.text import Text -from pip._internal.utils.compat import WINDOWS from pip._internal.utils.logging import get_console, get_indentation logger = logging.getLogger(__name__) +_active_spinner: SpinnerInterface | None = None SPINNER_CHARS: Final = r"-\|/" SPINS_PER_SECOND: Final = 8 +NONINTERACTIVE_SPINNER_INTERVAL: Final = 60 -class SpinnerInterface: - def spin(self) -> None: - raise NotImplementedError() - - def finish(self, final_status: str) -> None: - raise NotImplementedError() - - -class InteractiveSpinner(SpinnerInterface): - def __init__( - self, - message: str, - file: IO[str] | None = None, - spin_chars: str = SPINNER_CHARS, - # Empirically, 8 updates/second looks nice - min_update_interval_seconds: float = 1 / SPINS_PER_SECOND, - ): - self._message = message - if file is None: - file = sys.stdout - self._file = file - self._rate_limiter = RateLimiter(min_update_interval_seconds) - self._finished = False - - self._spin_cycle = itertools.cycle(spin_chars) - - self._file.write(" " * get_indentation() + self._message + " ... ") - self._width = 0 - - def _write(self, status: str) -> None: - assert not self._finished - # Erase what we wrote before by backspacing to the beginning, writing - # spaces to overwrite the old text, and then backspacing again - backup = "\b" * self._width - self._file.write(backup + " " * self._width + backup) - # Now we have a blank slate to add our status - self._file.write(status) - self._width = len(status) - self._file.flush() - self._rate_limiter.reset() - - def spin(self) -> None: - if self._finished: - return - if not self._rate_limiter.ready(): - return - self._write(next(self._spin_cycle)) - - def finish(self, final_status: str) -> None: - if self._finished: - return - self._write(final_status) - self._file.write("\n") - self._file.flush() - self._finished = True - - -# Used for dumb terminals, non-interactive installs (no tty), etc. -# We still print updates occasionally (once every 60 seconds by default) to -# act as a keep-alive for systems like Travis-CI that take lack-of-output as -# an indication that a task has frozen. -class NonInteractiveSpinner(SpinnerInterface): - def __init__(self, message: str, min_update_interval_seconds: float = 60.0) -> None: - self._message = message - self._finished = False - self._rate_limiter = RateLimiter(min_update_interval_seconds) - self._update("started") - - def _update(self, status: str) -> None: - assert not self._finished - self._rate_limiter.reset() - logger.info("%s: %s", self._message, status) - - def spin(self) -> None: - if self._finished: - return - if not self._rate_limiter.ready(): - return - self._update("still running...") +class SpinnerInterface(Protocol): + """Common interface for status spinners. - def finish(self, final_status: str) -> None: - if self._finished: - return - self._update(f"finished with status '{final_status}'") - self._finished = True + If finish() is called when the spinner is already done, it will do + nothing, allowing for more robust error handling. + Please note that on (first) finish, a final status message will be + shown even if the spinner was never started. + """ -class RateLimiter: - def __init__(self, min_update_interval_seconds: float) -> None: - self._min_update_interval_seconds = min_update_interval_seconds - self._last_update: float = 0 - - def ready(self) -> bool: - now = time.time() - delta = now - self._last_update - return delta >= self._min_update_interval_seconds + def start(self) -> None: ... + def finish(self, label: str) -> None: ... - def reset(self) -> None: - self._last_update = time.time() +class NoopSpinner(SpinnerInterface): + """No-op spinner for when absolutely zero output is desired.""" -@contextlib.contextmanager -def open_spinner(message: str) -> Generator[SpinnerInterface, None, None]: - # Interactive spinner goes directly to sys.stdout rather than being routed - # through the logging system, but it acts like it has level INFO, - # i.e. it's only displayed if we're at level INFO or better. - # Non-interactive spinner goes through the logging system, so it is always - # in sync with logging configuration. - if sys.stdout.isatty() and logger.getEffectiveLevel() <= logging.INFO: - spinner: SpinnerInterface = InteractiveSpinner(message) - else: - spinner = NonInteractiveSpinner(message) - try: - with hidden_cursor(sys.stdout): - yield spinner - except KeyboardInterrupt: - spinner.finish("canceled") - raise - except Exception: - spinner.finish("error") - raise - else: - spinner.finish("done") + def start(self) -> None: + pass + def finish(self, label: str) -> None: + pass -class _PipRichSpinner: - """ - Custom rich spinner that matches the style of the legacy spinners. - (*) Updates will be handled in a background thread by a rich live panel - which will call render() automatically at the appropriate time. - """ +class RichSpinner(SpinnerInterface): + """Status spinner for interactive terminals.""" - def __init__(self, label: str) -> None: + def __init__(self, label: str, console: Console) -> None: self.label = label + self._console = console self._spin_cycle = itertools.cycle(SPINNER_CHARS) self._spinner_text = "" self._finished = False self._indent = get_indentation() * " " + self._live: Live | None = None - def __rich_console__( - self, console: Console, options: ConsoleOptions - ) -> RenderResult: - yield self.render() - - def __rich_measure__( - self, console: Console, options: ConsoleOptions - ) -> Measurement: - text = self.render() - return Measurement.get(console, options, text) - - def render(self) -> RenderableType: + def __rich__(self) -> Text: + # This is called as needed at the right pace by the rich live instance. if not self._finished: self._spinner_text = next(self._spin_cycle) - return Text.assemble(self._indent, self.label, " ... ", self._spinner_text) + line = f"{self._indent}{self.label} ... {self._spinner_text}" + return Text(line, overflow="fold", no_wrap=False) + + def start(self) -> None: + self._live = Live( + self, + refresh_per_second=SPINS_PER_SECOND, + console=self._console, + transient=True, + ) + self._live.start(refresh=True) def finish(self, status: str) -> None: """Stop spinning and set a final status message.""" - self._spinner_text = status - self._finished = True + if not self._finished: + self._finished = True + if self._live is not None: + self._live.stop() + logger.info("%s ... %s", self.label, status) -@contextlib.contextmanager -def open_rich_spinner(label: str, console: Console | None = None) -> Generator[None]: - if not logger.isEnabledFor(logging.INFO): - # Don't show spinner if --quiet is given. - yield - return +class NonInteractiveSpinner(SpinnerInterface): + """ + Used for dumb terminals, non-interactive installs (no tty), etc. + We still print updates occasionally (once every 60 seconds by default) to + act as a keep-alive for systems like Travis-CI that take lack-of-output as + an indication that a task has frozen. + """ - console = console or get_console() - spinner = _PipRichSpinner(label) - with Live(spinner, refresh_per_second=SPINS_PER_SECOND, console=console): - try: - yield - except KeyboardInterrupt: - spinner.finish("canceled") - raise - except Exception: - spinner.finish("error") - raise + def __init__(self, label: str, console: Console) -> None: + self._label = label + self._console = console + self._indent = get_indentation() * " " + self._thread: Thread | None = None + self._finish_event = Event() + self._print_line("started") + + def _print_line(self, message: str, use_logger: bool = True) -> None: + if use_logger: + logger.info("%s: %s", self._label, message) else: - spinner.finish("done") + self._console.print(Text(f"{self._indent}{self._label}: {message}")) + + def _report_progress(self) -> None: + while not self._finish_event.wait(NONINTERACTIVE_SPINNER_INTERVAL): + # NOTE: the logger can't be used here since logging may be captured + # while this spinner is active (e.g., when installing build dependencies). + self._print_line("still running...", use_logger=False) + def start(self) -> None: + self._thread = Thread(target=self._report_progress) + self._thread.start() -HIDE_CURSOR = "\x1b[?25l" -SHOW_CURSOR = "\x1b[?25h" + def finish(self, status: str) -> None: + if not self._finish_event.is_set(): + self._finish_event.set() + if self._thread is not None: + self._thread.join() + self._print_line(f"finished with status '{status}'") @contextlib.contextmanager -def hidden_cursor(file: IO[str]) -> Generator[None, None, None]: - # The Windows terminal does not support the hide/show cursor ANSI codes, - # even via colorama. So don't even try. - if WINDOWS: - yield - # We don't want to clutter the output with control characters if we're - # writing to a file, or if the user is running with --quiet. - # See https://github.com/pypa/pip/issues/3418 - elif not file.isatty() or logger.getEffectiveLevel() > logging.INFO: - yield +def open_spinner( + label: str, console: Console | None = None, *, autostart: bool = True +) -> Generator[SpinnerInterface]: + """Helper for opening a status spinner. + + It will select the right spinner type for the current environment and + automatically handle starting and finishing the spinner as needed. + """ + global _active_spinner + if _active_spinner is not None: + yield NoopSpinner() + return + if logger.getEffectiveLevel() > logging.INFO: + # Don't write *anything* if --quiet is given. + yield NoopSpinner() + return + + console = console or get_console() + if console.is_interactive: + spinner: SpinnerInterface = RichSpinner(label, console) else: - file.write(HIDE_CURSOR) - try: - yield - finally: - file.write(SHOW_CURSOR) + spinner = NonInteractiveSpinner(label, console) + if autostart: + spinner.start() + try: + _active_spinner = spinner + yield spinner + except KeyboardInterrupt: + spinner.finish("canceled") + raise + except Exception: + spinner.finish("error") + raise + except BaseException: + spinner.finish("unknown") + raise + else: + spinner.finish("done") + finally: + _active_spinner = None diff --git a/src/pip/_internal/utils/subprocess.py b/src/pip/_internal/utils/subprocess.py index 06f45645d0..0bce0a0b57 100644 --- a/src/pip/_internal/utils/subprocess.py +++ b/src/pip/_internal/utils/subprocess.py @@ -118,6 +118,9 @@ def call_subprocess( # Only use the spinner if we're not showing the subprocess output # and we have a spinner. use_spinner = not showing_subprocess and spinner is not None + if use_spinner: + assert spinner is not None + spinner.start() log_subprocess("Running command %s", command_desc) env = os.environ.copy() @@ -160,10 +163,6 @@ def call_subprocess( # Show the line immediately. log_subprocess(line) - # Update the spinner. - if use_spinner: - assert spinner - spinner.spin() try: proc.wait() finally: @@ -237,7 +236,7 @@ def runner( cwd: str | None = None, extra_environ: Mapping[str, Any] | None = None, ) -> None: - with open_spinner(message) as spinner: + with open_spinner(message, autostart=False) as spinner: call_subprocess( cmd, command_desc=message, diff --git a/tests/conftest.py b/tests/conftest.py index c1a3d60ce3..6c5a5cda8c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -37,6 +37,8 @@ from installer.sources import WheelFile from pip._internal.locations import _USE_SYSCONFIG +from pip._internal.utils import logging as pip_logging_module +from pip._internal.utils.logging import PipConsole from pip._internal.utils.temp_dir import global_tempdir_manager from tests.lib import ( @@ -394,6 +396,20 @@ def scoped_global_tempdir_manager(request: pytest.FixtureRequest) -> Iterator[No yield +@pytest.fixture(autouse=True) +def setup_console() -> Iterator[None]: + """Set up rich consoles so tests that produce rich output as a side-effect + don't crash when trying to print. + """ + stdout_console = PipConsole(file=sys.stdout, no_color=True, soft_wrap=True) + stderr_console = PipConsole(file=sys.stderr, no_color=True, soft_wrap=True) + with ( + patch.object(pip_logging_module, "_stdout_console", stdout_console), + patch.object(pip_logging_module, "_stderr_console", stderr_console), + ): + yield + + @pytest.fixture(scope="session") def pip_src(tmpdir_factory: pytest.TempPathFactory) -> Path: def not_code_files_and_folders(path: str, names: list[str]) -> Iterable[str]: diff --git a/tests/unit/test_cli_spinners.py b/tests/unit/test_cli_spinners.py index 43736dfe8d..1f322f83a4 100644 --- a/tests/unit/test_cli_spinners.py +++ b/tests/unit/test_cli_spinners.py @@ -4,14 +4,16 @@ from collections.abc import Callable, Generator from contextlib import contextmanager from io import StringIO -from unittest.mock import Mock +from threading import Event +from typing import Any +from unittest.mock import Mock, patch import pytest from pip._vendor.rich.console import Console from pip._internal.cli import spinners -from pip._internal.cli.spinners import open_rich_spinner +from pip._internal.cli.spinners import open_spinner @contextmanager @@ -25,40 +27,132 @@ def patch_logger_level(level: int) -> Generator[None]: spinners.logger.setLevel(original_level) -class TestRichSpinner: - @pytest.mark.parametrize( - "status, func", - [ - ("done", lambda: None), - ("error", lambda: 1 / 0), - ("canceled", Mock(side_effect=KeyboardInterrupt)), - ], - ) - def test_finish(self, status: str, func: Callable[[], None]) -> None: - """ - Check that the spinner finish message is set correctly depending - on how the spinner came to a stop. - """ - stream = StringIO() - try: - with patch_logger_level(logging.INFO): - with open_rich_spinner("working", Console(file=stream)): - func() - except BaseException: +@pytest.mark.parametrize( + "status, func", + [ + ("done", lambda: None), + ("error", Mock(side_effect=ValueError)), + ("canceled", Mock(side_effect=KeyboardInterrupt)), + ], +) +@pytest.mark.parametrize( + "isatty, expected_output", + [ + (True, "working ... {status}"), + (False, "working: started\nworking: finished with status '{status}'"), + ], +) +def test_finish( + status: str, + func: Callable[[], None], + isatty: bool, + expected_output: str, + caplog: pytest.LogCaptureFixture, +) -> None: + """Check that the helper reports final statuses in each stdout mode.""" + console = Console(force_interactive=isatty) + try: + with patch_logger_level(logging.INFO): + with open_spinner("working", console, autostart=False): + func() + except BaseException: + pass + + assert caplog.messages == expected_output.format(status=status).splitlines() + + +@pytest.mark.parametrize( + "level, isatty, expected_type", + [ + (logging.INFO, True, spinners.RichSpinner), + (logging.WARNING, True, spinners.NoopSpinner), + (logging.INFO, False, spinners.NonInteractiveSpinner), + (logging.ERROR, False, spinners.NoopSpinner), + ], +) +def test_selects_spinner_for_environment( + level: int, isatty: bool, expected_type: type[object] +) -> None: + """Check that spinner selection follows verbosity and stdout mode.""" + console = Console(force_interactive=isatty) + with patch_logger_level(level): + with open_spinner("working", console, autostart=False) as spinner: + assert isinstance(spinner, expected_type) + + +@pytest.mark.parametrize( + "isatty, spinner_type", + [(True, spinners.RichSpinner), (False, spinners.NonInteractiveSpinner)], +) +@pytest.mark.parametrize("autostart", [False, True]) +def test_starts_spinner_when_requested( + isatty: bool, spinner_type: type[object], autostart: bool +) -> None: + """Check that autostart controls whether the selected spinner starts.""" + console = Console(force_interactive=isatty) + with patch_logger_level(logging.INFO), patch.object(spinner_type, "start") as start: + with open_spinner("working", console, autostart=autostart): pass - output = stream.getvalue() - assert output == f"working ... {status}" - - @pytest.mark.parametrize( - "level, visible", - [(logging.ERROR, False), (logging.INFO, True), (logging.DEBUG, True)], - ) - def test_verbosity(self, level: int, visible: bool) -> None: - """Is the spinner hidden at the appropriate verbosity?""" - stream = StringIO() - with patch_logger_level(level): - with open_rich_spinner("working", Console(file=stream)): - pass - - assert bool(stream.getvalue()) == visible + assert start.called is autostart + + +@patch_logger_level(logging.INFO) +def test_noninteractive_spinner_lifecycle( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Ensure the noninteractive spinner starts, spins, and finishes correctly.""" + stream = StringIO() + spinner = spinners.NonInteractiveSpinner("step", Console(file=stream)) + heartbeat_seen = Event() + print_line = spinner._print_line + + def record_printed_line(message: str, *args: Any, **kwargs: Any) -> None: + print_line(message, *args, **kwargs) + if message == "still running...": + heartbeat_seen.set() + + monkeypatch.setattr(spinners, "NONINTERACTIVE_SPINNER_INTERVAL", 0.01) + monkeypatch.setattr(spinner, "_print_line", record_printed_line) + + spinner.start() + try: + assert heartbeat_seen.wait(timeout=1) + finally: + spinner.finish("done") + + assert spinner._thread is not None + assert not spinner._thread.is_alive() + assert "step: started" == caplog.messages[0] + assert "step: still running...\n" in stream.getvalue() + assert "step: finished with status 'done'" == caplog.messages[1] + + +def test_no_op_spinner_does_not_write_output(caplog: pytest.LogCaptureFixture) -> None: + """Check that quiet mode suppresses all spinner output.""" + stream = StringIO() + with patch_logger_level(logging.ERROR): + with open_spinner("working", Console(file=stream)) as spinner: + spinner.start() + spinner.finish("done") + + assert stream.getvalue() == "" + assert not caplog.messages + + +def test_repeated_spinner_is_blocked(caplog: pytest.LogCaptureFixture) -> None: + stream = StringIO() + with patch_logger_level(logging.INFO): + console = Console(file=stream) + with open_spinner("working", console) as spinner: + spinner.start() + with open_spinner("working2", console) as spinner2: + spinner2.start() + spinner2.finish("error") + spinner.finish("done") + + assert stream.getvalue() == "" + assert caplog.messages == [ + "working: started", + "working: finished with status 'done'", + ] diff --git a/tests/unit/test_utils_subprocess.py b/tests/unit/test_utils_subprocess.py index acd793df25..4222d9e800 100644 --- a/tests/unit/test_utils_subprocess.py +++ b/tests/unit/test_utils_subprocess.py @@ -79,11 +79,11 @@ def test_call_subprocess_stdout_only( class FakeSpinner(SpinnerInterface): def __init__(self) -> None: - self.spin_count = 0 + self.did_start = False self.final_status: str | None = None - def spin(self) -> None: - self.spin_count += 1 + def start(self) -> None: + self.did_start = True def finish(self, final_status: str) -> None: self.final_status = final_status @@ -102,7 +102,7 @@ def check_result( spinner: FakeSpinner, result: str | None, expected: tuple[list[str] | None, list[tuple[str, int, str]]], - expected_spinner: tuple[int, str | None], + expected_spinner: tuple[bool, str | None], ) -> None: """ Check the result of calling call_subprocess(). @@ -118,7 +118,7 @@ def check_result( 2) `expected_records` is the expected value of caplog.record_tuples. :param expected_spinner: a 2-tuple of the spinner's expected - (spin_count, final_status). + (did_start, final_status). """ expected_proc, expected_records = expected @@ -147,7 +147,7 @@ def check_result( # chronologically. assert expected_record[2] in record[2] - assert (spinner.spin_count, spinner.final_status) == expected_spinner + assert (spinner.did_start, spinner.final_status) == expected_spinner def prepare_call( self, @@ -195,7 +195,7 @@ def test_debug_logging( spinner, result, expected, - expected_spinner=(0, None), + expected_spinner=(False, None), ) def test_info_logging( @@ -216,8 +216,8 @@ def test_info_logging( ["Hello", "world"], [], ) - # The spinner should spin twice in this case since the subprocess - # output isn't being written to the console. + # The spinner should spin since the subprocess output isn't being + # written to the console. self.check_result( capfd, caplog, @@ -225,7 +225,7 @@ def test_info_logging( spinner, result, expected, - expected_spinner=(2, "done"), + expected_spinner=(True, "done"), ) def test_info_logging__subprocess_error( @@ -265,8 +265,8 @@ def test_info_logging__subprocess_error( ("pip.subprocessor", ERROR, "subprocess error exited with 1"), ], ) - # The spinner should spin three times in this case since the - # subprocess output isn't being written to the console. + # The spinner should spin since the subprocess output isn't + # being written to the console. self.check_result( capfd, caplog, @@ -274,7 +274,7 @@ def test_info_logging__subprocess_error( spinner, result, expected, - expected_spinner=(3, "error"), + expected_spinner=(True, "error"), ) def test_info_logging_with_show_stdout_true( @@ -309,7 +309,7 @@ def test_info_logging_with_show_stdout_true( spinner, result, expected, - expected_spinner=(0, None), + expected_spinner=(False, None), ) @pytest.mark.parametrize( @@ -318,20 +318,20 @@ def test_info_logging_with_show_stdout_true( # The spinner should show here because show_stdout=False means # the subprocess should get logged at DEBUG level, but the passed # log level is only INFO. - (0, False, None, INFO, (None, "done", 2)), + (0, False, None, INFO, (None, "done")), # Test some cases where the spinner should not be shown. - (0, False, None, DEBUG, (None, None, 0)), + (0, False, None, DEBUG, (None, None)), # Test show_stdout=True. - (0, True, None, DEBUG, (None, None, 0)), - (0, True, None, INFO, (None, None, 0)), + (0, True, None, DEBUG, (None, None)), + (0, True, None, INFO, (None, None)), # The spinner should show here because show_stdout=True means # the subprocess should get logged at INFO level, but the passed # log level is only WARNING. - (0, True, None, WARNING, (None, "done", 2)), + (0, True, None, WARNING, (None, "done")), # Test a non-zero exit status. - (3, False, None, INFO, (InstallationSubprocessError, "error", 2)), + (3, False, None, INFO, (InstallationSubprocessError, "error")), # Test a non-zero exit status also in extra_ok_returncodes. - (3, False, (3,), INFO, (None, "done", 2)), + (3, False, (3,), INFO, (None, "done")), ], ) def test_spinner_finish( @@ -341,14 +341,13 @@ def test_spinner_finish( extra_ok_returncodes: tuple[int, ...] | None, log_level: int, caplog: pytest.LogCaptureFixture, - expected: tuple[type[Exception] | None, str | None, int], + expected: tuple[type[Exception] | None, str | None], ) -> None: """ Test that the spinner finishes correctly. """ expected_exc_type = expected[0] expected_final_status = expected[1] - expected_spin_count = expected[2] command = f'print("Hello"); print("world"); exit({exit_status})' args, spinner = self.prepare_call(caplog, log_level, command=command) @@ -368,7 +367,6 @@ def test_spinner_finish( assert exc_type == expected_exc_type assert spinner.final_status == expected_final_status - assert spinner.spin_count == expected_spin_count def test_closes_stdin(self) -> None: with pytest.raises(InstallationSubprocessError):