feat(spinner): add pause and resume functionality to spinner; update loading behavior in menus

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,13 +174,32 @@ class BaseMenu(metaclass=PostInitCaller):
         raise NotImplementedError
 
     def is_loading(self, state: bool) -> None:
-        if not self.spinner and state:
+        if state:
+            if self.spinner is None:
                 self.spinner = Spinner(self.loading_msg)
                 self.spinner.start()
         else:
+            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))
         menu_title = "╔═══════════════════════════════════════════════════════╗\n"
@@ -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()}"

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"

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

kiauh/core/menus/update_menu.py

@@ -256,6 +256,7 @@ class UpdateMenu(BaseMenu):
         # even when package metadata is unavailable. Dependency installation still
         # fails fast elsewhere.
         try:
+            with self.pause_loading():
                 update_system_package_lists(silent=True)
         except RuntimeError as exc:
             Logger.print_warn(
@@ -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,6 +369,7 @@ class UpdateMenu(BaseMenu):
 
             Logger.print_status("Upgrading system packages ...")
 
+            with self.pause_loading():
                 upgrade_system_packages(self.packages)
             self._fetch_system_package_update_status()
         except Exception as e:

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()
+        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()

kiauh/core/tests/test_spinner.py

@@ -0,0 +1,101 @@
+# ======================================================================= #
+#  Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com>        #
+#                                                                         #
+#  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()

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)

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)
@@ -96,3 +92,26 @@ class TestMainDispatch:
         monkeypatch.setattr(main_module, "MainMenu", _InterruptingMenu)
 
         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]

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",