From f34392bfd5c091e9279b1be1ff7e7a8f5ea507a4 Mon Sep 17 00:00:00 2001 From: Staubgeborener Date: Thu, 22 Aug 2024 19:31:54 +0200 Subject: [PATCH] refactor(extension): adjust code in Klipper-Backup (#502) * refactor(extension): clean up klipper backup extension code - replace subprocess.run with remove_system_servicefunction,remove_with_sudo and cmd_sysctl_service - add missing remove_moonraker_entry call - remove unnecessary Logger commands * refactor(extension): add daemon-reload and reset-failed to klipper backup extension * refactor(extension): remove unnecessary lines in klipper backup extension * refactor(extension): adjust uninstall_service function in klipper backup extension and also add some commentary * refactor(extension): remove unnecessary try except block klipper backup extension remove_with_sudo() function has it's own try except block, if an error occurs, the remaining code is not executed by raise statement anyway --- .../klipper_backup_extension.py | 182 +++++++----------- 1 file changed, 68 insertions(+), 114 deletions(-) diff --git a/kiauh/extensions/klipper_backup/klipper_backup_extension.py b/kiauh/extensions/klipper_backup/klipper_backup_extension.py index 507be06..95b34ee 100644 --- a/kiauh/extensions/klipper_backup/klipper_backup_extension.py +++ b/kiauh/extensions/klipper_backup/klipper_backup_extension.py @@ -12,9 +12,9 @@ import os import shutil import subprocess - from core.constants import SYSTEMD from core.logger import Logger +from pathlib import Path from extensions.base_extension import BaseExtension from extensions.klipper_backup import ( KLIPPERBACKUP_CONFIG_DIR, @@ -22,152 +22,106 @@ from extensions.klipper_backup import ( KLIPPERBACKUP_REPO_URL, MOONRAKER_CONF, ) -from utils.fs_utils import check_file_exist +from utils.fs_utils import check_file_exist, remove_with_sudo +from utils.git_utils import git_cmd_clone from utils.input_utils import get_confirm -from utils.sys_utils import unit_file_exists +from utils.sys_utils import cmd_sysctl_manage, remove_system_service, unit_file_exists -# noinspection PyMethodMayBeStatic class KlipperbackupExtension(BaseExtension): def remove_extension(self, **kwargs) -> None: - def uninstall_service(service_name: str, unit_type: str) -> bool: - try: - full_service_name = f"{service_name}.{unit_type}" - subprocess.run(["sudo", "systemctl", "stop", full_service_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE) - subprocess.run(["sudo", "systemctl", "disable", full_service_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE) - service_path = f"{SYSTEMD}/{full_service_name}" - subprocess.run(["sudo", "rm", service_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) - subprocess.run(["sudo", "systemctl", "daemon-reload"], check=True) - subprocess.run(["sudo", "systemctl", "reset-failed"], check=True) - return True - except subprocess.CalledProcessError: - return False - - def check_crontab_entry(entry) -> bool: - try: - crontab_content = subprocess.check_output( - ["crontab", "-l"], stderr=subprocess.DEVNULL, text=True - ) - except subprocess.CalledProcessError: - return False - for line in crontab_content.splitlines(): - if entry in line: - return True - return False - - extension_installed = check_file_exist(KLIPPERBACKUP_DIR) - if not extension_installed: + if not check_file_exist(KLIPPERBACKUP_DIR): Logger.print_info("Extension does not seem to be installed! Skipping ...") return + def uninstall_service(service_name: str, unit_type: str) -> bool: + try: + full_service_name = f"{service_name}.{unit_type}" + if unit_type == "service": + remove_system_service(full_service_name) + elif unit_type == "timer": + full_service_path: Path = SYSTEMD.joinpath(full_service_name) + Logger.print_status(f"Removing {full_service_name} ...") + remove_with_sudo(full_service_path) + Logger.print_ok(f"{service_name}.{unit_type} successfully removed!") + cmd_sysctl_manage("daemon-reload") + cmd_sysctl_manage("reset-failed") + else: + Logger.print_error(f"Unknown unit type {unit_type} of {full_service_name}") + except: + Logger.print_error(f"Failed to remove {full_service_name}: {str(e)}") + + def check_crontab_entry(entry) -> bool: + try: + crontab_content = subprocess.check_output(["crontab", "-l"], stderr=subprocess.DEVNULL, text=True) + except subprocess.CalledProcessError: + return False + return any(entry in line for line in crontab_content.splitlines()) + def remove_moonraker_entry(): original_file_path = MOONRAKER_CONF - comparison_file_path = os.path.join( - str(KLIPPERBACKUP_DIR), "install-files", "moonraker.conf" - ) - if not os.path.exists(original_file_path) or not os.path.exists( - comparison_file_path - ): + comparison_file_path = os.path.join(str(KLIPPERBACKUP_DIR), "install-files", "moonraker.conf") + if not (os.path.exists(original_file_path) and os.path.exists(comparison_file_path)): return False - with open(original_file_path, "r") as original_file, open( - comparison_file_path, "r" - ) as comparison_file: + with open(original_file_path, "r") as original_file, open(comparison_file_path, "r") as comparison_file: original_content = original_file.read() comparison_content = comparison_file.read() if comparison_content in original_content: - modified_content = original_content.replace( - comparison_content, "" - ).strip() - modified_content = "\n".join( - line for line in modified_content.split("\n") if line.strip() - ) + Logger.print_status("Removing Klipper-Backup moonraker entry ...") + modified_content = original_content.replace(comparison_content, "").strip() + modified_content = "\n".join(line for line in modified_content.split("\n") if line.strip()) with open(original_file_path, "w") as original_file: original_file.write(modified_content) - return True - else: - return False - - question = "Do you really want to remove the extension?" - if get_confirm(question, True, False): - # Remove Klipper-Backup services - service_names = [ - "klipper-backup-on-boot", - "klipper-backup-filewatch", - "klipper-backup", - ] + Logger.print_ok("Klipper-Backup moonraker entry successfully removed!") + return True + return False + if get_confirm("Do you really want to remove the extension?", True, False): + # Remove systemd timer and services + service_names = ["klipper-backup-on-boot", "klipper-backup-filewatch", "klipper-backup"] unit_types = ["timer", "service"] for service_name in service_names: - try: - Logger.print_status(f"Check whether a {service_name} unit is installed ...") + for unit_type in unit_types: + if unit_file_exists(service_name, unit_type): + uninstall_service(service_name, unit_type) - for unit_type in unit_types: - full_service_name = f"{service_name}.{unit_type}" - Logger.print_info(f"Checking for {unit_type} unit {full_service_name}") - - if unit_file_exists(service_name, unit_type): - Logger.print_info(f"{unit_type.capitalize()} unit {full_service_name} detected.") - - if uninstall_service(service_name, unit_type): - Logger.print_ok(f"The {unit_type} unit {full_service_name} has been successfully uninstalled.") - else: - Logger.print_error(f"Error uninstalling {full_service_name}.") - else: - Logger.print_info(f"No {unit_type} unit for {full_service_name} detected.") - except Exception as e: - Logger.print_error(f"Unable to remove the {service_name} service: {e}") - - # Remove Klipper-Backup cron - Logger.print_status("Check for Klipper-Backup cron entry ...") - entry_to_check = "/klipper-backup/script.sh" + # Remnove crontab entry try: - if check_crontab_entry(entry_to_check): - crontab_content = subprocess.check_output( - ["crontab", "-l"], text=True - ) - modified_content = "\n".join( - line - for line in crontab_content.splitlines() - if entry_to_check not in line - ) - if not modified_content.endswith("\n"): - modified_content += "\n" - - subprocess.run( - ["crontab", "-"], input=modified_content, text=True, check=True - ) - Logger.print_ok("The Klipper-Backup entry has been removed from the crontab.") - else: - Logger.print_info("The Klipper-Backup entry is not present in the crontab. Skipping ...") + if check_crontab_entry("/klipper-backup/script.sh"): + Logger.print_status("Removing Klipper-Backup crontab entry ...") + crontab_content = subprocess.check_output(["crontab", "-l"], text=True) + modified_content = "\n".join(line for line in crontab_content.splitlines() if "/klipper-backup/script.sh" not in line) + subprocess.run(["crontab", "-"], input=modified_content + "\n", text=True, check=True) + Logger.print_ok("Klipper-Backup crontab entry successfully removed!") except subprocess.CalledProcessError: Logger.print_error("Unable to remove the Klipper-Backup cron entry") - - # Remove Klipper-Backup - Logger.print_status(f"Removing '{KLIPPERBACKUP_DIR}' ...") + # Remove moonraker entry try: - shutil.rmtree(KLIPPERBACKUP_DIR) - config_backup_exists = check_file_exist(KLIPPERBACKUP_CONFIG_DIR) - if config_backup_exists: - shutil.rmtree(KLIPPERBACKUP_CONFIG_DIR) + remove_moonraker_entry() + except: + Logger.print_error("Unable to remove the Klipper-Backup moonraker entry") + + # Remove Klipper-backup extension + Logger.print_status("Removing Klipper-Backup extension ...") + try: + remove_with_sudo(KLIPPERBACKUP_DIR) + if check_file_exist(KLIPPERBACKUP_CONFIG_DIR): + remove_with_sudo(KLIPPERBACKUP_CONFIG_DIR) Logger.print_ok("Extension Klipper-Backup successfully removed!") - except OSError as e: - Logger.print_error(f"Unable to remove extension: {e}") + except: + Logger.print_error(f"Unable to remove Klipper-Backup extension") def install_extension(self, **kwargs) -> None: if not KLIPPERBACKUP_DIR.exists(): - subprocess.run( - ["git", "clone", str(KLIPPERBACKUP_REPO_URL), str(KLIPPERBACKUP_DIR)] - ) + git_cmd_clone(KLIPPERBACKUP_REPO_URL, KLIPPERBACKUP_DIR) subprocess.run(["chmod", "+x", str(KLIPPERBACKUP_DIR / "install.sh")]) subprocess.run([str(KLIPPERBACKUP_DIR / "install.sh")]) def update_extension(self, **kwargs) -> None: - extension_installed = check_file_exist(KLIPPERBACKUP_DIR) - if not extension_installed: + if not check_file_exist(KLIPPERBACKUP_DIR): Logger.print_info("Extension does not seem to be installed! Skipping ...") - remove_extension() - else: - subprocess.run([str(KLIPPERBACKUP_DIR / "install.sh"), "check_updates"]) + return + subprocess.run([str(KLIPPERBACKUP_DIR / "install.sh"), "check_updates"])