feat(KIAUH): move filesystem related methods to own module

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

kiauh/core/instance_manager/base_instance.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from typing import List, Union, Optional, Type, TypeVar
 
 from kiauh.utils.constants import SYSTEMD, CURRENT_USER
-from kiauh.utils.system_utils import create_directory
+from kiauh.utils.filesystem_utils import create_directory
 
 B = TypeVar(name="B", bound="BaseInstance", covariant=True)
 

kiauh/modules/mainsail/mainsail_setup.py

@@ -24,35 +24,31 @@ from kiauh.modules.mainsail import (
     MAINSAIL_CONFIG_DIR,
     MAINSAIL_CONFIG_REPO_URL,
     MODULE_PATH,
-)
+    )
 from kiauh.modules.mainsail.mainsail_dialogs import (
     print_moonraker_not_found_dialog,
     print_mainsail_already_installed_dialog,
     print_install_mainsail_config_dialog,
     print_mainsail_port_select_dialog,
-)
+    )
 from kiauh.modules.mainsail.mainsail_utils import (
     restore_config_json,
     enable_mainsail_remotemode,
     backup_config_json,
     symlink_webui_nginx_log,
-)
+    )
 from kiauh.modules.moonraker.moonraker import Moonraker
 from kiauh.utils.common import check_install_dependencies
+from kiauh.utils.filesystem_utils import unzip, create_upstream_nginx_cfg, create_common_vars_nginx_cfg, \
+    delete_default_nginx_cfg, create_nginx_cfg, enable_nginx_cfg
 from kiauh.utils.input_utils import get_confirm, get_number_input
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import (
     download_file,
-    unzip,
-    create_upstream_nginx_cfg,
-    create_nginx_cfg,
-    delete_default_nginx_cfg,
-    enable_nginx_cfg,
     set_nginx_permissions,
     get_ipv4_addr,
     control_systemd_service,
-    create_common_vars_nginx_cfg,
+    )
-)
 
 
 def run_mainsail_installation() -> None:

kiauh/modules/moonraker/moonraker_setup.py

