refactor: replace CURRENT_USER with get_current_user in multiple components; add user group utility functions

kiauh/components/klipper/klipper.py

@@ -23,11 +23,10 @@ from components.klipper import (
     KLIPPER_SERVICE_TEMPLATE,
     KLIPPER_UDS_NAME,
 )
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import BaseInstance
 from core.logger import Logger
 from utils.fs_utils import create_folders, get_data_dir
-from utils.sys_utils import get_service_file_path
+from utils.sys_utils import get_current_user, get_service_file_path
 
 
 # noinspection PyMethodMayBeStatic
@@ -93,7 +92,7 @@ class Klipper:
 
         service_content = template_content.replace(
             "%USER%",
-            CURRENT_USER,
+            get_current_user(),
         )
         service_content = service_content.replace(
             "%KLIPPER_DIR%",

kiauh/components/klipper/klipper_utils.py

@@ -8,8 +8,6 @@
 # ======================================================================= #
 from __future__ import annotations
 
-import grp
-import os
 import shutil
 from pathlib import Path
 from subprocess import CalledProcessError, run
@@ -28,7 +26,6 @@ from components.klipper.klipper_dialogs import (
 )
 from components.webui_client.base_data import BaseWebClient
 from components.webui_client.client_utils import create_client_config_symlink
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import SUFFIX_BLACKLIST
 from core.logger import DialogType, Logger
 from core.services.backup_service import BackupService
@@ -42,6 +39,8 @@ from utils.input_utils import get_confirm, get_number_input, get_string_input
 from utils.instance_utils import get_instances
 from utils.sys_utils import (
     cmd_sysctl_service,
+    get_current_user,
+    get_user_groups,
     install_python_packages,
     parse_packages_from_file,
 )
@@ -93,12 +92,14 @@ def check_user_groups(interactive: bool = True) -> None:
     must confirm before groups are modified. When ``interactive`` is false (the
     headless CLI path), groups are added automatically without prompting.
     """
-    user_groups = [grp.getgrgid(gid).gr_name for gid in os.getgroups()]
+    user_groups = get_user_groups()
     missing_groups = [g for g in ["tty", "dialout"] if g not in user_groups]
 
     if not missing_groups:
         return
 
+    current_user = get_current_user()
+
     if interactive:
         Logger.print_dialog(
             DialogType.ATTENTION,
@@ -116,22 +117,22 @@ def check_user_groups(interactive: bool = True) -> None:
             ],
         )
 
-        if not get_confirm(f"Add user '{CURRENT_USER}' to group(s) now?"):
+        if not get_confirm(f"Add user '{current_user}' to group(s) now?"):
             log = "Skipped adding user to required groups. You might encounter issues."
             Logger.print_warn(log)
             return
     else:
         Logger.print_info(
-            f"Adding user '{CURRENT_USER}' to required groups: "
+            f"Adding user '{current_user}' to required groups: "
             f"{', '.join(missing_groups)}"
         )
 
     try:
         for group in missing_groups:
-            Logger.print_status(f"Adding user '{CURRENT_USER}' to group {group} ...")
-            command = ["sudo", "usermod", "-a", "-G", group, CURRENT_USER]
+            Logger.print_status(f"Adding user '{current_user}' to group {group} ...")
+            command = ["sudo", "usermod", "-a", "-G", group, current_user]
             run(command, check=True)
-            Logger.print_ok(f"Group {group} assigned to user '{CURRENT_USER}'.")
+            Logger.print_ok(f"Group {group} assigned to user '{current_user}'.")
     except CalledProcessError as e:
         Logger.print_error(f"Unable to add user to usergroups: {e}")
         raise

kiauh/components/klipper/services/tests/test_klipper_setup_service.py

@@ -330,10 +330,11 @@ class TestCheckUserGroups:
     def test_interactive_mode_prompts_before_adding_user(self, monkeypatch) -> None:
         from components.klipper.klipper_utils import check_user_groups
 
-        monkeypatch.setattr("os.getgroups", lambda: [])
         monkeypatch.setattr(
-            "grp.getgrgid",
-            lambda gid: type("Group", (), {"gr_name": "tty"})(),
+            "components.klipper.klipper_utils.get_user_groups", lambda: []
+        )
+        monkeypatch.setattr(
+            "components.klipper.klipper_utils.get_current_user", lambda: "tester"
         )
 
         prompted: List[str] = []
@@ -357,10 +358,11 @@ class TestCheckUserGroups:
     def test_headless_mode_auto_adds_without_prompt(self, monkeypatch) -> None:
         from components.klipper.klipper_utils import check_user_groups
 
-        monkeypatch.setattr("os.getgroups", lambda: [])
         monkeypatch.setattr(
-            "grp.getgrgid",
-            lambda gid: type("Group", (), {"gr_name": "tty"})(),
+            "components.klipper.klipper_utils.get_user_groups", lambda: []
+        )
+        monkeypatch.setattr(
+            "components.klipper.klipper_utils.get_current_user", lambda: "tester"
         )
 
         monkeypatch.setattr(

kiauh/components/moonraker/moonraker.py

@@ -22,14 +22,13 @@ from components.moonraker import (
     MOONRAKER_LOG_NAME,
     MOONRAKER_SERVICE_TEMPLATE,
 )
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import BaseInstance
 from core.logger import Logger
 from core.simple_config_parser.simple_config_parser import (
     SimpleConfigParser,
 )
 from utils.fs_utils import create_folders
-from utils.sys_utils import get_service_file_path
+from utils.sys_utils import get_current_user, get_service_file_path
 
 
 # noinspection PyMethodMayBeStatic
@@ -98,7 +97,7 @@ class Moonraker:
 
         service_content = template_content.replace(
             "%USER%",
-            CURRENT_USER,
+            get_current_user(),
         )
         service_content = service_content.replace(
             "%MOONRAKER_DIR%",

kiauh/core/backends/tests/test_backends.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import subprocess
+import sys
 from pathlib import Path
 
 import pytest
@@ -13,12 +14,14 @@ from utils import fs_utils, sys_utils
 class TestSubprocessRunner:
     def test_run_executes_command(self) -> None:
         runner = SubprocessRunner()
-        result = runner.run(["true"])
+        result = runner.run([sys.executable, "-c", ""])
         assert result.returncode == 0
 
     def test_check_output_returns_stdout(self) -> None:
         runner = SubprocessRunner()
-        output = runner.check_output(["echo", "hello"], text=True)
+        output = runner.check_output(
+            [sys.executable, "-c", "print('hello')"], text=True
+        )
         assert "hello" in output
 
 

kiauh/core/cli/tests/test_cli.py

@@ -507,9 +507,10 @@ class TestPackaging:
     def test_pyproject_metadata_allows_editable_dev_install(self) -> None:
         project_root = Path(__file__).resolve().parents[4]
         import subprocess as sp
+        import sys
 
         result = sp.run(
-            ["python", "-m", "pip", "install", "--dry-run", "-e", ".[dev]"],
+            [sys.executable, "-m", "pip", "install", "--dry-run", "-e", ".[dev]"],
             cwd=project_root,
             capture_output=True,
             text=True,

kiauh/core/constants.py

@@ -7,8 +7,6 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
-import os
-import pwd
 from pathlib import Path
 
 # global dependencies
@@ -17,9 +15,6 @@ GLOBAL_DEPS = ["git", "wget", "curl", "unzip", "dfu-util", "python3-virtualenv"]
 # strings
 INVALID_CHOICE = "Invalid choice. Please select a valid value."
 
-# current user
-CURRENT_USER = pwd.getpwuid(os.getuid())[0]
-
 # dirs
 SYSTEMD = Path("/etc/systemd/system")
 NGINX_SITES_AVAILABLE = Path("/etc/nginx/sites-available")

kiauh/extensions/obico/moonraker_obico.py

@@ -13,7 +13,6 @@ from pathlib import Path
 from subprocess import CalledProcessError, run
 
 from components.moonraker.moonraker import Moonraker
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import BaseInstance
 from core.logger import Logger
 from core.simple_config_parser.simple_config_parser import (
@@ -30,7 +29,7 @@ from extensions.obico import (
     OBICO_SERVICE_TEMPLATE,
 )
 from utils.fs_utils import create_folders
-from utils.sys_utils import get_service_file_path
+from utils.sys_utils import get_current_user, get_service_file_path
 
 
 # noinspection PyMethodMayBeStatic
@@ -105,7 +104,7 @@ class MoonrakerObico:
 
         service_content = template_content.replace(
             "%USER%",
-            CURRENT_USER,
+            get_current_user(),
         )
         service_content = service_content.replace(
             "%OBICO_DIR%",

kiauh/extensions/octoprint/octoprint.py

@@ -13,7 +13,6 @@ from pathlib import Path
 from textwrap import dedent
 
 from components.klipper.klipper import Klipper
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import BaseInstance
 from core.logger import Logger
 from extensions.octoprint import (
@@ -22,7 +21,11 @@ from extensions.octoprint import (
     OP_LOG_NAME,
 )
 from utils.fs_utils import create_folders
-from utils.sys_utils import create_service_file, get_service_file_path
+from utils.sys_utils import (
+    create_service_file,
+    get_current_user,
+    get_service_file_path,
+)
 
 
 @dataclass
@@ -88,7 +91,7 @@ class Octoprint:
             Environment="LC_ALL=C.UTF-8"
             Environment="LANG=C.UTF-8"
             Type=simple
-            User={CURRENT_USER}
+            User={get_current_user()}
             ExecStart={octo_exec} --basedir {basedir} --config {cfg} --port={port} serve
 
             [Install]

kiauh/extensions/telegram_bot/moonraker_telegram_bot.py

@@ -13,7 +13,6 @@ from pathlib import Path
 from subprocess import CalledProcessError
 
 from components.moonraker.moonraker import Moonraker
-from core.constants import CURRENT_USER
 from core.instance_manager.base_instance import BaseInstance
 from core.logger import Logger
 from extensions.telegram_bot import (
@@ -26,7 +25,7 @@ from extensions.telegram_bot import (
     TG_BOT_SERVICE_TEMPLATE,
 )
 from utils.fs_utils import create_folders
-from utils.sys_utils import get_service_file_path
+from utils.sys_utils import get_current_user, get_service_file_path
 
 
 # noinspection PyMethodMayBeStatic
@@ -86,7 +85,7 @@ class MoonrakerTelegramBot:
 
         service_content = template_content.replace(
             "%USER%",
-            CURRENT_USER,
+            get_current_user(),
         )
         service_content = service_content.replace(
             "%TELEGRAM_BOT_DIR%",

kiauh/utils/instance_utils.py

@@ -32,6 +32,9 @@ def get_instances(
     name = convert_camelcase_to_kebabcase(instance_type.__name__)
     pattern = re.compile(f"^{name}(-[0-9a-zA-Z]+)?.service$")
 
+    if not SYSTEMD.exists():
+        return []
+
     service_list = [
         Path(SYSTEMD, service)
         for service in SYSTEMD.iterdir()

kiauh/utils/sys_utils.py

@@ -28,6 +28,29 @@ from core.logger import Logger
 from utils.fs_utils import check_file_exist, remove_with_sudo
 from utils.input_utils import get_confirm
 
+
+def get_current_user() -> str:
+    """Return the current user's login name (cross-platform)."""
+    if os.name == "posix":
+        import pwd
+
+        return pwd.getpwuid(os.getuid())[0]
+
+    import getpass
+
+    return getpass.getuser()
+
+
+def get_user_groups() -> List[str]:
+    """Return the current user's group names (empty on non-Unix)."""
+    if os.name != "posix":
+        return []
+
+    import grp
+
+    return [grp.getgrgid(gid).gr_name for gid in os.getgroups()]
+
+
 SysCtlServiceAction = Literal[
     "start",
     "stop",

kiauh/utils/tests/test_fs_utils.py

@@ -151,7 +151,7 @@ class TestRemoveFile:
         with pytest.warns(DeprecationWarning):
             remove_file(Path("/some/file"), sudo=True)
 
-        assert runs == [("sudo rm -f /some/file", True)]
+        assert runs == [(f"sudo rm -f {Path('/some/file')}", True)]
 
 
 class TestRunRemoveRoutines:

kiauh/utils/tests/test_sys_utils.py

@@ -18,11 +18,13 @@ from utils.sys_utils import (
     create_service_file,
     download_file,
     download_progress,
+    get_current_user,
     get_distro_info,
     get_ipv4_addr,
     get_service_file_path,
     get_system_timezone,
     get_upgradable_packages,
+    get_user_groups,
     install_python_packages,
     install_python_requirements,
     install_system_packages,
@@ -425,7 +427,7 @@ class TestDownloadFile:
             "utils.sys_utils.urllib.request.urlretrieve", fake_urlretrieve
         )
         download_file("http://x/file", Path("/target"), show_progress=False)
-        assert calls == [("http://x/file", "/target", None)]
+        assert calls == [("http://x/file", str(Path("/target")), None)]
 
     def test_with_progress(self, monkeypatch) -> None:
         calls: List[tuple] = []
@@ -727,3 +729,44 @@ class TestGetSystemTimezone:
             lambda *a, **k: (_ for _ in ()).throw(CalledProcessError(1, "timedatectl")),
         )
         assert get_system_timezone() == "UTC"
+
+
+class TestGetCurrentUser:
+    def test_posix_uses_pwd(self, monkeypatch) -> None:
+        import sys
+        import types
+
+        fake_pwd = types.SimpleNamespace(getpwuid=lambda uid: ["alice", "x", uid])
+        monkeypatch.setitem(sys.modules, "pwd", fake_pwd)
+        monkeypatch.setattr("utils.sys_utils.os.name", "posix")
+        monkeypatch.setattr("utils.sys_utils.os.getuid", lambda: 1000, raising=False)
+
+        assert get_current_user() == "alice"
+
+    def test_non_posix_uses_getpass(self, monkeypatch) -> None:
+        monkeypatch.setattr("utils.sys_utils.os.name", "nt")
+        monkeypatch.setattr("getpass.getuser", lambda: "bob")
+
+        assert get_current_user() == "bob"
+
+
+class TestGetUserGroups:
+    def test_posix_maps_gids_to_names(self, monkeypatch) -> None:
+        import sys
+        import types
+
+        fake_grp = types.SimpleNamespace(
+            getgrgid=lambda gid: types.SimpleNamespace(gr_name=f"g{gid}")
+        )
+        monkeypatch.setitem(sys.modules, "grp", fake_grp)
+        monkeypatch.setattr("utils.sys_utils.os.name", "posix")
+        monkeypatch.setattr(
+            "utils.sys_utils.os.getgroups", lambda: [10, 20], raising=False
+        )
+
+        assert get_user_groups() == ["g10", "g20"]
+
+    def test_non_posix_returns_empty(self, monkeypatch) -> None:
+        monkeypatch.setattr("utils.sys_utils.os.name", "nt")
+
+        assert get_user_groups() == []