From 58e3eb9f067bf876d6e0cfe34ae703c4d2f1f1a9 Mon Sep 17 00:00:00 2001 From: dw-0 Date: Sat, 11 Jul 2026 00:28:49 +0200 Subject: [PATCH] feat(klipper): support headless install/remove/update Add interactive parameter to KlipperSetupService.install/remove/update so the CLI can run without prompts. Make check_user_groups honor the interactive flag and add headless tests. --- kiauh/components/klipper/klipper_utils.py | 61 ++- .../klipper/services/klipper_setup_service.py | 347 ++++++++++---- .../klipper/services/tests/__init__.py | 0 .../tests/test_klipper_setup_service.py | 445 ++++++++++++++++++ 4 files changed, 728 insertions(+), 125 deletions(-) create mode 100644 kiauh/components/klipper/services/tests/__init__.py create mode 100644 kiauh/components/klipper/services/tests/test_klipper_setup_service.py diff --git a/kiauh/components/klipper/klipper_utils.py b/kiauh/components/klipper/klipper_utils.py index 5bf1207f..1f1e4293 100644 --- a/kiauh/components/klipper/klipper_utils.py +++ b/kiauh/components/klipper/klipper_utils.py @@ -27,9 +27,7 @@ from components.klipper.klipper_dialogs import ( print_select_instance_count_dialog, ) from components.webui_client.base_data import BaseWebClient -from components.webui_client.client_config.client_config_setup import ( - create_client_config_symlink, -) +from components.webui_client.client_utils import create_client_config_symlink from core.constants import CURRENT_USER from core.instance_manager.base_instance import SUFFIX_BLACKLIST from core.logger import DialogType, Logger @@ -88,33 +86,45 @@ def assign_custom_name(key: int, name_dict: Dict[int, str]) -> None: name_dict[key] = get_string_input(question, exclude=existing_names, regex=pattern) -def check_user_groups() -> None: +def check_user_groups(interactive: bool = True) -> None: + """Ensure the current user is in the ``tty`` and ``dialout`` groups. + + When ``interactive`` is true (the TUI path), the user is shown a dialog and + must confirm before groups are modified. When ``interactive`` is false (the + headless CLI path), groups are added automatically without prompting. + """ user_groups = [grp.getgrgid(gid).gr_name for gid in os.getgroups()] missing_groups = [g for g in ["tty", "dialout"] if g not in user_groups] if not missing_groups: return - Logger.print_dialog( - DialogType.ATTENTION, - [ - "Your current user is not in group:", - *[f"● {g}" for g in missing_groups], - "\n\n", - "It is possible that you won't be able to successfully connect and/or " - "flash the controller board without your user being a member of that " - "group. If you want to add the current user to the group(s) listed above, " - "answer with 'Y'. Else skip with 'n'.", - "\n\n", - "INFO:", - "Relog required for group assignments to take effect!", - ], - ) + if interactive: + Logger.print_dialog( + DialogType.ATTENTION, + [ + "Your current user is not in group:", + *[f"● {g}" for g in missing_groups], + "\n\n", + "It is possible that you won't be able to successfully connect and/or " + "flash the controller board without your user being a member of that " + "group. If you want to add the current user to the group(s) listed above, " + "answer with 'Y'. Else skip with 'n'.", + "\n\n", + "INFO:", + "Relog required for group assignments to take effect!", + ], + ) - if not get_confirm(f"Add user '{CURRENT_USER}' to group(s) now?"): - log = "Skipped adding user to required groups. You might encounter issues." - Logger.print_warn(log) - return + if not get_confirm(f"Add user '{CURRENT_USER}' to group(s) now?"): + log = "Skipped adding user to required groups. You might encounter issues." + Logger.print_warn(log) + return + else: + Logger.print_info( + f"Adding user '{CURRENT_USER}' to required groups: " + f"{', '.join(missing_groups)}" + ) try: for group in missing_groups: @@ -126,8 +136,9 @@ def check_user_groups() -> None: Logger.print_error(f"Unable to add user to usergroups: {e}") raise - log = "Remember to relog/restart this machine for the group(s) to be applied!" - Logger.print_warn(log) + if interactive: + log = "Remember to relog/restart this machine for the group(s) to be applied!" + Logger.print_warn(log) def handle_disruptive_system_packages() -> None: diff --git a/kiauh/components/klipper/services/klipper_setup_service.py b/kiauh/components/klipper/services/klipper_setup_service.py index 7a6598fa..5be85226 100644 --- a/kiauh/components/klipper/services/klipper_setup_service.py +++ b/kiauh/components/klipper/services/klipper_setup_service.py @@ -8,6 +8,7 @@ # ======================================================================= # from __future__ import annotations +import traceback from copy import copy from typing import Dict, List, Tuple @@ -94,132 +95,234 @@ class KlipperSetupService: self.msgsvc = MessageService() - def __refresh_state(self) -> None: + def _refresh_state(self) -> None: self.kisvc.load_instances() self.klipper_list = self.kisvc.get_all_instances() self.misvc.load_instances() self.moonraker_list = self.misvc.get_all_instances() - def install(self) -> None: - self.__refresh_state() + def install( + self, + count: int | None = None, + custom_names: Dict[int, str] | None = None, + create_example_cfg: bool | None = None, + match_moonraker: bool = False, + interactive: bool = True, + ) -> bool: + """Install Klipper. + + When called without arguments from the TUI, all choices are prompted + interactively. The CLI passes explicit values and ``interactive=False``. + + Returns ``True`` on success and ``False`` when installation cannot proceed. + """ + self._refresh_state() Logger.print_status("Installing Klipper ...") - match_moonraker: bool = False + name_dict: Dict[int, str] = {} - # if there are more moonraker instances than klipper instances, ask the user to - # match the klipper instance count to the count of moonraker instances with the same suffix - if len(self.moonraker_list) > len(self.klipper_list): - is_confirmed = self.__display_moonraker_info() - if not is_confirmed: + if custom_names is not None: + name_dict = custom_names + elif match_moonraker and len(self.moonraker_list) > len(self.klipper_list): + if interactive: + if not self._display_moonraker_info(): + Logger.print_status(EXIT_KLIPPER_SETUP) + return True + name_dict = { + i: moonraker.suffix for i, moonraker in enumerate(self.moonraker_list) + } + elif count is not None: + name_dict = {i: "" for i in range(count)} + elif interactive: + install_count, name_dict = self.__get_install_count_and_name_dict() + + if install_count == 0: Logger.print_status(EXIT_KLIPPER_SETUP) - return - match_moonraker = True + return True - install_count, name_dict = self.__get_install_count_and_name_dict() + is_multi_install = install_count > 1 or ( + len(name_dict) >= 1 and install_count >= 1 + ) + if not name_dict and install_count == 1: + name_dict = {0: ""} + elif is_multi_install and not self.__count_from_moonraker_match( + install_count, name_dict + ): + use_custom_names = self.__use_custom_names_or_go_back() + if use_custom_names is None: + Logger.print_status(EXIT_KLIPPER_SETUP) + return True - if install_count == 0: - Logger.print_status(EXIT_KLIPPER_SETUP) - return - - is_multi_install = install_count > 1 or ( - len(name_dict) >= 1 and install_count >= 1 - ) - if not name_dict and install_count == 1: + self.__handle_instance_names(install_count, name_dict, use_custom_names) + else: name_dict = {0: ""} - elif is_multi_install and not match_moonraker: - custom_names = self.__use_custom_names_or_go_back() - if custom_names is None: - Logger.print_status(EXIT_KLIPPER_SETUP) - return - self.__handle_instance_names(install_count, name_dict, custom_names) + if not name_dict: + Logger.print_status(EXIT_KLIPPER_SETUP) + return True + + if create_example_cfg is None: + create_example_cfg = ( + get_confirm("Create example printer.cfg?") if interactive else False + ) - create_example_cfg = get_confirm("Create example printer.cfg?") - # run the actual installation try: - self.__run_setup(name_dict, create_example_cfg) - except Exception as e: - Logger.print_error(e) + self.__run_setup(name_dict, create_example_cfg, interactive=interactive) + except Exception: + Logger.print_error(traceback.format_exc()) Logger.print_error("Klipper installation failed!") - return + return False - def update(self) -> None: - Logger.print_dialog( - DialogType.WARNING, - [ - "Do NOT continue if there are ongoing prints running!", - "All Klipper instances will be restarted during the update process and " - "ongoing prints WILL FAIL.", - ], - ) + return True - if not get_confirm("Update Klipper now?"): - return + def update(self, interactive: bool = True) -> bool: + """Update Klipper. - self.__refresh_state() + When called from the TUI, a warning and confirmation are shown. The CLI + passes ``interactive=False`` to run silently. - if self.settings.kiauh.backup_before_update: - backup_klipper_dir() + Returns ``True`` on success and ``False`` if the update could not be completed. + """ + if interactive: + Logger.print_dialog( + DialogType.WARNING, + [ + "Do NOT continue if there are ongoing prints running!", + "All Klipper instances will be restarted during the update process and " + "ongoing prints WILL FAIL.", + ], + ) - InstanceManager.stop_all(self.klipper_list) - git_pull_wrapper(KLIPPER_DIR) - install_klipper_packages() - install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) - InstanceManager.start_all(self.klipper_list) + if not get_confirm("Update Klipper now?"): + return False + + self._refresh_state() + + try: + if self.settings.kiauh.backup_before_update: + backup_klipper_dir() + + InstanceManager.stop_all(self.klipper_list) + git_pull_wrapper(KLIPPER_DIR) + install_klipper_packages() + install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) + InstanceManager.start_all(self.klipper_list) + except Exception: + Logger.print_error(traceback.format_exc()) + Logger.print_error("Error while updating Klipper!") + return False + + return True def remove( self, remove_service: bool, remove_dir: bool, remove_env: bool, - ) -> None: - self.__refresh_state() + *, + remove_all: bool = False, + instance_suffixes: List[str] | None = None, + interactive: bool = True, + ) -> bool: + """Remove Klipper. - completion_msg = Message( - title="Klipper Removal Process completed", - color=Color.GREEN, - ) + When called from the TUI, the user selects instances interactively. In + headless mode (``interactive=False``) the caller MUST express explicit + intent: pass ``remove_all=True`` to wipe every instance or + ``instance_suffixes=[...]`` to remove a named subset. Without explicit + intent the service refuses and removes nothing so a CLI user can never + accidentally destroy every Klipper instance on the machine. - if remove_service: - Logger.print_status("Removing Klipper instances ...") - if self.klipper_list: - instances_to_remove = self.__get_instances_to_remove() - self.__remove_instances(instances_to_remove) - if instances_to_remove: - instance_names = [ - i.service_file_path.stem for i in instances_to_remove + Returns ``True`` on success and ``False`` if removal could not be completed. + """ + self._refresh_state() + + try: + if interactive: + completion_msg = Message( + title="Klipper Removal Process completed", + color=Color.GREEN, + ) + + if remove_service: + Logger.print_status("Removing Klipper instances ...") + if self.klipper_list: + instances_to_remove = self._get_instances_to_remove() + self.__remove_instances(instances_to_remove) + if instances_to_remove: + instance_names = [ + i.service_file_path.stem for i in instances_to_remove + ] + txt = f"● Klipper instances removed: {', '.join(instance_names)}" + completion_msg.text.append(txt) + else: + Logger.print_info("No Klipper Services installed! Skipped ...") + + if (remove_dir or remove_env) and unit_file_exists( + "klipper", suffix="service" + ): + completion_msg.text = [ + "Some Klipper services are still installed:", + f"● '{KLIPPER_DIR}' was not removed, even though selected for removal.", + f"● '{KLIPPER_ENV_DIR}' was not removed, even though selected for removal.", ] - txt = f"● Klipper instances removed: {', '.join(instance_names)}" - completion_msg.text.append(txt) + else: + if remove_dir: + Logger.print_status("Removing Klipper local repository ...") + if run_remove_routines(KLIPPER_DIR): + completion_msg.text.append( + "● Klipper local repository removed" + ) + if remove_env: + Logger.print_status("Removing Klipper Python environment ...") + if run_remove_routines(KLIPPER_ENV_DIR): + completion_msg.text.append( + "● Klipper Python environment removed" + ) + + if completion_msg.text: + completion_msg.text.insert( + 0, "The following actions were performed:" + ) + else: + completion_msg.color = Color.YELLOW + completion_msg.centered = True + completion_msg.text = ["Nothing to remove."] + + self.msgsvc.set_message(completion_msg) else: - Logger.print_info("No Klipper Services installed! Skipped ...") + if remove_service and self.klipper_list: + selected = self._select_instances_for_headless_removal( + remove_all, instance_suffixes + ) + if selected is None: + Logger.print_error( + "Refusing to remove Klipper instances: no explicit " + "intent. Pass remove_all=True or instance_suffixes." + ) + return False + self.__remove_instances(selected) - if (remove_dir or remove_env) and unit_file_exists("klipper", suffix="service"): - completion_msg.text = [ - "Some Klipper services are still installed:", - f"● '{KLIPPER_DIR}' was not removed, even though selected for removal.", - f"● '{KLIPPER_ENV_DIR}' was not removed, even though selected for removal.", - ] - else: - if remove_dir: - Logger.print_status("Removing Klipper local repository ...") - if run_remove_routines(KLIPPER_DIR): - completion_msg.text.append("● Klipper local repository removed") - if remove_env: - Logger.print_status("Removing Klipper Python environment ...") - if run_remove_routines(KLIPPER_ENV_DIR): - completion_msg.text.append("● Klipper Python environment removed") + if (remove_dir or remove_env) and unit_file_exists( + "klipper", suffix="service" + ): + Logger.print_info( + "Klipper services still installed; skipping repository/env removal." + ) + return True - if completion_msg.text: - completion_msg.text.insert(0, "The following actions were performed:") - else: - completion_msg.color = Color.YELLOW - completion_msg.centered = True - completion_msg.text = ["Nothing to remove."] + if remove_dir: + run_remove_routines(KLIPPER_DIR) + if remove_env: + run_remove_routines(KLIPPER_ENV_DIR) + except Exception: + Logger.print_error(traceback.format_exc()) + Logger.print_error("Error while removing Klipper!") + return False - self.msgsvc.set_message(completion_msg) + return True def __get_install_count_and_name_dict(self) -> Tuple[int, Dict[int, str]]: install_count: int | None @@ -240,9 +343,16 @@ class KlipperSetupService: return install_count, name_dict - def __run_setup(self, name_dict: Dict[int, str], create_example_cfg: bool) -> None: + def __run_setup( + self, + name_dict: Dict[int, str], + create_example_cfg: bool, + interactive: bool = True, + ) -> None: if not self.klipper_list: - self.__install_deps() + # Only create a fresh venv when none exists; existing venvs are + # preserved in both TUI and CLI modes. + self.__install_deps(interactive=interactive) for i in name_dict: # skip this iteration if there is already an instance with the name @@ -266,9 +376,9 @@ class KlipperSetupService: handle_disruptive_system_packages() # step 5: check for required group membership - check_user_groups() + check_user_groups(interactive=interactive) - def __install_deps(self) -> None: + def __install_deps(self, interactive: bool = True) -> None: default_repo = (KLIPPER_REPO_URL, "master") repo = self.settings.klipper.repositories # pull the first repo defined in kiauh.cfg or fallback to the official Klipper repo @@ -277,13 +387,19 @@ class KlipperSetupService: try: install_klipper_packages() - if create_python_venv(KLIPPER_ENV_DIR, False, False, self.settings.klipper.use_python_binary): + if create_python_venv( + KLIPPER_ENV_DIR, + force=False, + allow_access_to_system_site_packages=False, + use_python_binary=self.settings.klipper.use_python_binary, + interactive=interactive, + ): install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) except Exception: Logger.print_error("Error during installation of Klipper requirements!") raise - def __display_moonraker_info(self) -> bool: + def _display_moonraker_info(self) -> bool: # todo: only show the klipper instances that are not already installed Logger.print_dialog( DialogType.INFO, @@ -308,6 +424,17 @@ class KlipperSetupService: else: name_dict[key] = str(len(name_dict) + 1) + def __count_from_moonraker_match( + self, install_count: int, name_dict: Dict[int, str] + ) -> bool: + """Return True when the count/names came from matching Moonraker instances.""" + if len(self.moonraker_list) <= len(self.klipper_list): + return False + if install_count != len(self.moonraker_list): + return False + expected = [m.suffix for m in self.moonraker_list] + return list(name_dict.values()) == expected + def __use_custom_names_or_go_back(self) -> bool | None: print_select_custom_name_dialog() _input: bool | None = get_confirm( @@ -317,7 +444,7 @@ class KlipperSetupService: ) return _input - def __get_instances_to_remove(self) -> List[Klipper] | None: + def _get_instances_to_remove(self) -> List[Klipper] | None: start_index = 1 curr_instances: List[Klipper] = self.klipper_list instance_count = len(curr_instances) @@ -341,6 +468,26 @@ class KlipperSetupService: return [instance_map[selection]] + def _select_instances_for_headless_removal( + self, + remove_all: bool, + instance_suffixes: List[str] | None, + ) -> List[Klipper] | None: + """Resolve which instances to remove in headless mode. + + Returns the list of instances to remove, or ``None`` when the caller did + not express explicit intent (no ``remove_all`` and no ``instance_suffixes``). + A ``None`` return is the "refuse to wipe everything" signal the CLI path + relies on. Kept as a single-public-seam helper (no name mangling) so + tests can patch it without brittle ``_Class__method`` access. + """ + if remove_all: + return list(self.klipper_list) + if instance_suffixes: + wanted = set(instance_suffixes) + return [i for i in self.klipper_list if i.suffix in wanted] + return None + def __remove_instances( self, instance_list: List[Klipper] | None, @@ -353,11 +500,11 @@ class KlipperSetupService: f"Removing instance {instance.service_file_path.stem} ..." ) InstanceManager.remove(instance) - self.__delete_klipper_env_file(instance) + self._delete_klipper_env_file(instance) - self.__refresh_state() + self._refresh_state() - def __delete_klipper_env_file(self, instance: Klipper): + def _delete_klipper_env_file(self, instance: Klipper): Logger.print_status(f"Remove '{instance.env_file}'") if not instance.env_file.exists(): msg = f"Env file in {instance.base.sysd_dir} not found. Skipped ..." diff --git a/kiauh/components/klipper/services/tests/__init__.py b/kiauh/components/klipper/services/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/components/klipper/services/tests/test_klipper_setup_service.py b/kiauh/components/klipper/services/tests/test_klipper_setup_service.py new file mode 100644 index 00000000..a80547e1 --- /dev/null +++ b/kiauh/components/klipper/services/tests/test_klipper_setup_service.py @@ -0,0 +1,445 @@ +from __future__ import annotations + +from typing import Any, Dict, List + +import pytest +from components.klipper.services.klipper_setup_service import KlipperSetupService + + +@pytest.fixture +def reset_service(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(KlipperSetupService, "_KlipperSetupService__cls_instance", None) + + +class FakeKlipper: + def __init__(self, suffix: str = "") -> None: + self.suffix = suffix + self.create_calls: List[Any] = [] + + def create(self) -> None: + self.create_calls.append(True) + + +@pytest.fixture +def patched_install_deps( + monkeypatch: pytest.MonkeyPatch, reset_service +) -> Dict[str, List[Any]]: + calls: Dict[str, List[Any]] = { + "klipper_create": [], + "enable": [], + "start": [], + } + + module = "components.klipper.services.klipper_setup_service" + + def fake_klipper(suffix: str = "") -> FakeKlipper: + instance = FakeKlipper(suffix) + calls["klipper_create"].append(instance) + return instance + + monkeypatch.setattr(f"{module}.Klipper", fake_klipper) + monkeypatch.setattr( + f"{module}.InstanceManager.enable", + staticmethod(lambda instance: calls["enable"].append(instance.suffix)), + ) + monkeypatch.setattr( + f"{module}.InstanceManager.start", + staticmethod(lambda instance: calls["start"].append(instance.suffix)), + ) + monkeypatch.setattr(f"{module}.git_clone_wrapper", lambda *a, **k: None) + monkeypatch.setattr(f"{module}.install_klipper_packages", lambda: None) + monkeypatch.setattr(f"{module}.create_python_venv", lambda *a, **k: True) + monkeypatch.setattr(f"{module}.install_python_requirements", lambda *a, **k: None) + monkeypatch.setattr(f"{module}.handle_disruptive_system_packages", lambda: None) + monkeypatch.setattr(f"{module}.check_user_groups", lambda interactive=True: None) + monkeypatch.setattr(f"{module}.cmd_sysctl_manage", lambda *a, **k: None) + + return calls + + +class TestKlipperInstallHeadless: + def test_installs_single_instance_by_default( + self, patched_install_deps, monkeypatch + ) -> None: + service = KlipperSetupService() + service.install(interactive=False) + + assert len(patched_install_deps["klipper_create"]) == 1 + assert patched_install_deps["enable"] == [""] + assert patched_install_deps["start"] == [""] + + def test_installs_multiple_instances_by_count( + self, patched_install_deps, monkeypatch + ) -> None: + service = KlipperSetupService() + service.install(count=2, interactive=False) + + assert len(patched_install_deps["klipper_create"]) == 2 + assert patched_install_deps["enable"] == ["", ""] + assert patched_install_deps["start"] == ["", ""] + + def test_installs_with_custom_names( + self, patched_install_deps, monkeypatch + ) -> None: + service = KlipperSetupService() + service.install(custom_names={0: "a", 1: "b"}, interactive=False) + + assert len(patched_install_deps["klipper_create"]) == 2 + instances = patched_install_deps["klipper_create"] + assert instances[0].suffix == "a" + assert instances[1].suffix == "b" + + +class TestKlipperRemoveHeadless: + def _make_fake_instance(self, suffix: str = ""): + Path = __import__("pathlib").Path + return type( + "FakeInstance", + (), + { + "suffix": suffix, + "service_file_path": Path(f"klipper-{suffix}.service"), + "env_file": Path("/tmp/klipper.env"), + "base": type("Base", (), {"sysd_dir": Path("/tmp")})(), + }, + )() + + def _patch_remove_internals(self, monkeypatch, removed): + module = "components.klipper.services.klipper_setup_service" + monkeypatch.setattr( + f"{module}.KlipperSetupService._refresh_state", + lambda self: None, + ) + monkeypatch.setattr( + f"{module}.InstanceManager.remove", + staticmethod(lambda instance: removed["instances"].append(instance.suffix)), + ) + monkeypatch.setattr(f"{module}.unit_file_exists", lambda *a, **k: False) + monkeypatch.setattr( + f"{module}.run_remove_routines", + lambda path: removed["paths"].append(str(path)) or True, + ) + + def test_removes_explicit_all_services_and_files( + self, reset_service, monkeypatch + ) -> None: + removed: Dict[str, List[Any]] = {"instances": [], "paths": []} + self._patch_remove_internals(monkeypatch, removed) + fake_instance = self._make_fake_instance("") + service = KlipperSetupService() + service.klipper_list = [fake_instance] + service.remove( + remove_service=True, + remove_dir=True, + remove_env=True, + remove_all=True, + interactive=False, + ) + + assert removed["instances"] == [""] + assert any("klipper" in p for p in removed["paths"]) + + def test_without_explicit_intent_removes_nothing( + self, reset_service, monkeypatch + ) -> None: + # non-interactive remove with no --all and no --instance must + # NOT call InstanceManager.remove or run_remove_routines and must + # refuse with a non-zero (False) result. + removed: Dict[str, List[Any]] = {"instances": [], "paths": []} + self._patch_remove_internals(monkeypatch, removed) + fake_instance = self._make_fake_instance("a") + service = KlipperSetupService() + service.klipper_list = [fake_instance] + result = service.remove( + remove_service=True, + remove_dir=False, + remove_env=False, + interactive=False, + ) + + assert result is False + assert removed["instances"] == [] + assert removed["paths"] == [] + + def test_with_instance_suffix_removes_only_matching( + self, reset_service, monkeypatch + ) -> None: + removed: Dict[str, List[Any]] = {"instances": [], "paths": []} + self._patch_remove_internals(monkeypatch, removed) + service = KlipperSetupService() + service.klipper_list = [ + self._make_fake_instance("a"), + self._make_fake_instance("b"), + ] + service.remove( + remove_service=True, + remove_dir=False, + remove_env=False, + instance_suffixes=["a"], + interactive=False, + ) + + assert removed["instances"] == ["a"] + + +class TestKlipperUpdateHeadless: + def test_update_runs_expected_steps(self, reset_service, monkeypatch) -> None: + module = "components.klipper.services.klipper_setup_service" + calls: List[str] = [] + + monkeypatch.setattr( + f"{module}.backup_klipper_dir", lambda: calls.append("backup") + ) + monkeypatch.setattr( + f"{module}.InstanceManager.stop_all", + staticmethod(lambda instances: calls.append("stop")), + ) + monkeypatch.setattr( + f"{module}.git_pull_wrapper", lambda *a, **k: calls.append("pull") + ) + monkeypatch.setattr( + f"{module}.install_klipper_packages", lambda: calls.append("packages") + ) + monkeypatch.setattr( + f"{module}.install_python_requirements", + lambda *a, **k: calls.append("requirements"), + ) + monkeypatch.setattr( + f"{module}.InstanceManager.start_all", + staticmethod(lambda instances: calls.append("start")), + ) + + service = KlipperSetupService() + service.settings.kiauh.backup_before_update = True + result = service.update(interactive=False) + + assert result is True + assert calls == ["backup", "stop", "pull", "packages", "requirements", "start"] + + def test_update_cancelled_by_user_returns_false( + self, reset_service, monkeypatch + ) -> None: + module = "components.klipper.services.klipper_setup_service" + pulled: List[str] = [] + monkeypatch.setattr( + f"{module}.git_pull_wrapper", lambda *a, **k: pulled.append("pull") + ) + monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: False) + + service = KlipperSetupService() + result = service.update(interactive=True) + + assert result is False + assert pulled == [] + + +class FakeMoonraker: + def __init__(self, suffix: str = "") -> None: + self.suffix = suffix + + +class TestKlipperInteractiveMoonrakerMatch: + def test_installs_exactly_one_klipper_per_moonraker( + self, reset_service, patched_install_deps, monkeypatch + ) -> None: + module = "components.klipper.services.klipper_setup_service" + monkeypatch.setattr( + f"{module}.KlipperSetupService._refresh_state", + lambda self: None, + ) + monkeypatch.setattr( + f"{module}.KlipperSetupService._display_moonraker_info", + lambda self: True, + ) + monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True) + + service = KlipperSetupService() + service.klipper_list = [] + service.moonraker_list = [FakeMoonraker(""), FakeMoonraker("b")] + + result = service.install(interactive=True, create_example_cfg=False) + + assert result is True + assert len(patched_install_deps["klipper_create"]) == 2 + instances = patched_install_deps["klipper_create"] + assert instances[0].suffix == "" + assert instances[1].suffix == "b" + + def test_headless_match_moonraker_skips_dialog( + self, reset_service, patched_install_deps, monkeypatch + ) -> None: + module = "components.klipper.services.klipper_setup_service" + monkeypatch.setattr( + f"{module}.KlipperSetupService._refresh_state", + lambda self: None, + ) + dialog_calls: List[Any] = [] + monkeypatch.setattr( + f"{module}.KlipperSetupService._display_moonraker_info", + lambda self: dialog_calls.append(True) or False, + ) + + service = KlipperSetupService() + service.klipper_list = [] + service.moonraker_list = [FakeMoonraker("a"), FakeMoonraker("b")] + + result = service.install(match_moonraker=True, interactive=False) + + assert result is True + assert dialog_calls == [] + assert len(patched_install_deps["klipper_create"]) == 2 + instances = patched_install_deps["klipper_create"] + assert [i.suffix for i in instances] == ["a", "b"] + + +class TestKlipperVenvNonDestructive: + """a headless install must not force-recreate an existing Klipper + venv. ``__install_deps`` must pass ``force=False`` and ``interactive=False`` + to ``create_python_venv`` so an existing venv is left untouched (no prompt, + no ``rmtree``).""" + + def test_headless_install_does_not_force_recreate_venv( + self, reset_service, monkeypatch + ) -> None: + module = "components.klipper.services.klipper_setup_service" + monkeypatch.setattr( + f"{module}.KlipperSetupService._refresh_state", + lambda self: None, + ) + venv_calls: List[Any] = [] + monkeypatch.setattr( + f"{module}.create_python_venv", + lambda *a, **k: venv_calls.append(k) or True, + ) + monkeypatch.setattr(f"{module}.git_clone_wrapper", lambda *a, **k: None) + monkeypatch.setattr(f"{module}.install_klipper_packages", lambda: None) + monkeypatch.setattr( + f"{module}.install_python_requirements", lambda *a, **k: None + ) + + service = KlipperSetupService() + service.klipper_list = [] + service.install(interactive=False) + + assert venv_calls, "create_python_venv should have been called" + assert venv_calls[0]["force"] is False + assert venv_calls[0]["interactive"] is False + + +class TestCheckUserGroups: + def test_interactive_mode_prompts_before_adding_user(self, monkeypatch) -> None: + from components.klipper.klipper_utils import check_user_groups + + monkeypatch.setattr("os.getgroups", lambda: []) + monkeypatch.setattr( + "grp.getgrgid", + lambda gid: type("Group", (), {"gr_name": "tty"})(), + ) + + prompted: List[str] = [] + monkeypatch.setattr( + "components.klipper.klipper_utils.get_confirm", + lambda question, *a, **k: prompted.append(question) or True, + ) + run_calls: List[List[str]] = [] + monkeypatch.setattr( + "components.klipper.klipper_utils.run", + lambda cmd, **kwargs: ( + run_calls.append(cmd) or type("R", (), {"returncode": 0})() + ), + ) + + check_user_groups(interactive=True) + + assert any("group" in q.lower() for q in prompted) + assert run_calls + + def test_headless_mode_auto_adds_without_prompt(self, monkeypatch) -> None: + from components.klipper.klipper_utils import check_user_groups + + monkeypatch.setattr("os.getgroups", lambda: []) + monkeypatch.setattr( + "grp.getgrgid", + lambda gid: type("Group", (), {"gr_name": "tty"})(), + ) + + monkeypatch.setattr( + "components.klipper.klipper_utils.get_confirm", + lambda *a, **k: pytest.fail("should not prompt in headless mode"), + ) + run_calls: List[List[str]] = [] + monkeypatch.setattr( + "components.klipper.klipper_utils.run", + lambda cmd, **kwargs: ( + run_calls.append(cmd) or type("R", (), {"returncode": 0})() + ), + ) + + check_user_groups(interactive=False) + + assert run_calls + + +class TestKlipperRemoveInteractiveTui: + """Exercise the interactive (TUI) remove branch so the message-assembly + path stays covered: the TUI path must remain unchanged.""" + + def _make_fake_instance(self, suffix: str = ""): + Path = __import__("pathlib").Path + return type( + "FakeInstance", + (), + { + "suffix": suffix, + "service_file_path": Path(f"klipper-{suffix}.service"), + "env_file": Path("/tmp/klipper.env"), + "base": type("Base", (), {"sysd_dir": Path("/tmp")})(), + }, + )() + + def test_interactive_remove_sets_completion_message( + self, reset_service, monkeypatch + ) -> None: + module = "components.klipper.services.klipper_setup_service" + fake_instance = self._make_fake_instance("a") + removed: Dict[str, List[Any]] = {"instances": [], "paths": []} + + monkeypatch.setattr( + f"{module}.KlipperSetupService._refresh_state", + lambda self: None, + ) + monkeypatch.setattr( + f"{module}.KlipperSetupService._get_instances_to_remove", + lambda self: [fake_instance], + ) + monkeypatch.setattr( + f"{module}.InstanceManager.remove", + staticmethod(lambda instance: removed["instances"].append(instance)), + ) + monkeypatch.setattr( + f"{module}.KlipperSetupService._delete_klipper_env_file", + lambda self, inst: None, + ) + monkeypatch.setattr(f"{module}.unit_file_exists", lambda *a, **k: False) + monkeypatch.setattr( + f"{module}.run_remove_routines", + lambda path: removed["paths"].append(str(path)) or True, + ) + set_messages: List[Any] = [] + monkeypatch.setattr( + f"{module}.MessageService", + lambda: type( + "MS", (), {"set_message": lambda self, m: set_messages.append(m)} + )(), + ) + + service = KlipperSetupService() + service.klipper_list = [fake_instance] + result = service.remove( + remove_service=True, remove_dir=True, remove_env=True, interactive=True + ) + + assert result is True + assert removed["instances"] == [fake_instance] + assert set_messages, "TUI remove must set the completion message" + assert any("klipper-a" in line for line in set_messages[0].text)