From 4208d5a98470f8097e7362d6ebcd54aa5cbf13c6 Mon Sep 17 00:00:00 2001 From: dw-0 Date: Sat, 11 Jul 2026 23:29:16 +0200 Subject: [PATCH] feat(spinner): add pause and resume functionality to spinner; update loading behavior in menus --- kiauh/core/menus/base_menu.py | 40 +++++-- kiauh/core/menus/tests/test_base_menu.py | 116 +++++++++++++++++++++ kiauh/core/menus/tests/test_update_menu.py | 54 +++++++++- kiauh/core/menus/update_menu.py | 20 ++-- kiauh/core/spinner.py | 44 ++++++-- kiauh/core/tests/__init__.py | 0 kiauh/core/tests/test_spinner.py | 101 ++++++++++++++++++ kiauh/main.py | 3 + kiauh/tests/test_main.py | 33 ++++-- pyproject.toml | 1 + 10 files changed, 386 insertions(+), 26 deletions(-) create mode 100644 kiauh/core/tests/__init__.py create mode 100644 kiauh/core/tests/test_spinner.py diff --git a/kiauh/core/menus/base_menu.py b/kiauh/core/menus/base_menu.py index 8a4de85c..6449d4c3 100644 --- a/kiauh/core/menus/base_menu.py +++ b/kiauh/core/menus/base_menu.py @@ -14,8 +14,9 @@ import sys import textwrap import traceback from abc import abstractmethod +from contextlib import contextmanager from enum import Enum -from typing import Dict, Type +from typing import Dict, Iterator, Type from core.logger import Logger from core.menus import FooterType, Option @@ -173,12 +174,31 @@ class BaseMenu(metaclass=PostInitCaller): raise NotImplementedError def is_loading(self, state: bool) -> None: - if not self.spinner and state: - self.spinner = Spinner(self.loading_msg) - self.spinner.start() + if state: + if self.spinner is None: + self.spinner = Spinner(self.loading_msg) + self.spinner.start() else: - self.spinner.stop() - self.spinner = None + if self.spinner is not None: + self.spinner.stop() + self.spinner = None + + @contextmanager + def pause_loading(self) -> Iterator[None]: + """Temporarily pause the loading spinner while an interactive command runs. + + Use this around subprocess calls that may prompt the user on the + terminal (for example ``sudo`` asking for a password). Pausing clears + the spinner line so the prompt is not overwritten. + """ + spinner = self.spinner + if spinner is not None: + spinner.pause() + try: + yield + finally: + if spinner is not None: + spinner.resume() def __print_menu_title(self) -> None: count = 62 - len(str(self.title_color)) - len(str(Color.RST)) @@ -233,6 +253,14 @@ class BaseMenu(metaclass=PostInitCaller): self.run() + except KeyboardInterrupt: + # Stop the spinner so the terminal is left in a clean state and the + # animation thread does not keep running during interpreter shutdown. + if self.spinner is not None: + self.spinner.stop() + self.spinner = None + raise + except Exception as e: Logger.print_error( f"An unexpected error occured:\n{e}\n{traceback.format_exc()}" diff --git a/kiauh/core/menus/tests/test_base_menu.py b/kiauh/core/menus/tests/test_base_menu.py index 3ea12eea..c0975342 100644 --- a/kiauh/core/menus/tests/test_base_menu.py +++ b/kiauh/core/menus/tests/test_base_menu.py @@ -88,6 +88,122 @@ class TestBaseMenuLifecycle: assert exits == [0] +class TestBaseMenuLoading: + def test_is_loading_starts_and_stops_spinner(self, monkeypatch) -> None: + calls: list[str | tuple[str, str]] = [] + + class FakeSpinner: + def __init__(self, message: str) -> None: + calls.append(("init", message)) + + def start(self) -> None: + calls.append("start") + + def stop(self) -> None: + calls.append("stop") + + monkeypatch.setattr("core.menus.base_menu.Spinner", FakeSpinner) + menu = ConcreteMenu() + menu.loading_msg = "Test" + + menu.is_loading(True) + assert calls[-2:] == [("init", "Test"), "start"] + + menu.is_loading(False) + assert calls[-1] == "stop" + assert menu.spinner is None + + def test_is_loading_true_twice_does_not_restart_spinner(self, monkeypatch) -> None: + calls: list[str | tuple[str, str]] = [] + + class FakeSpinner: + def __init__(self, message: str) -> None: + calls.append(("init", message)) + + def start(self) -> None: + calls.append("start") + + def stop(self) -> None: + calls.append("stop") + + monkeypatch.setattr("core.menus.base_menu.Spinner", FakeSpinner) + menu = ConcreteMenu() + + menu.is_loading(True) + menu.is_loading(True) + assert calls.count("start") == 1 + + def test_pause_loading_pauses_and_resumes_active_spinner(self, monkeypatch) -> None: + calls: list[str] = [] + + class FakeSpinner: + def __init__(self, message: str) -> None: + pass + + def start(self) -> None: + calls.append("start") + + def stop(self) -> None: + calls.append("stop") + + def pause(self) -> None: + calls.append("pause") + + def resume(self) -> None: + calls.append("resume") + + monkeypatch.setattr("core.menus.base_menu.Spinner", FakeSpinner) + menu = ConcreteMenu() + menu.loading_msg = "Test" + menu.is_loading(True) + + with menu.pause_loading(): + calls.append("inside") + + assert calls == ["start", "pause", "inside", "resume"] + + def test_pause_loading_is_noop_when_no_spinner(self) -> None: + menu = ConcreteMenu() + # should not raise + with menu.pause_loading(): + pass + + def test_keyboard_interrupt_stops_active_spinner(self, monkeypatch) -> None: + calls: list[str] = [] + + class FakeSpinner: + def __init__(self, message: str) -> None: + pass + + def start(self) -> None: + pass + + def stop(self) -> None: + calls.append("stop") + + def pause(self) -> None: + pass + + def resume(self) -> None: + pass + + monkeypatch.setattr("core.menus.base_menu.Spinner", FakeSpinner) + monkeypatch.setattr( + "core.menus.base_menu.get_selection_input", + lambda *a, **k: (_ for _ in ()).throw(KeyboardInterrupt()), + ) + + menu = ConcreteMenu() + menu.loading_msg = "Test" + menu.is_loading(True) + + with pytest.raises(KeyboardInterrupt): + menu.run() + + assert "stop" in calls + assert menu.spinner is None + + class TestMenuTitleStyle: def test_style_values(self) -> None: assert MenuTitleStyle.PLAIN.value == "plain" diff --git a/kiauh/core/menus/tests/test_update_menu.py b/kiauh/core/menus/tests/test_update_menu.py index 3a502350..b4e506d6 100644 --- a/kiauh/core/menus/tests/test_update_menu.py +++ b/kiauh/core/menus/tests/test_update_menu.py @@ -1,6 +1,7 @@ from __future__ import annotations -from typing import Any, List +from contextlib import contextmanager +from typing import Any, Generator, List import pytest from core.menus.update_menu import UpdateMenu @@ -53,6 +54,12 @@ def patched_menu(monkeypatch: pytest.MonkeyPatch) -> UpdateMenu: def stop(self): pass + def pause(self): + pass + + def resume(self): + pass + monkeypatch.setattr("core.menus.base_menu.Spinner", FakeSpinner) return UpdateMenu() @@ -156,6 +163,51 @@ class TestSystemUpdates: assert upgraded == [["curl", "git"]] +class TestSpinnerPauseDuringSystemUpdateFetch: + def test_fetch_system_package_status_pauses_spinner_during_apt_update( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The loading spinner must be paused while ``apt-get update`` runs so + that a possible sudo password prompt is not overwritten by the spinner. + """ + calls: list[str | tuple[str, bool]] = [] + + @contextmanager + def recording_pause_loading(self: UpdateMenu) -> Generator[None, None, None]: + calls.append("pause") + yield + calls.append("resume") + + monkeypatch.setattr(UpdateMenu, "pause_loading", recording_pause_loading) + monkeypatch.setattr( + "core.menus.update_menu.update_system_package_lists", + lambda silent: calls.append(("apt", silent)), + ) + monkeypatch.setattr( + "core.menus.update_menu.get_upgradable_packages", + lambda: ["curl"], + ) + monkeypatch.setattr( + "core.menus.update_menu.Logger.print_warn", lambda *a, **k: None + ) + + menu = UpdateMenu() + # the constructor already calls _fetch_system_package_update_status once + assert calls == ["pause", ("apt", True), "resume"] + + menu._fetch_system_package_update_status() + + # every fetch must pause the spinner around the apt call + assert calls == [ + "pause", + ("apt", True), + "resume", + "pause", + ("apt", True), + "resume", + ] + + class TestUpdateAll: def test_update_all_invokes_each_component_update( self, patched_menu: UpdateMenu, monkeypatch: pytest.MonkeyPatch diff --git a/kiauh/core/menus/update_menu.py b/kiauh/core/menus/update_menu.py index ce0e3bc1..12b6d032 100644 --- a/kiauh/core/menus/update_menu.py +++ b/kiauh/core/menus/update_menu.py @@ -256,7 +256,8 @@ class UpdateMenu(BaseMenu): # even when package metadata is unavailable. Dependency installation still # fails fast elsewhere. try: - update_system_package_lists(silent=True) + with self.pause_loading(): + update_system_package_lists(silent=True) except RuntimeError as exc: Logger.print_warn( "Could not update the system package lists; " @@ -274,7 +275,7 @@ class UpdateMenu(BaseMenu): elif local_version != remote_version: color = Color.YELLOW - return str(Color.apply(local_version or '-', color)) + return str(Color.apply(local_version or "-", color)) def _set_status_data(self, name: str, status_fn: Callable, *args) -> None: comp_status: ComponentStatus = status_fn(*args) @@ -331,13 +332,19 @@ class UpdateMenu(BaseMenu): elif name == "moonraker": self._set_status_data("moonraker", get_moonraker_status) elif name == "mainsail": - self._set_status_data("mainsail", get_client_status, self.mainsail_data, True) + self._set_status_data( + "mainsail", get_client_status, self.mainsail_data, True + ) elif name == "mainsail_config": - self._set_status_data("mainsail_config", get_client_config_status, self.mainsail_data) + self._set_status_data( + "mainsail_config", get_client_config_status, self.mainsail_data + ) elif name == "fluidd": self._set_status_data("fluidd", get_client_status, self.fluidd_data, True) elif name == "fluidd_config": - self._set_status_data("fluidd_config", get_client_config_status, self.fluidd_data) + self._set_status_data( + "fluidd_config", get_client_config_status, self.fluidd_data + ) elif name == "klipperscreen": self._set_status_data("klipperscreen", get_klipperscreen_status) elif name == "crowsnest": @@ -362,7 +369,8 @@ class UpdateMenu(BaseMenu): Logger.print_status("Upgrading system packages ...") - upgrade_system_packages(self.packages) + with self.pause_loading(): + upgrade_system_packages(self.packages) self._fetch_system_package_update_status() except Exception as e: Logger.print_error(f"Error upgrading system packages:\n{e}") diff --git a/kiauh/core/spinner.py b/kiauh/core/spinner.py index db603ac7..a1de107d 100644 --- a/kiauh/core/spinner.py +++ b/kiauh/core/spinner.py @@ -1,7 +1,7 @@ import sys import threading import time -from typing import List, Literal +from typing import List, Literal, Set from core.types.color import Color @@ -9,6 +9,10 @@ SpinnerColor = Literal["white", "red", "green", "yellow"] class Spinner: + # Keep track of every running spinner so a KeyboardInterrupt can stop all + # of them before the interpreter shuts down. + _registry: Set["Spinner"] = set() + def __init__( self, message: str = "Loading", @@ -17,26 +21,54 @@ class Spinner: self.message = f"{message} ..." self.interval = interval self._stop_event = threading.Event() - self._thread = threading.Thread(target=self._animate) + self._pause_event = threading.Event() + self._thread = threading.Thread(target=self._animate, daemon=True) def _animate(self) -> None: animation: List[str] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"] while not self._stop_event.is_set(): + if self._pause_event.is_set(): + time.sleep(self.interval) + continue for char in animation: + if self._pause_event.is_set() or self._stop_event.is_set(): + break sys.stdout.write(f"\r{Color.GREEN}{char}{Color.RST} {self.message}") sys.stdout.flush() time.sleep(self.interval) - if self._stop_event.is_set(): - break sys.stdout.write("\r" + " " * (len(self.message) + 1) + "\r") sys.stdout.flush() def start(self) -> None: self._stop_event.clear() if not self._thread.is_alive(): - self._thread = threading.Thread(target=self._animate) + self._thread = threading.Thread(target=self._animate, daemon=True) self._thread.start() + Spinner._registry.add(self) + + def pause(self) -> None: + """Pause animation and clear the current spinner line. + + Clearing the line is essential: if a child process (for example + ``sudo apt-get``) writes a password prompt to the terminal, the prompt + must appear on a blank line instead of being overwritten by the + spinner frame. + """ + self._pause_event.set() + sys.stdout.write("\r" + " " * (len(self.message) + 1) + "\r") + sys.stdout.flush() + + def resume(self) -> None: + self._pause_event.clear() def stop(self) -> None: self._stop_event.set() - self._thread.join() + if self._thread.is_alive(): + self._thread.join() + Spinner._registry.discard(self) + + @classmethod + def stop_all(cls) -> None: + """Stop every spinner that is still registered.""" + for spinner in list(cls._registry): + spinner.stop() diff --git a/kiauh/core/tests/__init__.py b/kiauh/core/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/core/tests/test_spinner.py b/kiauh/core/tests/test_spinner.py new file mode 100644 index 00000000..1b4ee7ab --- /dev/null +++ b/kiauh/core/tests/test_spinner.py @@ -0,0 +1,101 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2026 Dominik Willner # +# # +# This file is part of KIAUH - Klipper Installation And Update Helper # +# https://github.com/dw-0/kiauh # +# # +# This file may be distributed under the terms of the GNU GPLv3 license # +# ======================================================================= # +from __future__ import annotations + +import time + +from core.spinner import Spinner + + +class TestSpinnerLifecycle: + def test_spinner_animates_while_running(self, capsys) -> None: + spinner = Spinner(message="Working", interval=0.01) + spinner.start() + time.sleep(0.05) + spinner.stop() + + out = capsys.readouterr().out + assert "Working ..." in out + + def test_spinner_clears_line_on_stop(self, capsys) -> None: + spinner = Spinner(message="Done", interval=0.01) + spinner.start() + time.sleep(0.03) + spinner.stop() + + out = capsys.readouterr().out + # the final clear writes spaces and returns to column 0 + assert out.rstrip(" ").endswith("\r") + + +class TestSpinnerPause: + def test_pause_sets_pause_event_and_clears_line(self, capsys) -> None: + spinner = Spinner(message="Hold", interval=0.01) + spinner.start() + time.sleep(0.03) + + spinner.pause() + + assert spinner._pause_event.is_set() + out = capsys.readouterr().out + # the pause write must be a cleared line ending at column 0 + assert out.rstrip(" ").endswith("\r") + + spinner.resume() + assert not spinner._pause_event.is_set() + spinner.stop() + + def test_pause_while_already_paused_is_safe(self) -> None: + spinner = Spinner(message="Hold", interval=0.01) + spinner.start() + spinner.pause() + # pausing again must not deadlock or raise + spinner.pause() + spinner.resume() + spinner.stop() + + def test_stop_while_paused_joins_cleanly(self) -> None: + spinner = Spinner(message="Stop", interval=0.01) + spinner.start() + spinner.pause() + spinner.stop() + assert not spinner._thread.is_alive() + + +class TestSpinnerRegistry: + def test_spinner_thread_is_daemon(self) -> None: + spinner = Spinner(message="Daemon", interval=0.01) + spinner.start() + assert spinner._thread.daemon + spinner.stop() + + def test_stop_all_stops_active_spinners(self) -> None: + spinner1 = Spinner(message="One", interval=0.01) + spinner2 = Spinner(message="Two", interval=0.01) + spinner1.start() + spinner2.start() + + Spinner.stop_all() + + assert not spinner1._thread.is_alive() + assert not spinner2._thread.is_alive() + assert spinner1 not in Spinner._registry + assert spinner2 not in Spinner._registry + + def test_stop_all_is_safe_when_registry_empty(self) -> None: + Spinner.stop_all() + assert Spinner._registry == set() + + def test_stop_is_idempotent(self) -> None: + spinner = Spinner(message="Again", interval=0.01) + spinner.start() + spinner.stop() + spinner.stop() + assert spinner not in Spinner._registry + assert not spinner._thread.is_alive() diff --git a/kiauh/main.py b/kiauh/main.py index e457778f..fe75abeb 100644 --- a/kiauh/main.py +++ b/kiauh/main.py @@ -13,6 +13,7 @@ from core.cli import run_cli from core.logger import Logger from core.menus.main_menu import MainMenu from core.settings.kiauh_settings import KiauhSettings +from core.spinner import Spinner def ensure_encoding() -> None: @@ -29,6 +30,8 @@ def main() -> None: ensure_encoding() MainMenu().run() except KeyboardInterrupt: + # in case any spinner is still running, stop it before exiting + Spinner.stop_all() Logger.print_ok("\nHappy printing!\n", prefix=False) elif rc > 0: sys.exit(rc) diff --git a/kiauh/tests/test_main.py b/kiauh/tests/test_main.py index 5cc91a1a..01dd7798 100644 --- a/kiauh/tests/test_main.py +++ b/kiauh/tests/test_main.py @@ -56,9 +56,7 @@ class TestMainDispatch: assert _FakeMainMenu.instances assert all(m._run for m in _FakeMainMenu.instances) - def test_cli_success_returns_cleanly( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: + def test_cli_success_returns_cleanly(self, monkeypatch: pytest.MonkeyPatch) -> None: # rc == 0 means the CLI succeeded; the TUI must NOT start and main must # NOT call sys.exit. monkeypatch.setattr(main_module, "run_cli", lambda: 0) @@ -68,9 +66,7 @@ class TestMainDispatch: assert _FakeMainMenu.instances == [] - def test_cli_failure_exits_nonzero( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: + def test_cli_failure_exits_nonzero(self, monkeypatch: pytest.MonkeyPatch) -> None: # rc > 0 means the CLI reported a failure; main must propagate via sys.exit. monkeypatch.setattr(main_module, "run_cli", lambda: 2) _patch_tui_seeds(monkeypatch) @@ -95,4 +91,27 @@ class TestMainDispatch: monkeypatch.setattr(main_module, "ensure_encoding", lambda: None) monkeypatch.setattr(main_module, "MainMenu", _InterruptingMenu) - main_module.main() # must not raise; KeyboardInterrupt is absorbed \ No newline at end of file + main_module.main() # must not raise; KeyboardInterrupt is absorbed + + def test_tui_keyboard_interrupt_stops_spinners( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Any loading spinner still running when the user hits Ctrl-C must be + # stopped before the interpreter starts shutting down. + stopped: list[bool] = [] + + class _InterruptingMenu(_FakeMainMenu): + def run(self) -> None: + raise KeyboardInterrupt() + + monkeypatch.setattr(main_module, "run_cli", lambda: -1) + monkeypatch.setattr(main_module, "KiauhSettings", lambda: None) + monkeypatch.setattr(main_module, "ensure_encoding", lambda: None) + monkeypatch.setattr(main_module, "MainMenu", _InterruptingMenu) + monkeypatch.setattr( + main_module.Spinner, "stop_all", lambda: stopped.append(True) + ) + + main_module.main() + + assert stopped == [True] diff --git a/pyproject.toml b/pyproject.toml index eb3a0bf9..5d5f40fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,7 @@ testpaths = [ "kiauh/core/instance_manager/tests", "kiauh/core/menus/tests", "kiauh/core/services/tests", + "kiauh/core/tests", "kiauh/core/settings/tests", "kiauh/core/simple_config_parser/tests", "kiauh/extensions/tests",