diff --git a/kiauh/components/moonraker/services/moonraker_setup_service.py b/kiauh/components/moonraker/services/moonraker_setup_service.py index 9d40db7b..b2d89ab6 100644 --- a/kiauh/components/moonraker/services/moonraker_setup_service.py +++ b/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: - self.__refresh_state() + def install( + 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): - return + When called from the TUI, the Klipper instance is selected interactively. + 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: - suffix: str = self.klipper_list[0].suffix - new_inst = self.misvc.create_new_instance(suffix) - new_instances.append(new_inst) + if klipper_suffixes is not None: + for suffix in klipper_suffixes: + new_instances.append(self.misvc.create_new_instance(suffix)) + elif interactive: + selected_option: str | Klipper - else: - print_moonraker_overview( - self.klipper_list, - 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) + if len(self.klipper_list) == 1: + selected_suffix: str = self.klipper_list[0].suffix + new_instances.append(self.misvc.create_new_instance(selected_suffix)) else: - klipper_instance: Klipper | None = options.get(selected_option) - if klipper_instance is None: - raise Exception("Error selecting instance!") - new_inst = self.misvc.create_new_instance(klipper_instance.suffix) - new_instances.append(new_inst) + print_moonraker_overview( + self.klipper_list, + 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) - 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) - except Exception as e: - Logger.print_error(f"Error while installing Moonraker: {e}") - return + self._run_setup(new_instances, create_example_cfg, interactive=interactive) + except Exception: + Logger.print_error(traceback.format_exc()) + Logger.print_error("Error while installing Moonraker!") + return False - def update(self) -> None: - 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.", - ], - ) + return True - if not get_confirm("Update Moonraker now?"): - return + def update(self, interactive: bool = True) -> bool: + """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: - backup_moonraker_dir() + Returns ``True`` on success and ``False`` if the update could not be completed. + """ + 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) - git_pull_wrapper(MOONRAKER_DIR) - install_moonraker_packages() - install_python_requirements(MOONRAKER_ENV_DIR, MOONRAKER_REQ_FILE) - InstanceManager.start_all(self.moonraker_list) + if not get_confirm("Update Moonraker now?"): + return False + + self._refresh_state() + + 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( - title="Moonraker 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 Moonraker instance. - if remove_service: - Logger.print_status("Removing Moonraker instances ...") - if self.moonraker_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="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)}" - completion_msg.text.append(txt) + 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: + 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( - "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.", - ] - 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 (remove_polkit or remove_dir or remove_env) and unit_file_exists( + "moonraker", suffix="service" + ): + Logger.print_info( + "Moonraker 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_polkit: + remove_polkit_rules() + if remove_dir: + run_remove_routines(MOONRAKER_DIR) + if remove_env: + 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( - self, new_instances: List[Moonraker], create_example_cfg: bool + def _run_setup( + 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( - DialogType.CUSTOM, - custom_title="Moonraker successfully installed!", - custom_color=Color.GREEN, - content=dialog_content, - ) + if interactive: + Logger.print_dialog( + DialogType.CUSTOM, + custom_title="Moonraker successfully installed!", + 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"] - result = run( - command, - stderr=PIPE, - stdout=DEVNULL, - text=True, - ) - if result.returncode != 0 or result.stderr: - Logger.print_error(f"{result.stderr}", False) - Logger.print_error("Installing Moonraker policykit rules failed!") - return + command = [POLKIT_SCRIPT, "--disable-systemctl"] + result = run( + command, + stderr=PIPE, + stdout=DEVNULL, + text=True, + ) + if result.returncode != 0 or result.stderr: + Logger.print_error(f"{result.stderr}", False) + Logger.print_error("Installing Moonraker policykit rules failed!") + # Intentional fail-soft: polkit rules are optional on many systems + # and a failure here must not abort the whole Moonraker installation. + return - 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) + Logger.print_ok("Moonraker policykit rules successfully installed!") - 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}'") diff --git a/kiauh/components/moonraker/services/tests/__init__.py b/kiauh/components/moonraker/services/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/components/moonraker/services/tests/conftest.py b/kiauh/components/moonraker/services/tests/conftest.py new file mode 100644 index 00000000..f00fcecb --- /dev/null +++ b/kiauh/components/moonraker/services/tests/conftest.py @@ -0,0 +1,4 @@ +from __future__ import annotations + + + diff --git a/kiauh/components/moonraker/services/tests/test_moonraker_setup_service.py b/kiauh/components/moonraker/services/tests/test_moonraker_setup_service.py new file mode 100644 index 00000000..e310da89 --- /dev/null +++ b/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"] diff --git a/kiauh/components/moonraker/utils/tests/__init__.py b/kiauh/components/moonraker/utils/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/components/moonraker/utils/tests/test_utils.py b/kiauh/components/moonraker/utils/tests/test_utils.py new file mode 100644 index 00000000..452a91ba --- /dev/null +++ b/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 == {}