feat(moonraker): support headless install/remove/update

Add interactive parameter to MoonrakerSetupService.install/remove/update and

headless tests for the service.

Add unit tests for moonraker utility functions.

kiauh/components/moonraker/services/moonraker_setup_service.py

@@ -8,8 +8,9 @@
 # ======================================================================= #
 from __future__ import annotations
 
+import traceback
 from copy import copy
-from subprocess import DEVNULL, PIPE, CalledProcessError, run
+from subprocess import DEVNULL, PIPE, run
 from typing import List
 
 from components.klipper.klipper import Klipper
@@ -104,87 +105,128 @@ class MoonrakerSetupService:
 
         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:
+    def install(
-        self.__refresh_state()
+        self,
+        klipper_suffixes: List[str] | None = None,
+        create_example_cfg: bool | None = None,
+        interactive: bool = True,
+    ) -> bool:
+        """Install Moonraker.
 
-        if not self.__check_requirements(self.klipper_list):
+        When called from the TUI, the Klipper instance is selected interactively.
-            return
+        The CLI passes explicit suffixes and ``interactive=False``.
+
+        Returns ``True`` on success and ``False`` when installation cannot proceed.
+        """
+        self._refresh_state()
+
+        if not self._check_requirements(self.klipper_list):
+            return False
 
         new_instances: List[Moonraker] = []
-        selected_option: str | Klipper
 
-        if len(self.klipper_list) == 1:
+        if klipper_suffixes is not None:
-            suffix: str = self.klipper_list[0].suffix
+            for suffix in klipper_suffixes:
-            new_inst = self.misvc.create_new_instance(suffix)
+                new_instances.append(self.misvc.create_new_instance(suffix))
-            new_instances.append(new_inst)
+        elif interactive:
+            selected_option: str | Klipper
 
-        else:
+            if len(self.klipper_list) == 1:
-            print_moonraker_overview(
+                selected_suffix: str = self.klipper_list[0].suffix
-                self.klipper_list,
+                new_instances.append(self.misvc.create_new_instance(selected_suffix))
-                self.moonraker_list,
-                show_index=True,
-                show_select_all=True,
-            )
-            options = {str(i + 1): k for i, k in enumerate(self.klipper_list)}
-            additional_options = {"a": None, "b": None}
-            options = {**options, **additional_options}
-            question = "Select Klipper instance to setup Moonraker for"
-            selected_option = get_selection_input(question, options)
-
-            if selected_option == "b":
-                Logger.print_status(EXIT_MOONRAKER_SETUP)
-                return
-
-            if selected_option == "a":
-                new_inst_list: List[Moonraker] = [
-                    self.misvc.create_new_instance(k.suffix) for k in self.klipper_list
-                ]
-                new_instances.extend(new_inst_list)
             else:
-                klipper_instance: Klipper | None = options.get(selected_option)
+                print_moonraker_overview(
-                if klipper_instance is None:
+                    self.klipper_list,
-                    raise Exception("Error selecting instance!")
+                    self.moonraker_list,
-                new_inst = self.misvc.create_new_instance(klipper_instance.suffix)
+                    show_index=True,
-                new_instances.append(new_inst)
+                    show_select_all=True,
+                )
+                options = {str(i + 1): k for i, k in enumerate(self.klipper_list)}
+                additional_options = {"a": None, "b": None}
+                options = {**options, **additional_options}
+                question = "Select Klipper instance to setup Moonraker for"
+                selected_option = get_selection_input(question, options)
 
-        create_example_cfg = get_confirm("Create example moonraker.conf?")
+                if selected_option == "b":
+                    Logger.print_status(EXIT_MOONRAKER_SETUP)
+                    return True
+
+                if selected_option == "a":
+                    new_inst_list: List[Moonraker] = [
+                        self.misvc.create_new_instance(k.suffix)
+                        for k in self.klipper_list
+                    ]
+                    new_instances.extend(new_inst_list)
+                else:
+                    klipper_instance: Klipper | None = options.get(selected_option)
+                    if klipper_instance is None:
+                        raise Exception("Error selecting instance!")
+                    new_instances.append(
+                        self.misvc.create_new_instance(klipper_instance.suffix)
+                    )
+        else:
+            for k in self.klipper_list:
+                new_instances.append(self.misvc.create_new_instance(k.suffix))
+
+        if create_example_cfg is None:
+            create_example_cfg = (
+                get_confirm("Create example moonraker.conf?") if interactive else False
+            )
 
         try:
-            self.__run_setup(new_instances, create_example_cfg)
+            self._run_setup(new_instances, create_example_cfg, interactive=interactive)
-        except Exception as e:
+        except Exception:
-            Logger.print_error(f"Error while installing Moonraker: {e}")
+            Logger.print_error(traceback.format_exc())
-            return
+            Logger.print_error("Error while installing Moonraker!")
+            return False
 
-    def update(self) -> None:
+        return True
-        Logger.print_dialog(
-            DialogType.WARNING,
-            [
-                "Be careful if there are ongoing prints running!",
-                "All Moonraker instances will be restarted during the update process and "
-                "ongoing prints COULD FAIL.",
-            ],
-        )
 