@@ -38,6 +38,7 @@ from kiauh.modules.moonraker import (
 from kiauh.modules.moonraker.moonraker import Moonraker
 from kiauh.modules.moonraker.moonraker_dialogs import print_moonraker_overview
 from kiauh.modules.moonraker.moonraker_utils import create_example_moonraker_conf
+from kiauh.utils.filesystem_utils import check_file_exist
 from kiauh.utils.input_utils import (
     get_confirm,
     get_selection_input,
@@ -49,7 +50,6 @@ from kiauh.utils.system_utils import (
     install_python_requirements,
     update_system_package_lists,
     install_system_packages,
-    check_file_exists,
 )
 
 
@@ -181,9 +181,9 @@ def install_moonraker_packages(moonraker_dir: Path) -> None:
 def install_moonraker_polkit() -> None:
     Logger.print_status("Installing Moonraker policykit rules ...")
 
-    legacy_file_exists = check_file_exists(Path(POLKIT_LEGACY_FILE))
+    legacy_file_exists = check_file_exist(Path(POLKIT_LEGACY_FILE))
-    polkit_file_exists = check_file_exists(Path(POLKIT_FILE))
+    polkit_file_exists = check_file_exist(Path(POLKIT_FILE))
-    usr_file_exists = check_file_exists(Path(POLKIT_USR_FILE))
+    usr_file_exists = check_file_exist(Path(POLKIT_USR_FILE))
 
     if legacy_file_exists or (polkit_file_exists and usr_file_exists):
         Logger.print_info("Moonraker policykit rules are already installed.")

kiauh/utils/filesystem_utils.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-
 # ======================================================================= #
 #  Copyright (C) 2020 - 2023 Dominik Willner <th33xitus@gmail.com>        #
 #                                                                         #
@@ -9,13 +8,46 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
+import os
+import shutil
 import subprocess
 from pathlib import Path
+from zipfile import ZipFile
 
-
+from kiauh.utils import NGINX_SITES_AVAILABLE, NGINX_SITES_ENABLED, MODULE_PATH, NGINX_CONFD
 from kiauh.utils.logger import Logger
 
 
+def check_file_exist(file_path: Path) -> bool:
+    """
+    Helper function for checking the existence of a file where
+    elevated permissions are required |
+    :param file_path: the absolute path of the file to check
+    :return: True if file exists, otherwise False
+    """
+    try:
+        command = ["sudo", "find", file_path]
+        subprocess.check_output(command, stderr=subprocess.DEVNULL)
+        return True
+    except subprocess.CalledProcessError:
+        return False
+
+
+def create_directory(_dir: Path) -> None:
+    """
+    Helper function for creating a directory or skipping if it already exists |
+    :param _dir: the directory to create
+    :return: None
+    """
+    try:
+        if not os.path.isdir(_dir):
+            os.makedirs(_dir, exist_ok=True)
+            Logger.print_ok(f"Created directory: {_dir}")
+    except OSError as e:
+        Logger.print_error(f"Error creating folder: {e}")
+        raise
+
+
 def remove_file(file_path: Path, sudo=False) -> None:
     try:
         command = f"{'sudo ' if sudo else ''}rm -f {file_path}"
@@ -24,3 +56,112 @@ def remove_file(file_path: Path, sudo=False) -> None:
         log = f"Cannot remove file {file_path}: {e.stderr.decode()}"
         Logger.print_error(log)
         raise
+
+def unzip(file: str, target_dir: str) -> None:
+    """
+    Helper function to unzip a zip-archive into a target directory |
+    :param file: the zip-file to unzip
+    :param target_dir: the target directory to extract the files into
+    :return: None
+    """
+    with ZipFile(file, "r") as _zip:
+        _zip.extractall(target_dir)
+
+
+def create_upstream_nginx_cfg() -> None:
+    """
+    Creates an upstream.conf in /etc/nginx/conf.d
+    :return: None
+    """
+    source = os.path.join(MODULE_PATH, "res", "upstreams.conf")
+    target = os.path.join(NGINX_CONFD, "upstreams.conf")
+    try:
+        command = ["sudo", "cp", source, target]
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
+        log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def create_common_vars_nginx_cfg() -> None:
+    """
+    Creates a common_vars.conf in /etc/nginx/conf.d
+    :return: None
+    """
+    source = os.path.join(MODULE_PATH, "res", "common_vars.conf")
+    target = os.path.join(NGINX_CONFD, "common_vars.conf")
+    try:
+        command = ["sudo", "cp", source, target]
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
+        log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def create_nginx_cfg(name: str, port: int, root_dir: str) -> None:
+    """
+    Creates an NGINX config from a template file and replaces all placeholders
+    :param name: name of the config to create
+    :param port: listen port
+    :param root_dir: directory of the static files
+    :return: None
+    """
+    tmp = f"{Path.home()}/{name}.tmp"
+    shutil.copy(os.path.join(MODULE_PATH, "res", "nginx_cfg"), tmp)
+    with open(tmp, "r+") as f:
+        content = f.read()
+        content = content.replace("%NAME%", name)
+        content = content.replace("%PORT%", str(port))
+        content = content.replace("%ROOT_DIR%", root_dir)
+        f.seek(0)
+        f.write(content)
+        f.truncate()
+
+    target = os.path.join(NGINX_SITES_AVAILABLE, name)
+    try:
+        command = ["sudo", "mv", tmp, target]
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
+        log = f"Unable to create '{target}': {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def delete_default_nginx_cfg() -> None:
+    """
+    Deletes a default NGINX config
+    :return: None
+    """
+    default_cfg = Path("/etc/nginx/sites-enabled/default")
+    if not check_file_exist(default_cfg):
+        return
+
+    try:
+        command = ["sudo", "rm", default_cfg]
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
+        log = f"Unable to delete '{default_cfg}': {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise
+
+
+def enable_nginx_cfg(name: str) -> None:
+    """
+    Helper method to enable an NGINX config |
+    :param name: name of the config to enable
+    :return: None
+    """
+    source = os.path.join(NGINX_SITES_AVAILABLE, name)
+    target = os.path.join(NGINX_SITES_ENABLED, name)
+    if check_file_exist(Path(target)):
+        return
+
+    try:
+        command = ["sudo", "ln", "-s", source, target]
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
+        log = f"Unable to create symlink: {e.stderr.decode()}"
+        Logger.print_error(log)
+        raise

kiauh/utils/system_utils.py

@@ -27,6 +27,7 @@ from kiauh.utils import (
     NGINX_SITES_AVAILABLE,
     NGINX_SITES_ENABLED,
 )
+from kiauh.utils.filesystem_utils import check_file_exist
 from kiauh.utils.input_utils import get_confirm
 from kiauh.utils.logger import Logger
 
@@ -215,21 +216,6 @@ def install_system_packages(packages: List[str]) -> None:
         kill(f"Error installing packages:\n{e.stderr.decode()}")
 
 
-def create_directory(_dir: Path) -> None:
-    """
-    Helper function for creating a directory or skipping if it already exists |
-    :param _dir: the directory to create
-    :return: None
-    """
-    try:
-        if not os.path.isdir(_dir):
-            os.makedirs(_dir, exist_ok=True)
-            Logger.print_ok(f"Created directory: {_dir}")
-    except OSError as e:
-        Logger.print_error(f"Error creating folder: {e}")
-        raise
-
-
 def mask_system_service(service_name: str) -> None:
     """
     Mask a system service to prevent it from starting |
@@ -245,21 +231,6 @@ def mask_system_service(service_name: str) -> None:
         raise
 
 
-def check_file_exists(file_path: Path) -> bool:
-    """
-    Helper function for checking the existence of a file where
-    elevated permissions are required |
-    :param file_path: the absolute path of the file to check
-    :return: True if file exists, otherwise False
-    """
-    try:
-        command = ["sudo", "find", file_path]
-        subprocess.check_output(command, stderr=subprocess.DEVNULL)
-        return True
-    except subprocess.CalledProcessError:
-        return False
-
-
 # this feels hacky and not quite right, but for now it works
 # see: https://stackoverflow.com/questions/166506/finding-local-ip-addresses-using-pythons-stdlib
 def get_ipv4_addr() -> str:
@@ -327,116 +298,6 @@ def download_progress(block_num, block_size, total_size) -> None:
     sys.stdout.flush()
 
 
-def unzip(file: str, target_dir: str) -> None:
-    """
-    Helper function to unzip a zip-archive into a target directory |
-    :param file: the zip-file to unzip
-    :param target_dir: the target directory to extract the files into
-    :return: None
-    """
-    with ZipFile(file, "r") as _zip:
-        _zip.extractall(target_dir)
-
-
-def create_upstream_nginx_cfg() -> None:
-    """
-    Creates an upstream.conf in /etc/nginx/conf.d
-    :return: None
-    """
-    source = os.path.join(MODULE_PATH, "res", "upstreams.conf")
-    target = os.path.join(NGINX_CONFD, "upstreams.conf")
-    try:
-        command = ["sudo", "cp", source, target]
-        subprocess.run(command, stderr=subprocess.PIPE, check=True)
-    except subprocess.CalledProcessError as e:
-        log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
-        Logger.print_error(log)
-        raise
-
-
-def create_common_vars_nginx_cfg() -> None:
-    """
-    Creates a common_vars.conf in /etc/nginx/conf.d
-    :return: None
-    """
-    source = os.path.join(MODULE_PATH, "res", "common_vars.conf")
-    target = os.path.join(NGINX_CONFD, "common_vars.conf")
-    try:
-        command = ["sudo", "cp", source, target]
-        subprocess.run(command, stderr=subprocess.PIPE, check=True)
-    except subprocess.CalledProcessError as e:
-        log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
-        Logger.print_error(log)
-        raise
-
-
-def create_nginx_cfg(name: str, port: int, root_dir: str) -> None:
-    """
-    Creates an NGINX config from a template file and replaces all placeholders
-    :param name: name of the config to create
-    :param port: listen port
-    :param root_dir: directory of the static files
-    :return: None
-    """
-    tmp = f"{Path.home()}/{name}.tmp"
-    shutil.copy(os.path.join(MODULE_PATH, "res", "nginx_cfg"), tmp)
-    with open(tmp, "r+") as f:
-        content = f.read()
-        content = content.replace("%NAME%", name)
-        content = content.replace("%PORT%", str(port))
-        content = content.replace("%ROOT_DIR%", root_dir)
-        f.seek(0)
-        f.write(content)
-        f.truncate()
-
-    target = os.path.join(NGINX_SITES_AVAILABLE, name)
-    try:
-        command = ["sudo", "mv", tmp, target]
-        subprocess.run(command, stderr=subprocess.PIPE, check=True)
-    except subprocess.CalledProcessError as e:
-        log = f"Unable to create '{target}': {e.stderr.decode()}"
-        Logger.print_error(log)
-        raise
-
-
-def delete_default_nginx_cfg() -> None:
-    """
-    Deletes a default NGINX config
-    :return: None
-    """
-    default_cfg = Path("/etc/nginx/sites-enabled/default")
-    if not check_file_exists(default_cfg):
-        return
-
-    try:
-        command = ["sudo", "rm", default_cfg]
-        subprocess.run(command, stderr=subprocess.PIPE, check=True)
-    except subprocess.CalledProcessError as e:
-        log = f"Unable to delete '{default_cfg}': {e.stderr.decode()}"
-        Logger.print_error(log)
-        raise
-
-
-def enable_nginx_cfg(name: str) -> None:
-    """
-    Helper method to enable an NGINX config |
-    :param name: name of the config to enable
-    :return: None
-    """
-    source = os.path.join(NGINX_SITES_AVAILABLE, name)
-    target = os.path.join(NGINX_SITES_ENABLED, name)
-    if check_file_exists(Path(target)):
-        return
-
-    try:
-        command = ["sudo", "ln", "-s", source, target]
-        subprocess.run(command, stderr=subprocess.PIPE, check=True)
-    except subprocess.CalledProcessError as e:
-        log = f"Unable to create symlink: {e.stderr.decode()}"
-        Logger.print_error(log)
-        raise
-
-
 def set_nginx_permissions() -> None:
     """
     Check if permissions of the users home directory