From c97587b3f5143c677464b432c7bdfd030f09a9ff Mon Sep 17 00:00:00 2001 From: Kivi Kaitaniemi Date: Fri, 26 Apr 2024 00:02:36 +0300 Subject: [PATCH] Add sh/cmd functions to simplify running commands --- src/decman/__init__.py | 66 ++++++++++++++++++++++++++++++++ src/decman/error.py | 12 ++++++ src/decman/lib/__init__.py | 51 ++++++++++-------------- src/decman/lib/aur.py | 33 ++++++++-------- tests/test_package_management.py | 3 +- 5 files changed, 118 insertions(+), 47 deletions(-) create mode 100644 src/decman/error.py diff --git a/src/decman/__init__.py b/src/decman/__init__.py index 5a7ce24..066b7f3 100644 --- a/src/decman/__init__.py +++ b/src/decman/__init__.py @@ -7,6 +7,72 @@ import pwd import grp import shutil import os +import subprocess +import decman.error + + +def sh(sh_cmd: str, + user: typing.Optional[str] = None, + env_overrides: typing.Optional[dict[str, str]] = None): + """ + Shortcut for running a shell command. + """ + if env_overrides is None: + env_overrides = {} + + env = os.environ.copy() + for var, val in env_overrides.items(): + env[var] = val + try: + + if user is None: + subprocess.run(sh_cmd, shell=True, check=True, env=env) + else: + uid = pwd.getpwnam(user).pw_uid + gid = pwd.getpwnam(user).pw_gid + + with subprocess.Popen(sh_cmd, + shell=True, + group=gid, + user=uid, + env=env) as process: + if process.wait() != 0: + raise decman.error.UserFacingError( + f"Running user shell command '{sh_cmd}' as {user} failed." + ) + except (subprocess.CalledProcessError, KeyError) as e: + raise decman.error.UserFacingError( + f"Running user shell command '{sh_cmd}' failed.") from e + + +def cmd(command: list[str], + user: typing.Optional[str] = None, + env_overrides: typing.Optional[dict[str, str]] = None): + """ + Shortcut for running a command. + """ + if env_overrides is None: + env_overrides = {} + + env = os.environ.copy() + for var, val in env_overrides.items(): + env[var] = val + + try: + if user is None: + subprocess.run(command, check=True, env=env) + else: + uid = pwd.getpwnam(user).pw_uid + gid = pwd.getpwnam(user).pw_gid + + with subprocess.Popen(command, group=gid, user=uid, + env=env) as process: + if process.wait() != 0: + raise decman.error.UserFacingError( + f"Running user command '{command}' as {user} failed.") + except (subprocess.CalledProcessError, KeyError) as e: + raise decman.error.UserFacingError( + f"Running user command '{command}' failed.") from e class File: diff --git a/src/decman/error.py b/src/decman/error.py new file mode 100644 index 0000000..699fd1f --- /dev/null +++ b/src/decman/error.py @@ -0,0 +1,12 @@ +""" +Errors used by decman. +""" + + +class UserFacingError(Exception): + """ + Execution of an important step failed and the program shouldn't continue. + """ + + def __init__(self, user_facing_msg: str): + self.user_facing_msg = user_facing_msg diff --git a/src/decman/lib/__init__.py b/src/decman/lib/__init__.py index 0b9efc2..cd9ba55 100644 --- a/src/decman/lib/__init__.py +++ b/src/decman/lib/__init__.py @@ -8,6 +8,7 @@ import json import os import typing import decman.config as conf +import decman.error as err import decman _DECMAN_MSG_TAG = "[\033[1;35mDECMAN\033[m]" @@ -181,7 +182,7 @@ class Store: with open(path, "wt", encoding="utf-8") as file: json.dump(d, file) except OSError as e: - raise UserFacingError("Failed to save store.") from e + raise err.UserFacingError("Failed to save store.") from e @staticmethod def restore() -> "Store": @@ -219,18 +220,9 @@ class Store: return store except json.JSONDecodeError as e: - raise UserFacingError("Failed to parse state json.") from e + raise err.UserFacingError("Failed to parse state json.") from e except OSError as e: - raise UserFacingError("Failed to read saved store.") from e - - -class UserFacingError(Exception): - """ - Execution of an important step failed and the program shouldn't continue. - """ - - def __init__(self, user_facing_msg: str): - self.user_facing_msg = user_facing_msg + raise err.UserFacingError("Failed to read saved store.") from e class Source: @@ -310,7 +302,7 @@ class Source: file.copy_to(target, variables) print_debug(f"Installing file to {target}.") except OSError as e: - raise UserFacingError( + raise err.UserFacingError( f"Failed to install file to {target}.") from e def install_dirs(dirs: dict[str, decman.Directory], @@ -320,7 +312,7 @@ class Source: print_debug(f"Installing directory to {target}.") directory.copy_to(target, variables) except OSError as e: - raise UserFacingError( + raise err.UserFacingError( f"Failed to install directory to {target}.") from e install_files(self.files) @@ -514,7 +506,7 @@ class Pacman: ).stdout.decode().split('\n') return packages except subprocess.CalledProcessError as error: - raise UserFacingError( + raise err.UserFacingError( f"Failed to get installed packages using '{error.cmd}'. Output: {error.stdout}." ) from error @@ -542,7 +534,7 @@ class Pacman: check=True, stdout=subprocess.PIPE).stdout.decode().strip().split('\n') except subprocess.CalledProcessError as error: - raise UserFacingError( + raise err.UserFacingError( f"Failed to get foreign packages using '{error.cmd}'. Output: {error.stdout}." ) from error @@ -550,7 +542,7 @@ class Pacman: return [(line.split(" ")[0], line.split(" ")[1]) for line in output] except IndexError as error: - raise UserFacingError( + raise err.UserFacingError( f"Failed to get foreign packages from pacman output. Output: {output}" ) from error @@ -561,7 +553,7 @@ class Pacman: try: subprocess.run(conf.commands.install_pkgs(packages), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError("Failed to install packages.") from error + raise err.UserFacingError("Failed to install packages.") from error def install_dependencies(self, deps: list[str]): """ @@ -570,7 +562,7 @@ class Pacman: try: subprocess.run(conf.commands.install_deps(deps), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError( + raise err.UserFacingError( "Failed to install dependency packages.") from error def install_files(self, files: list[str], as_explicit: list[str]): @@ -585,7 +577,7 @@ class Pacman: check=True, capture_output=conf.suppress_command_output) except subprocess.CalledProcessError as error: - raise UserFacingError( + raise err.UserFacingError( "Failed to install foreign packages.") from error def upgrade(self): @@ -595,7 +587,7 @@ class Pacman: try: subprocess.run(conf.commands.upgrade(), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError("Failed to update packages.") from error + raise err.UserFacingError("Failed to update packages.") from error def remove(self, packages: list[str]): """ @@ -604,7 +596,7 @@ class Pacman: try: subprocess.run(conf.commands.remove(packages), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError("Failed to remove packages.") from error + raise err.UserFacingError("Failed to remove packages.") from error class Systemd: @@ -622,7 +614,8 @@ class Systemd: try: subprocess.run(conf.commands.enable_units(units), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError("Failed to enable systemd units.") from error + raise err.UserFacingError( + "Failed to enable systemd units.") from error self.state.enabled_systemd_units += units def disable_units(self, units: list[str]): @@ -632,7 +625,7 @@ class Systemd: try: subprocess.run(conf.commands.disable_units(units), check=True) except subprocess.CalledProcessError as error: - raise UserFacingError( + raise err.UserFacingError( "Failed to disable systemd units.") from error for unit in units: try: @@ -649,14 +642,13 @@ class Systemd: gid = pwd.getpwnam(user).pw_gid with subprocess.Popen(conf.commands.enable_user_units(units), - start_new_session=True, group=gid, user=uid) as process: if process.wait() != 0: - raise UserFacingError( + raise err.UserFacingError( f"Failed to enable systemd units for {user}.") except KeyError as error: - raise UserFacingError( + raise err.UserFacingError( f"Failed to enable systemd units because user '{user}' doesn't exist." ) from error for unit in units: @@ -671,14 +663,13 @@ class Systemd: gid = pwd.getpwnam(user).pw_gid with subprocess.Popen(conf.commands.disable_user_units(units), - start_new_session=True, group=gid, user=uid) as process: if process.wait() != 0: - raise UserFacingError( + raise err.UserFacingError( f"Failed to disable systemd units for {user}.") except KeyError as error: - raise UserFacingError( + raise err.UserFacingError( f"Failed to disable systemd units because user '{user}' doesn't exist." ) from error diff --git a/src/decman/lib/aur.py b/src/decman/lib/aur.py index 35254c6..4117ee1 100644 --- a/src/decman/lib/aur.py +++ b/src/decman/lib/aur.py @@ -22,6 +22,7 @@ import requests import decman import decman.config as conf import decman.lib as l +import decman.error as err def strip_dependency(dep: str) -> str: @@ -208,7 +209,7 @@ class DepGraph: parent_node = self.package_nodes[parent_pkgname] if parent_node.is_pkgname_in_parents_recursive(child_pkgname): - raise l.UserFacingError( + raise err.UserFacingError( f"Foreign package dependency cycle detected involving '{child_pkgname}' \ and '{parent_pkgname}'. Foreign package dependencies are also required \ during package building and therefore dependency cycles cannot be handled.") @@ -292,7 +293,7 @@ class ExtendedPackageSearch: d = request.json() if d["type"] == "error": - raise l.UserFacingError( + raise err.UserFacingError( f"AUR RPC returned error: {d['error']}") for result in d["results"]: @@ -323,7 +324,7 @@ class ExtendedPackageSearch: l.print_debug("Request completed.") except (requests.RequestException, KeyError) as e: - raise l.UserFacingError( + raise err.UserFacingError( "Failed to fetch package information from AUR RPC.") from e def get_package_info(self, package: str) -> typing.Optional[PackageInfo]: @@ -352,7 +353,7 @@ class ExtendedPackageSearch: d = request.json() if d["type"] == "error": - raise l.UserFacingError( + raise err.UserFacingError( f"AUR RPC returned error: {d['error']}") if d["resultcount"] == 0: @@ -378,7 +379,7 @@ class ExtendedPackageSearch: return info except (requests.RequestException, KeyError) as e: - raise l.UserFacingError( + raise err.UserFacingError( "Failed to fetch package information from AUR RPC.") from e def find_provider( @@ -432,7 +433,7 @@ class ExtendedPackageSearch: d = request.json() if d["type"] == "error": - raise l.UserFacingError( + raise err.UserFacingError( f"AUR RPC returned error: {d['error']}") if d["resultcount"] == 0: @@ -451,7 +452,7 @@ class ExtendedPackageSearch: return self._choose_provider(stripped_dependency, results, "AUR") except (requests.RequestException, KeyError) as e: - raise l.UserFacingError( + raise err.UserFacingError( "Failed to fetch package information from AUR RPC.") from e def _choose_provider(self, dep: str, possible_providers: list[str], @@ -568,7 +569,7 @@ class ForeignPackageManager: info = self._search.get_package_info(pkg) if info is None: - raise l.UserFacingError(f"Failed to find package: {pkg}.") + raise err.UserFacingError(f"Failed to find package: {pkg}.") if self.should_upgrade_package(pkg, ver, info.version, upgrade_devel): @@ -626,7 +627,7 @@ class ForeignPackageManager: l.print_continuation("") if not l.prompt_confirm("Proceed?", default=True): - raise l.UserFacingError("Installing aborted.") + raise err.UserFacingError("Installing aborted.") l.print_summary("Installing foreign package dependencies from pacman.") self._pacman.install_dependencies( @@ -649,7 +650,7 @@ class ForeignPackageManager: builder.build_packages(pkgbase, packages, force) except (subprocess.CalledProcessError, OSError) as e: - raise l.UserFacingError("Failed to build packages.") from e + raise err.UserFacingError("Failed to build packages.") from e packages_to_install = list(resolved_dependencies.foreign_pkgs) packages_to_install += list(resolved_dependencies.foreign_dep_pkgs) @@ -710,7 +711,7 @@ class ForeignPackageManager: dep_info = self._search.find_provider(depname) if dep_info is None: - raise l.UserFacingError( + raise err.UserFacingError( f"Failed to find '{depname}' from AUR or user provided packages." ) @@ -728,7 +729,7 @@ class ForeignPackageManager: info = self._search.get_package_info(pkgname) if info is None: - raise l.UserFacingError( + raise err.UserFacingError( f"Failed to find '{pkgname}' from AUR or user provided packages." ) @@ -791,7 +792,7 @@ class ForeignPackageManager: ) return should_upgrade except (ValueError, subprocess.CalledProcessError) as error: - raise l.UserFacingError("Failed to compare versions.") from error + raise err.UserFacingError("Failed to compare versions.") from error class PackageBuilder: @@ -1056,7 +1057,7 @@ class PackageBuilder: continue if len(matches) != 1: - raise l.UserFacingError( + raise err.UserFacingError( f"Failed to build package '{pkgname}', because the pkg file cannot be determined." ) @@ -1096,9 +1097,9 @@ class PackageBuilder: self._store.pkgbuild_latest_reviewed_commits[ pkgbase] = commit_id else: - raise l.UserFacingError("Building aborted.") + raise err.UserFacingError("Building aborted.") except subprocess.CalledProcessError as error: - raise l.UserFacingError( + raise err.UserFacingError( f"Failed to clone and review PKGBUILD from {git_url}" ) from error diff --git a/tests/test_package_management.py b/tests/test_package_management.py index 8b1b439..f0a199c 100644 --- a/tests/test_package_management.py +++ b/tests/test_package_management.py @@ -1,7 +1,8 @@ # pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring import unittest -from decman.lib import UserFacingError, Pacman, Store +from decman.error import UserFacingError +from decman.lib import Pacman, Store from decman.lib.aur import ForeignPackageManager, DepGraph, ForeignPackage, ExtendedPackageSearch