-        if not get_confirm("Update Moonraker now?"):
+    def update(self, interactive: bool = True) -> bool:
-            return
+        """Update Moonraker.
 
-        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:
+        Returns ``True`` on success and ``False`` if the update could not be completed.
-            backup_moonraker_dir()
+        """
+        if interactive:
+            Logger.print_dialog(
+                DialogType.WARNING,
+                [
+                    "Be careful if there are ongoing prints running!",
+                    "All Moonraker instances will be restarted during the update process and "
+                    "ongoing prints COULD FAIL.",
+                ],
+            )
 
-        InstanceManager.stop_all(self.moonraker_list)
+            if not get_confirm("Update Moonraker now?"):
-        git_pull_wrapper(MOONRAKER_DIR)
+                return False
-        install_moonraker_packages()
+
-        install_python_requirements(MOONRAKER_ENV_DIR, MOONRAKER_REQ_FILE)
+        self._refresh_state()
-        InstanceManager.start_all(self.moonraker_list)
+
+        try:
+            if self.settings.kiauh.backup_before_update:
+                backup_moonraker_dir()
+
+            InstanceManager.stop_all(self.moonraker_list)
+            git_pull_wrapper(MOONRAKER_DIR)
+            install_moonraker_packages()
+            install_python_requirements(MOONRAKER_ENV_DIR, MOONRAKER_REQ_FILE)
+            InstanceManager.start_all(self.moonraker_list)
+        except Exception:
+            Logger.print_error(traceback.format_exc())
+            Logger.print_error("Error while updating Moonraker!")
+            return False
+
+        return True
 
     def remove(
         self,
@@ -192,65 +234,132 @@ class MoonrakerSetupService:
         remove_dir: bool,
         remove_env: bool,
         remove_polkit: bool,
-    ) -> None:
+        *,
-        self.__refresh_state()
+        remove_all: bool = False,
+        instance_suffixes: List[str] | None = None,
+        interactive: bool = True,
+    ) -> bool:
+        """Remove Moonraker.
 
-        completion_msg = Message(
+        When called from the TUI, the user selects instances interactively. In
-            title="Moonraker Removal Process completed",
+        headless mode (``interactive=False``) the caller MUST express explicit
-            color=Color.GREEN,
+        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 Moonraker instance.
 
-        if remove_service:
+        Returns ``True`` on success and ``False`` if removal could not be completed.
-            Logger.print_status("Removing Moonraker instances ...")
+        """
-            if self.moonraker_list:
+        self._refresh_state()
-                instances_to_remove = self.__get_instances_to_remove()
+
-                self.__remove_instances(instances_to_remove)
+        try:
-                if instances_to_remove:
+            if interactive:
-                    instance_names = [
+                completion_msg = Message(
-                        i.service_file_path.stem for i in instances_to_remove
+                    title="Moonraker Removal Process completed",
+                    color=Color.GREEN,
+                )
+
+                if remove_service:
+                    Logger.print_status("Removing Moonraker instances ...")
+                    if self.moonraker_list:
+                        selected = self._get_instances_to_remove()
+                        self.__remove_instances(selected)
+                        if selected:
+                            instance_names = [
+                                i.service_file_path.stem for i in selected
+                            ]
+                            txt = f"● Moonraker instances removed: {', '.join(instance_names)}"
+                            completion_msg.text.append(txt)
+                    else:
+                        Logger.print_info(
+                            "No Moonraker Services installed! Skipped ..."
+                        )
+
+                if (remove_polkit or remove_dir or remove_env) and unit_file_exists(
+                    "moonraker", suffix="service"
+                ):
+                    completion_msg.text = [
+                        "Some Klipper services are still installed:",
+                        "● Moonraker PolicyKit rules were not removed, even though selected for removal.",
+                        f"● '{MOONRAKER_DIR}' was not removed, even though selected for removal.",
+                        f"● '{MOONRAKER_ENV_DIR}' was not removed, even though selected for removal.",
                     ]
-                    txt = f"● Moonraker instances removed: {', '.join(instance_names)}"
+                else:
-                    completion_msg.text.append(txt)
+                    if remove_polkit:
+                        Logger.print_status(
+                            "Removing all Moonraker policykit rules ..."
+                        )
+                        if remove_polkit_rules():
+                            completion_msg.text.append(
+                                "● Moonraker policykit rules removed"
+                            )
+                    if remove_dir:
+                        Logger.print_status("Removing Moonraker local repository ...")
+                        if run_remove_routines(MOONRAKER_DIR):
+                            completion_msg.text.append(
+                                "● Moonraker local repository removed"
+                            )
+                    if remove_env:
+                        Logger.print_status("Removing Moonraker Python environment ...")
+                        if run_remove_routines(MOONRAKER_ENV_DIR):
+                            completion_msg.text.append(
+                                "● Moonraker 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 Moonraker Services installed! Skipped ...")
+                if remove_service and self.moonraker_list:
+                    selected = self._select_instances_for_headless_removal(
+                        remove_all, instance_suffixes
+                    )
+                    if selected is None:
+                        Logger.print_error(
+                            "Refusing to remove Moonraker instances: no explicit "
+                            "intent. Pass remove_all=True or instance_suffixes."
+                        )
+                        return False
+                    self.__remove_instances(selected)
 
-        if (remove_polkit or remove_dir or remove_env) and unit_file_exists(
+                if (remove_polkit or remove_dir or remove_env) and unit_file_exists(
-            "moonraker", suffix="service"
+                    "moonraker", suffix="service"
-        ):
+                ):
-            completion_msg.text = [
+                    Logger.print_info(
-                "Some Klipper services are still installed:",
+                        "Moonraker services still installed; skipping repository/env removal."
-                "● Moonraker PolicyKit rules were not removed, even though selected for removal.",
+                    )
-                f"● '{MOONRAKER_DIR}' was not removed, even though selected for removal.",
+                    return True
-                f"● '{MOONRAKER_ENV_DIR}' was not removed, even though selected for removal.",
-            ]
-        else:
-            if remove_polkit:
-                Logger.print_status("Removing all Moonraker policykit rules ...")
-                if remove_polkit_rules():
-                    completion_msg.text.append("● Moonraker policykit rules removed")
-            if remove_dir:
-                Logger.print_status("Removing Moonraker local repository ...")
-                if run_remove_routines(MOONRAKER_DIR):
-                    completion_msg.text.append("● Moonraker local repository removed")
-            if remove_env:
-                Logger.print_status("Removing Moonraker Python environment ...")
-                if run_remove_routines(MOONRAKER_ENV_DIR):
-                    completion_msg.text.append("● Moonraker Python environment removed")
 
-        if completion_msg.text:
+                if remove_polkit:
-            completion_msg.text.insert(0, "The following actions were performed:")
+                    remove_polkit_rules()
-        else:
+                if remove_dir:
-            completion_msg.color = Color.YELLOW
+                    run_remove_routines(MOONRAKER_DIR)
-            completion_msg.centered = True
+                if remove_env:
-            completion_msg.text = ["Nothing to remove."]
+                    run_remove_routines(MOONRAKER_ENV_DIR)
+        except Exception:
+            Logger.print_error(traceback.format_exc())
+            Logger.print_error("Error while removing Moonraker!")
+            return False
 
-        self.msgsvc.set_message(completion_msg)
+        return True
 
-    def __run_setup(
+    def _run_setup(
-        self, new_instances: List[Moonraker], create_example_cfg: bool
+        self,
+        new_instances: List[Moonraker],
+        create_example_cfg: bool,
+        interactive: bool = True,
     ) -> None:
         check_install_dependencies()
-        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)
 
         ports_map = self.misvc.get_instance_port_map()
         for i in new_instances:
@@ -289,14 +398,21 @@ class MoonrakerSetupService:
             dialog_content.append("You can access Moonraker via the following URL:")
             dialog_content.extend(url_list)
 
-        Logger.print_dialog(
+        if interactive:
-            DialogType.CUSTOM,
+            Logger.print_dialog(
-            custom_title="Moonraker successfully installed!",
+                DialogType.CUSTOM,
-            custom_color=Color.GREEN,
+                custom_title="Moonraker successfully installed!",
-            content=dialog_content,
+                custom_color=Color.GREEN,
-        )
+                content=dialog_content,
+            )
+        else:
+            if url_list:
+                for url in url_list:
+                    Logger.print_info(url)
+            else:
+                Logger.print_info("Moonraker successfully installed!")
 
-    def __check_requirements(self, klipper_list: List[Klipper]) -> bool:
+    def _check_requirements(self, klipper_list: List[Klipper]) -> bool:
         is_klipper_installed = len(klipper_list) >= 1
         if not is_klipper_installed:
             Logger.print_warn("Klipper not installed!")
@@ -306,7 +422,7 @@ class MoonrakerSetupService:
 
         return is_klipper_installed and is_python_ok
 
-    def __install_deps(self) -> None:
+    def _install_deps(self, interactive: bool = True) -> None:
         default_repo = (MOONRAKER_REPO_URL, "master")
         repo = self.settings.moonraker.repositories
         # pull the first repo defined in kiauh.cfg or fallback to the official Moonraker repo
@@ -315,18 +431,24 @@ class MoonrakerSetupService:
 
         try:
             install_moonraker_packages()
-            if create_python_venv(MOONRAKER_ENV_DIR, False, False, self.settings.moonraker.use_python_binary):
+            if create_python_venv(
+                MOONRAKER_ENV_DIR,
+                force=False,
+                allow_access_to_system_site_packages=False,
+                use_python_binary=self.settings.moonraker.use_python_binary,
+                interactive=interactive,
+            ):
                 install_python_requirements(MOONRAKER_ENV_DIR, MOONRAKER_REQ_FILE)
                 if self.settings.moonraker.optional_speedups:
                     install_python_requirements(
                         MOONRAKER_ENV_DIR, MOONRAKER_SPEEDUPS_REQ_FILE
                     )
-            self.__install_polkit()
+            self._install_polkit()
         except Exception:
             Logger.print_error("Error during installation of Moonraker requirements!")
             raise
 
-    def __install_polkit(self) -> None:
+    def _install_polkit(self) -> None:
         Logger.print_status("Installing Moonraker policykit rules ...")
 
         legacy_file_exists = check_file_exist(POLKIT_LEGACY_FILE, True)
@@ -337,27 +459,23 @@ class MoonrakerSetupService:
             Logger.print_info("Moonraker policykit rules are already installed.")
             return
 
-        try:
+        command = [POLKIT_SCRIPT, "--disable-systemctl"]
-            command = [POLKIT_SCRIPT, "--disable-systemctl"]
+        result = run(
-            result = run(
+            command,
-                command,
+            stderr=PIPE,
-                stderr=PIPE,
+            stdout=DEVNULL,
-                stdout=DEVNULL,
+            text=True,
-                text=True,
+        )
-            )
+        if result.returncode != 0 or result.stderr:
-            if result.returncode != 0 or result.stderr:
+            Logger.print_error(f"{result.stderr}", False)
-                Logger.print_error(f"{result.stderr}", False)
+            Logger.print_error("Installing Moonraker policykit rules failed!")
-                Logger.print_error("Installing Moonraker policykit rules failed!")
+            # Intentional fail-soft: polkit rules are optional on many systems
-                return
+            # and a failure here must not abort the whole Moonraker installation.
+            return
 
-            Logger.print_ok("Moonraker policykit rules successfully installed!")
+        Logger.print_ok("Moonraker policykit rules successfully installed!")
-        except CalledProcessError as e:
-            log = (
-                f"Error while installing Moonraker policykit rules: {e.stderr.decode()}"
-            )
-            Logger.print_error(log)
 
-    def __get_instances_to_remove(self) -> List[Moonraker] | None:
+    def _get_instances_to_remove(self) -> List[Moonraker] | None:
         start_index = 1
         curr_instances: List[Moonraker] = self.moonraker_list
         instance_count = len(curr_instances)
@@ -383,6 +501,26 @@ class MoonrakerSetupService:
 
         return [instance_map[selection]]
 
+    def _select_instances_for_headless_removal(
+        self,
+        remove_all: bool,
+        instance_suffixes: List[str] | None,
+    ) -> List[Moonraker] | 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.moonraker_list)
+        if instance_suffixes:
+            wanted = set(instance_suffixes)
+            return [i for i in self.moonraker_list if i.suffix in wanted]
+        return None
+
     def __remove_instances(
         self,
         instance_list: List[Moonraker] | None,
@@ -397,7 +535,7 @@ class MoonrakerSetupService:
             InstanceManager.remove(instance)
             self.__delete_env_file(instance)
 
-        self.__refresh_state()
+        self._refresh_state()
 
     def __delete_env_file(self, instance: Moonraker):
         Logger.print_status(f"Remove '{instance.env_file}'")

kiauh/components/moonraker/services/tests/conftest.py

@@ -0,0 +1,4 @@
+from __future__ import annotations
+
+
+

kiauh/components/moonraker/services/tests/test_moonraker_setup_service.py

@@ -0,0 +1,560 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+from components.moonraker.services.moonraker_setup_service import MoonrakerSetupService
+
+
+class FakeKlipper:
+    def __init__(self, suffix: str = "") -> None:
+        self.suffix = suffix
+        name = f"klipper-{suffix}" if suffix else "klipper"
+        self.service_file_path = Path(f"/etc/systemd/system/{name}.service")
+
+
+class FakeMoonraker:
+    def __init__(self, suffix: str = "") -> None:
+        self.suffix = suffix
+        name = f"moonraker-{suffix}" if suffix else "moonraker"
+        self.service_file_path = Path(f"/etc/systemd/system/{name}.service")
+        self.env_file = Path(f"/tmp/{name}.env")
+        self.port = 7125
+        self.base = type("Base", (), {"sysd_dir": Path("/tmp")})()
+
+    def create(self) -> None:
+        pass
+
+
+class FakeKlipperInstanceService:
+    def __init__(self, instances: List[FakeKlipper]) -> None:
+        self._instances = instances
+
+    def load_instances(self) -> None:
+        pass
+
+    def get_all_instances(self) -> List[FakeKlipper]:
+        return self._instances
+
+
+class FakeMoonrakerInstanceService:
+    def __init__(self, instances: List[FakeMoonraker]) -> None:
+        self._instances = instances
+        self.created: List[str] = []
+
+    def load_instances(self) -> None:
+        pass
+
+    def get_all_instances(self) -> List[FakeMoonraker]:
+        return self._instances
+
+    def create_new_instance(self, suffix: str) -> FakeMoonraker:
+        self.created.append(suffix)
+        return FakeMoonraker(suffix)
+
+    def get_instance_by_suffix(self, suffix: str) -> FakeMoonraker:
+        return FakeMoonraker(suffix)
+
+    def get_instance_port_map(self) -> Dict[str, int]:
+        return {}
+
+
+@pytest.fixture
+def reset_service(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        MoonrakerSetupService, "_MoonrakerSetupService__cls_instance", None
+    )
+
+
+@pytest.fixture
+def patch_instance_services(
+    monkeypatch: pytest.MonkeyPatch, reset_service
+) -> Dict[str, Any]:
+    state = {"klipper": [], "moonraker": []}
+
+    def make_kis(*args, **kwargs):
+        return FakeKlipperInstanceService(state["klipper"])
+
+    def make_mis(*args, **kwargs):
+        return FakeMoonrakerInstanceService(state["moonraker"])
+
+    module = "components.moonraker.services.moonraker_setup_service"
+    monkeypatch.setattr(f"{module}.KlipperInstanceService", make_kis)
+    monkeypatch.setattr(f"{module}.MoonrakerInstanceService", make_mis)
+
+    return state
+
+
+class TestMoonrakerInstall:
+    def test_installs_for_single_klipper_instance(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("")]
+
+        setup_calls: List[Any] = []
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda self, instances, cfg, interactive=True: setup_calls.append((
+                instances,
+                cfg,
+                interactive,
+            )),
+        )
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True)
+
+        service = MoonrakerSetupService()
+        service.install()
+
+        assert len(setup_calls) == 1
+        instances, cfg, _interactive = setup_calls[0]
+        assert len(instances) == 1
+        assert instances[0].suffix == ""
+        assert cfg is True
+
+    def test_installs_for_selected_klipper_instance(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("a"), FakeKlipper("b")]
+
+        setup_calls: List[Any] = []
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda self, instances, cfg, interactive=True: setup_calls.append((
+                instances,
+                cfg,
+                interactive,
+            )),
+        )
+        monkeypatch.setattr(f"{module}.get_selection_input", lambda *a, **k: "1")
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True)
+
+        service = MoonrakerSetupService()
+        service.install()
+
+        assert len(setup_calls) == 1
+        assert setup_calls[0][0][0].suffix == "a"
+
+
+class TestMoonrakerUpdate:
+    def test_update_runs_expected_steps(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [FakeMoonraker("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        calls: List[str] = []
+
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True)
+        monkeypatch.setattr(
+            f"{module}.backup_moonraker_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_moonraker_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 = MoonrakerSetupService()
+        service.settings.kiauh.backup_before_update = True
+        service.update()
+
+        assert calls == ["backup", "stop", "pull", "packages", "requirements", "start"]
+
+
+class TestMoonrakerRemove:
+    def test_removes_selected_instance(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [FakeMoonraker("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        removed: Dict[str, List[Any]] = {"instances": [], "paths": []}
+
+        fake_instance = FakeMoonraker("")
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._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}.unit_file_exists", lambda *a, **k: False)
+        monkeypatch.setattr(f"{module}.remove_polkit_rules", lambda: True)
+        monkeypatch.setattr(
+            f"{module}.run_remove_routines",
+            lambda path: removed["paths"].append(str(path)) or True,
+        )
+
+        service = MoonrakerSetupService()
+        service.remove(
+            remove_service=True, remove_dir=True, remove_env=True, remove_polkit=True
+        )
+
+        assert removed["instances"] == [fake_instance]
+        assert any("moonraker" in p for p in removed["paths"])
+
+
+class TestMoonrakerInstallHeadless:
+    def test_headless_install_does_not_show_success_dialog(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._install_deps",
+            lambda self, interactive: None,
+        )
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True)
+        monkeypatch.setattr(f"{module}.cmd_sysctl_service", lambda *a, **k: None)
+        monkeypatch.setattr(f"{module}.cmd_sysctl_manage", lambda *a, **k: None)
+        monkeypatch.setattr(
+            f"{module}.check_install_dependencies", lambda *a, **k: None
+        )
+        monkeypatch.setattr(f"{module}.get_ipv4_addr", lambda: "127.0.0.1")
+        monkeypatch.setattr(
+            f"{module}.Logger.print_dialog",
+            lambda *a, **k: pytest.fail("should not show dialog in headless install"),
+        )
+        errors: List[str] = []
+        monkeypatch.setattr(
+            f"{module}.Logger.print_error",
+            lambda msg, *a, **k: errors.append(str(msg)),
+        )
+
+        service = MoonrakerSetupService()
+        result = service.install(interactive=False)
+
+        assert errors == [], f"unexpected errors: {errors}"
+        assert result is True
+
+    def test_installs_with_explicit_klipper_suffixes(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("a"), FakeKlipper("b")]
+
+        setup_calls: List[Any] = []
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda self, instances, cfg, interactive=True: setup_calls.append((
+                instances,
+                cfg,
+                interactive,
+            )),
+        )
+
+        service = MoonrakerSetupService()
+        result = service.install(klipper_suffixes=["a", "b"], interactive=False)
+
+        assert result is True
+        assert len(setup_calls) == 1
+        instances, cfg, interactive = setup_calls[0]
+        assert [i.suffix for i in instances] == ["a", "b"]
+        assert cfg is False
+        assert interactive is False
+
+    def test_installs_for_all_klipper_instances_when_non_interactive(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("a"), FakeKlipper("b")]
+
+        setup_calls: List[Any] = []
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda self, instances, cfg, interactive=True: setup_calls.append((
+                instances,
+                cfg,
+                interactive,
+            )),
+        )
+
+        service = MoonrakerSetupService()
+        result = service.install(interactive=False)
+
+        assert result is True
+        assert [i.suffix for i in setup_calls[0][0]] == ["a", "b"]
+        assert setup_calls[0][2] is False
+
+    def test_returns_false_when_klipper_is_missing(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = []
+
+        service = MoonrakerSetupService()
+        result = service.install(interactive=False)
+
+        assert result is False
+
+    def test_returns_false_when_setup_raises(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")),
+        )
+
+        service = MoonrakerSetupService()
+        result = service.install(interactive=False)
+
+        assert result is False
+
+
+class TestMoonrakerUpdateHeadless:
+    def test_update_runs_without_confirmation(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [FakeMoonraker("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        calls: List[str] = []
+
+        monkeypatch.setattr(
+            f"{module}.backup_moonraker_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_moonraker_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 = MoonrakerSetupService()
+        service.settings.kiauh.backup_before_update = True
+        service.update(interactive=False)
+
+        assert calls == ["backup", "stop", "pull", "packages", "requirements", "start"]
+
+    def test_update_cancelled_by_user_returns_false(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [FakeMoonraker("")]
+        pulled: List[str] = []
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.git_pull_wrapper", lambda *a, **k: pulled.append("pull")
+        )
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: False)
+
+        service = MoonrakerSetupService()
+        result = service.update(interactive=True)
+
+        assert result is False
+        assert pulled == []
+
+
+class TestMoonrakerPolkitBehavior:
+    def test_install_polkit_failure_logs_error_and_continues(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        module = "components.moonraker.services.moonraker_setup_service"
+
+        class FakeResult:
+            returncode = 1
+            stderr = "polkit install failed"
+
+        monkeypatch.setattr(f"{module}.run", lambda *a, **k: FakeResult())
+        monkeypatch.setattr(
+            f"{module}.check_file_exist", lambda p, follow_symlinks=False: False
+        )
+
+        error_messages: List[str] = []
+        monkeypatch.setattr(
+            f"{module}.Logger.print_error",
+            lambda msg, *a, **k: error_messages.append(str(msg)),
+        )
+
+        service = MoonrakerSetupService()
+        service._install_polkit()
+
+        assert any("polkit" in m.lower() for m in error_messages)
+
+    def test_install_succeeds_when_polkit_rules_fail(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["klipper"] = [FakeKlipper("")]
+
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._check_requirements",
+            lambda self, kl: True,
+        )
+
+        setup_calls: List[Any] = []
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._run_setup",
+            lambda self, instances, cfg, interactive=True: setup_calls.append((
+                instances,
+                cfg,
+                interactive,
+            )),
+        )
+
+        class FakeResult:
+            returncode = 1
+            stderr = "polkit install failed"
+
+        monkeypatch.setattr(f"{module}.run", lambda *a, **k: FakeResult())
+        monkeypatch.setattr(
+            f"{module}.check_file_exist", lambda p, follow_symlinks=False: False
+        )
+        monkeypatch.setattr(f"{module}.get_confirm", lambda *a, **k: True)
+
+        def fake_install_deps(self, interactive: bool = True) -> None:
+            self._install_polkit()
+
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._install_deps",
+            fake_install_deps,
+        )
+
+        service = MoonrakerSetupService()
+        result = service.install()
+
+        assert result is True
+        assert len(setup_calls) == 1
+
+
+class TestMoonrakerRemoveHeadless:
+    def _patch_remove_internals(self, monkeypatch, removed):
+        module = "components.moonraker.services.moonraker_setup_service"
+        monkeypatch.setattr(
+            f"{module}.InstanceManager.remove",
+            staticmethod(lambda instance: removed["instances"].append(instance.suffix)),
+        )
+        monkeypatch.setattr(
+            f"{module}.MoonrakerSetupService._refresh_state",
+            lambda self: None,
+        )
+        monkeypatch.setattr(f"{module}.unit_file_exists", lambda *a, **k: False)
+        monkeypatch.setattr(
+            f"{module}.remove_polkit_rules",
+            lambda: removed["paths"].append("polkit") or True,
+        )
+        monkeypatch.setattr(
+            f"{module}.run_remove_routines",
+            lambda path: removed["paths"].append(str(path)) or True,
+        )
+
+    def test_removes_all_instances_when_explicit_all(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [FakeMoonraker("a"), FakeMoonraker("b")]
+
+        removed: Dict[str, List[Any]] = {"instances": [], "paths": []}
+        self._patch_remove_internals(monkeypatch, removed)
+
+        service = MoonrakerSetupService()
+        service.remove(
+            remove_service=True,
+            remove_dir=True,
+            remove_env=True,
+            remove_polkit=True,
+            remove_all=True,
+            interactive=False,
+        )
+
+        assert set(removed["instances"]) == {"a", "b"}
+        assert "polkit" in removed["paths"]
+
+    def test_without_explicit_intent_removes_nothing(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        # non-interactive remove with no --all / --instance must not destroy any instance and must refuse.
+        patch_instance_services["moonraker"] = [FakeMoonraker("a"), FakeMoonraker("b")]
+
+        removed: Dict[str, List[Any]] = {"instances": [], "paths": []}
+        self._patch_remove_internals(monkeypatch, removed)
+
+        service = MoonrakerSetupService()
+        result = service.remove(
+            remove_service=True,
+            remove_dir=False,
+            remove_env=False,
+            remove_polkit=False,
+            interactive=False,
+        )
+
+        assert result is False
+        assert removed["instances"] == []
+        assert removed["paths"] == []
+
+    def test_with_instance_suffix_removes_only_matching(
+        self, patch_instance_services, monkeypatch
+    ) -> None:
+        patch_instance_services["moonraker"] = [
+            FakeMoonraker("a"),
+            FakeMoonraker("b"),
+        ]
+
+        removed: Dict[str, List[Any]] = {"instances": [], "paths": []}
+        self._patch_remove_internals(monkeypatch, removed)
+
+        service = MoonrakerSetupService()
+        service.remove(
+            remove_service=True,
+            remove_dir=False,
+            remove_env=False,
+            remove_polkit=False,
+            instance_suffixes=["a"],
+            interactive=False,
+        )
+
+        assert removed["instances"] == ["a"]

kiauh/components/moonraker/utils/tests/test_utils.py

@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+from components.moonraker.utils import utils as moonraker_utils
+from components.moonraker.utils.utils import (
+    backup_moonraker_db_dir,
+    backup_moonraker_dir,
+    create_example_moonraker_conf,
+    get_moonraker_status,
+    install_moonraker_packages,
+    load_sysdeps_json,
+    remove_polkit_rules,
+)
+
+
+class FakeMoonraker:
+    def __init__(self, suffix: str = "") -> None:
+        self.suffix = suffix
+        self.data_dir = Path(f"/tmp/moonraker{suffix}_data")
+        self.db_dir = self.data_dir.joinpath("database")
+        self.cfg_file = self.data_dir.joinpath("moonraker.conf")
+        self.base = type(
+            "Base",
+            (),
+            {
+                "cfg_dir": self.data_dir,
+                "comms_dir": self.data_dir.joinpath("comms"),
+            },
+        )()
+
+
+@pytest.fixture
+def fake_instance(tmp_path: Path) -> FakeMoonraker:
+    instance = FakeMoonraker("")
+    instance.data_dir = tmp_path / "moonraker_data"
+    instance.cfg_file = instance.data_dir / "moonraker.conf"
+    instance.db_dir = instance.data_dir / "database"
+    instance.base = type(
+        "Base",
+        (),
+        {
+            "cfg_dir": instance.data_dir,
+            "comms_dir": instance.data_dir / "comms",
+        },
+    )()
+    return instance
+
+
+class TestGetMoonrakerStatus:
+    def test_delegates_to_get_install_status(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        called: List[Any] = []
+        monkeypatch.setattr(
+            moonraker_utils,
+            "get_install_status",
+            lambda *args: called.append(args) or type("S", (), {"status": 0})(),
+        )
+
+        status = get_moonraker_status()
+
+        assert called
+        assert status.status == 0
+
+
+class TestInstallMoonrakerPackages:
+    def test_parses_deps_json_when_present(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        deps_file = tmp_path / "moonraker_deps.json"
+        deps_file.write_text(json.dumps({"debian": ["pkg1", "pkg2"]}))
+        install_script = tmp_path / "install_moonraker.sh"
+        install_script.write_text("# dummy")
+
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DEPS_JSON_FILE", deps_file)
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_INSTALL_SCRIPT", install_script)
+
+        deps: List[str] = []
+        monkeypatch.setattr(
+            moonraker_utils, "check_install_dependencies", lambda p: deps.extend(p)
+        )
+
+        class FakeParser:
+            def parse_dependencies(self, data):
+                return ["pkg1", "pkg2"]
+
+        monkeypatch.setattr(moonraker_utils, "SysDepsParser", FakeParser)
+
+        install_moonraker_packages()
+
+        assert "pkg1" in deps
+        assert "pkg2" in deps
+
+    def test_falls_back_to_install_script(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        deps_file = tmp_path / "missing.json"
+        install_script = tmp_path / "install_moonraker.sh"
+        install_script.write_text("apt-get install pkg3 pkg4\n")
+
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DEPS_JSON_FILE", deps_file)
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_INSTALL_SCRIPT", install_script)
+
+        deps: List[str] = []
+        monkeypatch.setattr(
+            moonraker_utils, "check_install_dependencies", lambda p: deps.extend(p)
+        )
+        monkeypatch.setattr(
+            moonraker_utils,
+            "parse_packages_from_file",
+            lambda p: ["pkg3", "pkg4"],
+        )
+
+        install_moonraker_packages()
+
+        assert "pkg3" in deps
+
+    def test_raises_when_no_deps_found(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        deps_file = tmp_path / "missing.json"
+        install_script = tmp_path / "missing.sh"
+
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DEPS_JSON_FILE", deps_file)
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_INSTALL_SCRIPT", install_script)
+
+        with pytest.raises(ValueError):
+            install_moonraker_packages()
+
+
+class TestRemovePolkitRules:
+    def test_returns_false_when_moonraker_dir_missing(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DIR", tmp_path / "missing")
+
+        assert remove_polkit_rules() is False
+
+    def test_returns_true_on_success(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DIR", tmp_path)
+        monkeypatch.setattr(
+            moonraker_utils,
+            "run",
+            lambda *a, **k: type("R", (), {"returncode": 0})(),
+        )
+
+        assert remove_polkit_rules() is True
+
+    def test_returns_false_on_command_failure(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DIR", tmp_path)
+
+        def fake_run(*a, **k):
+            raise moonraker_utils.CalledProcessError(1, cmd="clear")
+
+        monkeypatch.setattr(moonraker_utils, "run", fake_run)
+
+        assert remove_polkit_rules() is False
+
+
+class TestCreateExampleMoonrakerConf:
+    def test_skips_when_config_already_exists(
+        self, monkeypatch: pytest.MonkeyPatch, fake_instance: FakeMoonraker
+    ) -> None:
+        fake_instance.cfg_file.parent.mkdir(parents=True, exist_ok=True)
+        fake_instance.cfg_file.write_text("existing")
+
+        create_example_moonraker_conf(fake_instance, {})
+
+        # no changes expected
+        assert fake_instance.cfg_file.read_text() == "existing"
+
+    def test_creates_config_with_default_port(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+        fake_instance: FakeMoonraker,
+    ) -> None:
+        fake_instance.cfg_file.parent.mkdir(parents=True, exist_ok=True)
+        assets_dir = tmp_path / "assets"
+        assets_dir.mkdir()
+        template = assets_dir / "moonraker.conf"
+        template.write_text(
+            "[server]\nport: %{PORT}%\nklippy_uds_address: %{UDS}%\n"
+            "[authorization]\ntrusted_clients:\n    %{CLIENTS}%\n"
+        )
+        monkeypatch.setattr(moonraker_utils, "MODULE_PATH", tmp_path)
+        monkeypatch.setattr(
+            moonraker_utils, "get_ipv4_addr", lambda: "192.168.1.10"
+        )
+
+        create_example_moonraker_conf(fake_instance, {})
+
+        content = fake_instance.cfg_file.read_text()
+        assert "192.168.0.0/16" in content
+
+
+class TestBackupMoonrakerDir:
+    def test_backs_up_repository_and_environment(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        calls: List[Dict[str, Any]] = []
+
+        class FakeBackup:
+            def backup_directory(self, **kwargs):
+                calls.append(kwargs)
+
+        monkeypatch.setattr(moonraker_utils, "BackupService", FakeBackup)
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_DIR", tmp_path / "moonraker")
+        monkeypatch.setattr(moonraker_utils, "MOONRAKER_ENV_DIR", tmp_path / "env")
+
+        backup_moonraker_dir()
+
+        assert len(calls) == 2
+
+
+class TestBackupMoonrakerDbDir:
+    def test_backs_up_db_for_each_instance(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        calls: List[Dict[str, Any]] = []
+
+        class FakeBackup:
+            def backup_directory(self, **kwargs):
+                calls.append(kwargs)
+
+        monkeypatch.setattr(moonraker_utils, "BackupService", FakeBackup)
+        monkeypatch.setattr(
+            moonraker_utils, "get_instances", lambda model: [FakeMoonraker("")]
+        )
+
+        backup_moonraker_db_dir()
+
+        assert len(calls) == 1
+
+    def test_falls_back_to_home_dirs(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+        printer_data = tmp_path / "printer_data"
+        printer_data.mkdir()
+        printer_data.joinpath("database").mkdir()
+
+        calls: List[Dict[str, Any]] = []
+
+        class FakeBackup:
+            def backup_directory(self, **kwargs):
+                calls.append(kwargs)
+
+        monkeypatch.setattr(moonraker_utils, "BackupService", FakeBackup)
+        monkeypatch.setattr(moonraker_utils, "get_instances", lambda model: [])
+
+        backup_moonraker_db_dir()
+
+        assert len(calls) == 1
+
+
+class TestLoadSysdepsJson:
+    def test_loads_valid_json(self, tmp_path: Path) -> None:
+        file = tmp_path / "deps.json"
+        file.write_text('{"debian": ["curl"]}')
+
+        result = load_sysdeps_json(file)
+
+        assert result == {"debian": ["curl"]}
+
+    def test_returns_empty_on_invalid_json(self, tmp_path: Path) -> None:
+        file = tmp_path / "deps.json"
+        file.write_text("not json")
+
+        result = load_sysdeps_json(file)
+
+        assert result == {}