Add sh/cmd functions to simplify running commands

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:

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

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
+            raise err.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
 
 
 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
 

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

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