diff --git a/kiauh/components/klipper/klipper.py b/kiauh/components/klipper/klipper.py index 305f9618..1c6df23e 100644 --- a/kiauh/components/klipper/klipper.py +++ b/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%", diff --git a/kiauh/components/klipper/klipper_utils.py b/kiauh/components/klipper/klipper_utils.py index 1f1e4293..2728a15a 100644 --- a/kiauh/components/klipper/klipper_utils.py +++ b/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 diff --git a/kiauh/components/klipper/services/tests/test_klipper_setup_service.py b/kiauh/components/klipper/services/tests/test_klipper_setup_service.py index a80547e1..05f82038 100644 --- a/kiauh/components/klipper/services/tests/test_klipper_setup_service.py +++ b/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( diff --git a/kiauh/components/moonraker/moonraker.py b/kiauh/components/moonraker/moonraker.py index 19172395..ce16c0b3 100644 --- a/kiauh/components/moonraker/moonraker.py +++ b/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%", diff --git a/kiauh/core/backends/tests/test_backends.py b/kiauh/core/backends/tests/test_backends.py index 153049a9..d492b8db 100644 --- a/kiauh/core/backends/tests/test_backends.py +++ b/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 diff --git a/kiauh/core/cli/tests/test_cli.py b/kiauh/core/cli/tests/test_cli.py index 2bb6c7d6..e7e712ef 100644 --- a/kiauh/core/cli/tests/test_cli.py +++ b/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, diff --git a/kiauh/core/constants.py b/kiauh/core/constants.py index 1589147e..4e79b003 100644 --- a/kiauh/core/constants.py +++ b/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") diff --git a/kiauh/extensions/obico/moonraker_obico.py b/kiauh/extensions/obico/moonraker_obico.py index e1039903..0e941707 100644 --- a/kiauh/extensions/obico/moonraker_obico.py +++ b/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%", diff --git a/kiauh/extensions/octoprint/octoprint.py b/kiauh/extensions/octoprint/octoprint.py index c003efe5..5b9ac87e 100644 --- a/kiauh/extensions/octoprint/octoprint.py +++ b/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] diff --git a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py b/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py index 75531fc6..194ce619 100644 --- a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py +++ b/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%", diff --git a/kiauh/utils/instance_utils.py b/kiauh/utils/instance_utils.py index b516d39f..5c4cd77f 100644 --- a/kiauh/utils/instance_utils.py +++ b/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() diff --git a/kiauh/utils/sys_utils.py b/kiauh/utils/sys_utils.py index 942de501..3a3dd028 100644 --- a/kiauh/utils/sys_utils.py +++ b/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", diff --git a/kiauh/utils/tests/test_fs_utils.py b/kiauh/utils/tests/test_fs_utils.py index d60919d1..c2e63fb6 100644 --- a/kiauh/utils/tests/test_fs_utils.py +++ b/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: diff --git a/kiauh/utils/tests/test_sys_utils.py b/kiauh/utils/tests/test_sys_utils.py index 8ab48086..94be5ef5 100644 --- a/kiauh/utils/tests/test_sys_utils.py +++ b/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() == []