commit 5d102f15f05fbbcaed5d0ea48dbe16e29a443e09 Author: Kivi Kaitaniemi Date: Mon Apr 22 19:27:40 2024 +0300 Initial commit - Added basic pacman/systemd wrappers - Added way to manage AUR and user packages diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c1b5289 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ + +build/ +*.egg-info/ + +venv/ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d358ea5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,17 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "decman" +version = "0.0.1" +description = "Declarative package/configuration manager for Arch Linux" +authors = [ + {name = "Kivi Kaitaniemi"} +] +dependencies = [ + "requests" +] + +[project.scripts] +decman = "decman.app:main" diff --git a/src/decman/__init__.py b/src/decman/__init__.py new file mode 100644 index 0000000..a8c9c20 --- /dev/null +++ b/src/decman/__init__.py @@ -0,0 +1,3 @@ +""" +Module for writing system configurations for decman. +""" diff --git a/src/decman/app.py b/src/decman/app.py new file mode 100644 index 0000000..891c53c --- /dev/null +++ b/src/decman/app.py @@ -0,0 +1,38 @@ +""" +Module containing the CLI Application. +""" + +import os +import sys + +from decman.lib import AUR, Pacman, Systemd, Store, print_error + + +def main(): + """ + Main entry for the CLI app + """ + + aur = AUR() + + print(aur.get_package_info("zapzap")) + + if not is_root(): + print_error("Not running as root. Please run decman as root.") + sys.exit(1) + + p = Pacman() + + print(p.get_versioned_foreign_packages()) + + +def is_root() -> bool: + """ + Returns True if the process is running as root. + """ + + return os.geteuid() == 0 + + +if __name__ == "__main__": + main() diff --git a/src/decman/config.py b/src/decman/config.py new file mode 100644 index 0000000..b4e1600 --- /dev/null +++ b/src/decman/config.py @@ -0,0 +1,187 @@ +""" +Module for decman configuration options. + +NOTE: Do NOT use from imports as global variables might not work as you expect. + +Only use: + +import decman.config + +or + +import decman.config as whatever + +-- Configuring commands -- + +Commands are stored as methods in the Commands-class. +The global variable 'commands' of this module is an instance of the Commands-class. + +To change the defalts, create a new child class of the Commands-class and set the 'commands' +variable to an instance of your class. Look in the README for examples. +""" + +import typing + + +class Commands: + """ + Default commands. + """ + + def list_pkgs(self) -> list[str]: + """ + Running this command outputs a newline seperated list of explicitly installed packages. + """ + return ["pacman", "-Qeq", "--color=never"] + + def list_foreign_pkgs_versioned(self) -> list[str]: + """ + Running this command outputs a newline seperated list of installed packages and their + versions that are not from pacman repositories. + """ + return ["pacman", "-Qm", "--color=never"] + + def install_pkgs(self, pkgs: list[str]) -> list[str]: + """ + Running this command installs the given packages from pacman repositories. + """ + return ["pacman", "-S", "--asexplicit"] + pkgs + + def install_files(self, pkg_files: list[str]) -> list[str]: + """ + Running this command installs the given packages files. + """ + return ["pacman", "-U", "--asdeps"] + pkg_files + + def set_as_explicitly_installed(self, pkgs: list[str]) -> list[str]: + """ + Running this command installs sets the given as explicitly installed. + """ + return ["pacman", "-D", "--asexplicit"] + pkgs + + def install_deps(self, deps: list[str]) -> list[str]: + """ + Running this command installs the given packages from pacman repositories. + The packages are installed as dependencies. + """ + return ["pacman", "-S", "--needed", "--asdeps"] + deps + + def is_installable(self, pkg: str) -> list[str]: + """ + This command exits with code 0 when a package is installable from pacman repositories. + """ + return ["pacman", "-Sddp", pkg] + + def upgrade(self) -> list[str]: + """ + Running this command upgrades all pacman packages. + """ + return ["pacman", "-Syu"] + + def remove(self, pkgs: list[str]) -> list[str]: + """ + Running this command removes the given packages and their dependencies + (that aren't required by other packages). + """ + return ["pacman", "-Rs"] + pkgs + + def enable_units(self, units: list[str]) -> list[str]: + """ + Running this command enables the given systemd units. + """ + return ["systemctl", "enable", "--now", "--quiet"] + units + + def disable_units(self, units: list[str]) -> list[str]: + """ + Running this command disables the given systemd units. + """ + return ["systemctl", "disable", "--quiet"] + units + + def enable_user_units(self, units: list[str]) -> list[str]: + """ + Running this command enables the given systemd units for the user it's run as. + """ + return ["systemctl", "enable", "--now", "--quiet", "--user"] + units + + def disable_user_units(self, units: list[str]) -> list[str]: + """ + Running this command disables the given systemd units fol the user it's run as. + """ + return ["systemctl", "disable", "--quiet"] + units + + def compare_versions(self, installed_version: str, + new_version: str) -> list[str]: + """ + Running this command outputs -1 when the installed version is older than the new version. + """ + return ["vercmp", installed_version, new_version] + + def git_clone(self, repo: str, dest: str) -> list[str]: + """ + Running this command clones a git repository to the the given destination. + """ + return ["git", "clone", repo, dest] + + def git_diff(self, from_commit: str) -> list[str]: + """ + Running this command outputs the difference between the given commit and + the current state of the repository. + """ + return ["git", "diff", from_commit] + + def git_get_commit_id(self) -> list[str]: + """ + Running this command outputs the current commit id. + """ + return ["git", "rev-parse", "HEAD"] + + def review_file(self, file: str) -> list[str]: + """ + Running this command outputs a file for the user to see. + """ + return ["less", file] + + def make_chroot(self, chroot_root_dir: str, + with_pkgs: list[str]) -> list[str]: + """ + Running this command creates a new arch chroot to the chroot directory and installs the + given packages there. + """ + return ["mkarchroot", chroot_root_dir] + with_pkgs + + def make_chroot_pkg(self, chroot_dir: str, user: str, + pkgfiles_to_install: list[str]) -> list[str]: + """ + Running this command creates a package file using the given chroot. + The package is created as the user and the pkg_files_to_install are installed + in the chroot before the package is created. + """ + makechrootpkg_cmd = ["makechrootpkg", "-r", chroot_dir, "-U", user] + + for pkgfile in pkgfiles_to_install: + makechrootpkg_cmd += ["-I", pkgfile] + + return makechrootpkg_cmd + + +commands: Commands = Commands() +debug_output: bool = False +quiet_output: bool = False + +valid_pkgexts: list[str] = [ + ".pkg.tar", + ".pkg.tar.gz", + ".pkg.tar.bz2", + ".pkg.tar.xz", + ".pkg.tar.zst", + ".pkg.tar.lzo", + ".pkg.tar.lrz", + ".pkg.tar.lz4", + ".pkg.tar.lz", + ".pkg.tar.Z", +] + +makepkg_user: str = "nobody" +build_dir: str = "/tmp/decman/build" +pkg_cache_dir: str = "/var/cache/decman" +aur_rpc_timeout: typing.Optional[int] = 30 diff --git a/src/decman/lib/__init__.py b/src/decman/lib/__init__.py new file mode 100644 index 0000000..a67e740 --- /dev/null +++ b/src/decman/lib/__init__.py @@ -0,0 +1,397 @@ +""" +Library module for decman. +""" + +import pwd +import subprocess +import json +import os +import typing +import decman.config as conf + +_DECMAN_MSG_TAG = "[\033[1;35mDECMAN\033[m]" +_RED_PREFIX = "\033[91m" +_YELLOW_PREFIX = "\033[93m" +_CYAN_PREFIX = "\033[96m" +_GREEN_PREFIX = "\033[92m" +_GRAY_PREFIX = "\033[90m" +_RESET_SUFFIX = "\033[m" + + +def print_error(error_msg: str): + """ + Prints an error message to the user. + """ + + print(f"{_DECMAN_MSG_TAG} {_RED_PREFIX}ERROR{_RESET_SUFFIX}: {error_msg}") + + +def print_warning(msg: str): + """ + Prints a warning to the user. + """ + + print(f"{_DECMAN_MSG_TAG} {_YELLOW_PREFIX}WARNING{_RESET_SUFFIX}: {msg}") + + +def print_summary(msg: str): + """ + Prints a summary message to the user. + """ + + print(f"{_DECMAN_MSG_TAG} {_CYAN_PREFIX}SUMMARY{_RESET_SUFFIX}: {msg}") + + +def print_info(msg: str): + """ + Prints a detailed message to the user if verbose output is not disabled. + """ + if conf.debug_output or not conf.quiet_output: + print(f"{_DECMAN_MSG_TAG} INFO: {msg}") + + +def print_debug(msg: str): + """ + Prints a detailed message to the user if debug messages are enabled. + """ + if conf.debug_output: + print(f"{_DECMAN_MSG_TAG} {_GRAY_PREFIX}DEBUG{_RESET_SUFFIX}: {msg}") + + +def prompt_number(msg: str, + min_num: int, + max_num: int, + default: typing.Optional[int] = None) -> int: + """ + Prompts the user for a integer. + """ + while True: + i = input( + f"{_DECMAN_MSG_TAG} {_GREEN_PREFIX}PROMPT{_RESET_SUFFIX}: {msg}" + ).strip() + + if default is not None and i == "": + return default + + try: + num = int(i) + if min_num <= num <= max_num: + return num + except ValueError: + pass + print_error("Invalid input.") + + +def prompt_confirm(msg: str, default: typing.Optional[bool] = None) -> int: + """ + Prompts the user for confirmation. + """ + + options_suffix = "(y/n)" + if default is not None: + if default: + options_suffix = "(Y/n)" + else: + options_suffix = "(y/N)" + + while True: + i = input( + f"{_DECMAN_MSG_TAG} {_GREEN_PREFIX}PROMPT{_RESET_SUFFIX} {options_suffix}: {msg} " + ).strip() + + if default is not None and i == "": + return default + + if i.lower() in ("y", "ye", "yes"): + return True + + if i.lower() in ("n", "no"): + return False + + print_error("Invalid input.") + + +_STORE_SAVE_DIR = "/var/lib/decman/" +_STORE_SAVE_FILENAME = "/var/lib/decman/store.json" + + +class Store: + """ + Stores information between decman invocations. + + This information is used for example to prevent re-enabling a service. + """ + + def __init__(self): + self.enabled_systemd_units: list[str] = [] + self.enabled_user_systemd_units: list[str] = [] + self.pkgbuild_latest_reviewed_commits: dict[str, str] = {} + self._package_file_cache: dict[str, tuple[str, str]] = {} + + def get_package(self, package: str) -> typing.Optional[tuple[str, str]]: + """ + Returns the version and the path of a package stored in the built packages cache as a tuple + (version, path). + """ + entry = self._package_file_cache.get(package) + if entry is None: + return None + version, path = entry + if os.path.exists(path): + return (version, path) + return None + + def add_package_to_cache(self, package: str, version: str, + path_to_built_pkg: str): + """ + Adds a built package to the package file cache. + """ + self._package_file_cache[package] = (version, path_to_built_pkg) + + def save(self): + """ + Writes the store to a file. + """ + + path = os.path.join(_STORE_SAVE_DIR, _STORE_SAVE_FILENAME) + + print_debug(f"Writing Store to '{path}'.") + + d = { + "enabled_systemd_units": self.enabled_systemd_units, + "enabled_user_systemd_units": self.enabled_user_systemd_units, + "package_file_cache": self._package_file_cache, + "pkgbuild_git_commits": self.pkgbuild_latest_reviewed_commits + } + + try: + os.makedirs(_STORE_SAVE_DIR, exist_ok=True) + 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 + + @staticmethod + def restore() -> "Store": + """ + Reads a saved Store from a file. + """ + path = os.path.join(_STORE_SAVE_DIR, _STORE_SAVE_FILENAME) + + print_debug(f"Reading Store from '{path}'.") + + try: + store = Store() + + with open(path, "rt", encoding="utf-8") as file: + d = json.load(file) + + store.enabled_systemd_units = d.get("enabled_systemd_units", + []) + store.enabled_user_systemd_units = d.get( + "enabled_user_systemd_units", []) + store._package_file_cache = d.get("package_file_cache", {}) + store.pkgbuild_latest_reviewed_commits = d.get( + "pkgbuild_git_commits", {}) + + return store + except json.JSONDecodeError as e: + raise 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 + + +class Pacman: + """ + Interface for interacting with pacman. + """ + + def get_installed(self) -> list[str]: + """ + Returns a list of installed packages. + """ + + try: + packages = subprocess.run( + conf.commands.list_pkgs(), + check=True, + stdout=subprocess.PIPE, + ).stdout.decode().split('\n') + return packages + except subprocess.CalledProcessError as error: + raise UserFacingError( + f"Failed to get installed packages using '{error.cmd}'. Output: {error.stdout}." + ) from error + + def is_installable(self, dep: str) -> bool: + """ + Returns True if a dependency can be installed using pacman. + """ + return subprocess.run(conf.commands.is_installable(dep), + check=False, + capture_output=True).returncode == 0 + + def get_versioned_foreign_packages(self) -> list[tuple[str, str]]: + """ + Returns a list of installed packages and their versions that aren't from pacman databases, + basically AUR packages. + """ + try: + output = subprocess.run( + conf.commands.list_foreign_pkgs_versioned(), + check=True, + stdout=subprocess.PIPE).stdout.decode().strip().split('\n') + except subprocess.CalledProcessError as error: + raise UserFacingError( + f"Failed to get foreign packages using '{error.cmd}'. Output: {error.stdout}." + ) from error + + try: + return [(line.split(" ")[0], line.split(" ")[1]) + for line in output] + except IndexError as error: + raise UserFacingError( + f"Failed to get foreign packages from pacman output. Output: {output}" + ) from error + + def install(self, packages: list[str]): + """ + Installs the given packages. + """ + try: + subprocess.run(conf.commands.install_pkgs(packages), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError("Failed to install packages.") from error + + def install_dependencies(self, deps: list[str]): + """ + Installs the given dependencies. + """ + try: + subprocess.run(conf.commands.install_deps(deps), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError( + "Failed to install dependency packages.") from error + + def install_files(self, files: list[str], as_explicit: list[str]): + """ + Installs the given files first as dependencies. Then the packages listed in as_explicit are + installed explicitly. + """ + try: + subprocess.run(conf.commands.install_files(files), check=True) + subprocess.run( + conf.commands.set_as_explicitly_installed(as_explicit), + check=True, + capture_output=True) + except subprocess.CalledProcessError as error: + raise UserFacingError( + "Failed to install foreign packages.") from error + + def upgrade(self): + """ + Upgrades all packages. + """ + try: + subprocess.run(conf.commands.upgrade(), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError("Failed to update packages.") from error + + def remove(self, packages: list[str]): + """ + Removes the given packages. + """ + try: + subprocess.run(conf.commands.remove(packages), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError("Failed to remove packages.") from error + + +class Systemd: + """ + Interface for interacting with systemd. + """ + + def __init__(self, state: Store): + self.state = state + + def enable_units(self, units: list[str]): + """ + Enables the given units. + """ + self.state.enabled_systemd_units += units + try: + subprocess.run(conf.commands.enable_units(units), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError("Failed to enable systemd units.") from error + + def disable_units(self, units: list[str]): + """ + Disables the given units. + """ + for unit in units: + try: + self.state.enabled_systemd_units.remove(unit) + except ValueError: + pass + try: + subprocess.run(conf.commands.disable_units(units), check=True) + except subprocess.CalledProcessError as error: + raise UserFacingError( + "Failed to disable systemd units.") from error + + def enable_user_units(self, units: list[str], user: str): + """ + Enables the given units for the given user. + """ + for unit in units: + self.state.enabled_user_systemd_units.append(f"{user}: {unit}") + try: + uid = pwd.getpwnam(user).pw_uid + 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( + f"Failed to enable systemd units for {user}.") + except KeyError as error: + raise UserFacingError( + f"Failed to enable systemd units because user '{user}' doesn't exist." + ) from error + + def disable_user_units(self, units: list[str], user: str): + """ + Disables the given units for the given user. + """ + for unit in units: + try: + self.state.enabled_user_systemd_units.remove(f"{user}: {unit}") + except ValueError: + pass + try: + uid = pwd.getpwnam(user).pw_uid + 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( + f"Failed to disable systemd units for {user}.") + except KeyError as error: + raise 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 new file mode 100644 index 0000000..5e4a7bf --- /dev/null +++ b/src/decman/lib/aur.py @@ -0,0 +1,836 @@ +""" +Module for interacting with the AUR. + +Optional dependencies are ignored when installing AUR packages. +Make and check dependencies are grouped together. + +Terminology: + +- package (pkg): name of an package from pacman repos or AUR +- dependency (dep): (virtual) package required when building and running a package +- dependency package (dep pkg): dependency that has been resolved to a package name +- build dependency: (virtual) package required when building a package (makedepends + checkdepends) +- build dependency package: build dependency that has been resolved to a package name +- all dependencies: normal dependencies and build dependencies combined +""" + +import shutil +import subprocess +import os +import re +import typing + +import requests + +import decman.config as conf +import decman.lib as l + + +def strip_dependency(dep: str) -> str: + """ + Removes version spefications from a dependency name. + """ + rx = re.compile("(=.*|>.*|<.*)") + return rx.sub("", dep) + + +class PackageInfo: + """ + Simplified information about an package. + + In case of AUR packages, these are fetched from AUR RPC. + """ + + def __init__(self, pkgname: str, pkgbase: str, version: str, + provides: list[str], dependencies: list[str], + make_and_check_dependencies: list[str], git_url: str): + self.pkgname = pkgname + self.pkgbase = pkgbase + self.version = version + self.dependencies = dependencies + self.build_dependencies = make_and_check_dependencies + self.provides = provides + self.git_url = git_url + self._aur_deps = None + self._pacman_deps = None + self._pacman_all_deps = None + + def pkg_file_prefix(self) -> str: + """ + Returns the beginning of the file created from building this package. + """ + return f"{self.pkgname}-{self.version}" + + def all_foreign_dependencies_stripped(self, pacman: l.Pacman) -> list[str]: + """ + Returs a list of dependencies that cannot be installed from pacman repos. + Includes build dependencies. + + Removes version spefications from package names. + """ + if self._aur_deps is not None: + return self._aur_deps + + result = [] + + for p in self.dependencies: + if not pacman.is_installable(p): + result.append(strip_dependency(p)) + + for p in self.build_dependencies: + if not pacman.is_installable(p): + result.append(strip_dependency(p)) + + self._aur_deps = result + return result + + def all_pacman_dependencies(self, pacman: l.Pacman) -> list[str]: + """ + Returs a list of dependencies that can be installed from pacman repos. + Includes build dependencies. + """ + if self._pacman_all_deps is not None: + return self._pacman_all_deps + + result = [] + + for p in self.dependencies: + if pacman.is_installable(p): + result.append(strip_dependency(p)) + + for p in self.build_dependencies: + if pacman.is_installable(p): + result.append(strip_dependency(p)) + + self._pacman_all_deps = result + return result + + def pacman_dependencies(self, pacman: l.Pacman) -> list[str]: + """ + Returs a list of dependencies that can be installed from pacman repos. + Doesn't include build dependencies. + """ + if self._pacman_deps is not None: + return self._pacman_deps + + result = [] + + for p in self.dependencies: + if pacman.is_installable(p): + result.append(strip_dependency(p)) + + self._pacman_deps = result + return result + + +class ForeignPackage: + """ + Class used to keep track of AUR/user dependency packages of an AUR/user package. + """ + + def __init__(self, name: str): + self.name = name + self._all_recursive_foreign_deps = set() + + def __eq__(self, value: object, /) -> bool: + if isinstance(value, self.__class__): + return self.name == value.name \ + and self._all_recursive_foreign_deps == value._all_recursive_foreign_deps + return False + + def __hash__(self) -> int: + return self.name.__hash__() + + def __repr__(self) -> str: + return f"{self.name}: [{' '.join(self._all_recursive_foreign_deps)}]" + + def add_foreign_dependency_packages(self, + package_names: typing.Iterable[str]): + """ + Adds dependencies to the package. + """ + self._all_recursive_foreign_deps.update(package_names) + + def get_all_recursive_foreign_deps(self) -> set[str]: + """ + Returns all dependencies and sub-dependencies of the package. + """ + return self._all_recursive_foreign_deps + + +class DepTreeNode: + """ + Foreign package and it's dependency packages. + """ + + def __init__(self, package: str, parent: typing.Optional[typing.Self]): + if parent is not None and package in parent.get_parent_package_names(): + raise l.UserFacingError( + f"Foreign package dependency cycle detected involving '{package}'. \ + foreign package dependencies are also required during package building \ + and therefore dependency cycles cannot be handled.") + + self._package = ForeignPackage(package) + self.parent = parent + self.children: dict[str, DepTreeNode] = {} + + def get_parent_package_names(self) -> list[str]: + """ + Returns package names of all parent nodes and self. + """ + if self.parent is not None: + return [self._package.name + ] + self.parent.get_parent_package_names() + return [self._package.name] + + def add_dependency_package(self, pkg: str, parents: list[str]): + """ + Adds a dependency package to this tree. + + parents is a list of names where the first element is the parent of the dependency, + the second element is the grandparent of the dependency and so on. + + Do not include this node in the parents list. + """ + if len(parents) == 0: + dep = DepTreeNode(pkg, self) + self.children[pkg] = dep + return + + last = parents.pop() + self.children[last].add_dependency_package(pkg, parents) + + def get_and_remove_outer_dep_pkgs(self) -> list[ForeignPackage]: + """ + Returns all leaf nodes of the dependency package tree and removes them. + """ + if len(self.children) == 0: + if self.parent is not None: + self.parent._package.add_foreign_dependency_packages( + [self._package.name]) + self.parent._package.add_foreign_dependency_packages( + self._package.get_all_recursive_foreign_deps()) + del self.parent.children[self._package.name] + return [self._package] + result = [] + for d in list(self.children.values()): + result.extend(d.get_and_remove_outer_dep_pkgs()) + return result + + +class ExtendedPackageSearch: + """ + Allows searcing for packages / providers from the AUR as well as user defined sources. + + Results are cached and user defined packages are preferred. + """ + + def __init__(self): + self._package_info_cache: dict[str, PackageInfo] = {} + self._dep_provider_cache: dict[str, PackageInfo] = {} + self._user_packages: list[PackageInfo] = [] + + def add_user_pkg(self, user_pkg: PackageInfo): + """ + Adds the given package to user packages. + """ + self._user_packages.append(user_pkg) + + def get_package_info(self, package: str) -> typing.Optional[PackageInfo]: + """ + Returns information about a package. + + If the package is not user defined, fetches information from the AUR. + Returns None if no such AUR package exists. + """ + if package in self._package_info_cache: + return self._package_info_cache[package] + + l.print_debug(f"Getting info for package '{package}'.") + + for user_package in self._user_packages: + if user_package.pkgname == package: + l.print_debug(f"'{package}' found in user packages.") + self._package_info_cache[package] = user_package + return user_package + + url = f"https://aur.archlinux.org/rpc/v5/info/{package}" + l.print_debug(f"Requesting info for '{package}' from AUR. URL = {url}") + try: + request = requests.get(url, timeout=conf.aur_rpc_timeout) + d = request.json() + + if d["type"] == "error": + raise l.UserFacingError( + f"AUR RPC returned error: {d['error']}") + + if d["resultcount"] == 0: + l.print_debug(f"'{package}' not found.") + return None + + l.print_debug(f"'{package}' found from AUR.") + + result = d["results"][0] + info = PackageInfo( + pkgname=result["Name"], + pkgbase=result["PackageBase"], + version=result["Version"], + dependencies=result.get("Depends", []), + make_and_check_dependencies=result.get("MakeDepends", []) + + result.get("CheckDepends", []), + provides=result.get("Provides", []), + git_url=f"https://aur.archlinux.org/{result['PackageBase']}.git" + ) + + self._package_info_cache[package] = info + + return info + except (requests.RequestException, KeyError) as e: + raise l.UserFacingError( + "Failed to fetch package information from AUR RPC.") from e + + def find_provider( + self, stripped_dependency: str) -> typing.Optional[PackageInfo]: + """ + Finds a provider for a dependency. + + May prompt the user to select if multiple are available. + """ + if stripped_dependency in self._dep_provider_cache: + return self._dep_provider_cache[stripped_dependency] + + l.print_debug(f"Finding provider for '{stripped_dependency}'.") + l.print_debug("Are there exact name matches?") + + exact_name_match = self.get_package_info(stripped_dependency) + + if exact_name_match is not None: + self._dep_provider_cache[stripped_dependency] = exact_name_match + return exact_name_match + + l.print_debug("No exact name matches found. Finding providers.") + + user_pkg_results = [] + for user_package in self._user_packages: + if stripped_dependency in user_package.provides: + user_pkg_results.append(user_package.pkgname) + + if len(user_pkg_results) == 1: + pkg = self.get_package_info(user_pkg_results[0]) + assert pkg is not None + l.print_debug( + f"Single provider for '{stripped_dependency}' found in user packages: '{pkg}'." + ) + self._dep_provider_cache[stripped_dependency] = pkg + return pkg + + if len(user_pkg_results) > 1: + return self._choose_provider(stripped_dependency, user_pkg_results, + "user packages") + + url = f"https://aur.archlinux.org/rpc/v5/search/{stripped_dependency}?by=provides" + l.print_debug( + f"Requesting providers for '{stripped_dependency}' from AUR. URL = {url}" + ) + try: + request = requests.get(url, timeout=conf.aur_rpc_timeout) + d = request.json() + + if d["type"] == "error": + raise l.UserFacingError( + f"AUR RPC returned error: {d['error']}") + + if d["resultcount"] == 0: + l.print_debug(f"'{stripped_dependency}' not found.") + return None + + results = list(map(lambda r: r["Name"], d["results"])) + + if len(results) == 1: + pkgname = results[0] + l.print_debug( + f"Single provider for '{stripped_dependency}' found from AUR: '{pkgname}'" + ) + info = self.get_package_info(pkgname) + return info + + return self._choose_provider(stripped_dependency, results, "AUR") + except (requests.RequestException, KeyError) as e: + raise l.UserFacingError( + "Failed to fetch package information from AUR RPC.") from e + + def _choose_provider(self, dep: str, possible_providers: list[str], + where: str) -> typing.Optional[PackageInfo]: + min_selection = 1 + max_selection = len(possible_providers) + l.print_summary( + f"Found {len(possible_providers)} providers for {dep} from {where}." + ) + + providers = "Providers: " + for index, name in enumerate(possible_providers): + providers += f"{index + 1}:{name} " + l.print_info(providers) + + selection = l.prompt_number( + f"Select a provider [{min_selection}-{max_selection}] (default: {min_selection}): ", + min_selection, + max_selection, + default=min_selection) + + info = self.get_package_info(possible_providers[selection - 1]) + if info is not None: + self._dep_provider_cache[dep] = info + return info + + +class ForeignPackageManager: + """ + Class for dealing with AUR/user packages. + """ + + def __init__(self, store: l.Store, pacman: l.Pacman, + search: ExtendedPackageSearch): + self._store = store + self._pacman = pacman + self._search = search + + def upgrade(self, upgrade_devel: bool = False): + """ + Upgrades all AUR/user packages. + """ + all_foreign_pkgs = self._pacman.get_versioned_foreign_packages() + all_explicit_pkgs = set(self._pacman.get_installed()) + l.print_debug( + f"Foreign packages to check for upgrades: {all_foreign_pkgs}") + + to_upgrade = [] + as_explicit = [] + for pkg, ver in all_foreign_pkgs: + info = self._search.get_package_info(pkg) + if info is None: + raise l.UserFacingError(f"Failed to find package: {pkg}.") + + if self.should_upgrade_package(pkg, ver, info.version, + upgrade_devel): + to_upgrade.append(pkg) + if pkg in all_explicit_pkgs: + as_explicit.append(pkg) + + l.print_debug( + f"The following foreign packages will be upgraded: {' '.join(to_upgrade)}" + ) + + self.install(to_upgrade, as_explicit, True) + + def install(self, + foreign_pkgs: list[str], + as_explicit: typing.Optional[list[str]] = None, + force: bool = False): + """ + Installs the given AUR/user packages and their dependencies (both pacman/AUR). + """ + + if as_explicit is None: + as_explicit = foreign_pkgs + + all_foreign_pkgs, pacman_deps = self.resolve_dependencies(foreign_pkgs) + + l.print_summary( + f"The following foreign packages will be installed: {' '.join(map(lambda p: p.name, all_foreign_pkgs))}" + ) + + if not l.prompt_confirm("Proceed?", default=True): + raise l.UserFacingError("Installing aborted.") + + l.print_summary( + "Installing AUR/user package dependencies from pacman.") + self._pacman.install_dependencies(list(pacman_deps)) + + to_install = [] + + while all_foreign_pkgs: + pkg_to_build = all_foreign_pkgs.pop(0) + + # resolve_dependencies gets info for every package so this cannot be None + pkgbase = self._search.get_package_info( + pkg_to_build.name + ).pkgbase # pyright: ignore[reportOptionalMemberAccess] + with_same_pkgbase = [] + + for other in all_foreign_pkgs: + other_pkgbase = self._search.get_package_info( + other.name + ).pkgbase # pyright: ignore[reportOptionalMemberAccess] + if other_pkgbase == pkgbase: + with_same_pkgbase.append(other) + + for other in with_same_pkgbase: + all_foreign_pkgs.remove(other) + + to_install += self._build_pkg(pkgbase, + [pkg_to_build] + with_same_pkgbase, + force) + + if to_install or force: + l.print_summary("Installing AUR/user packages.") + self._pacman.install_files(to_install, as_explicit) + else: + l.print_summary("No packages to install.") + + def resolve_dependencies( + self, foreign_packages: list[str] + ) -> tuple[list[ForeignPackage], set[str]]: + """ + Resolves AUR/user dependencies of AUR/user packages. + + Returns a tuple of (foreign_packages, pacman_deps) + + foreign_packages are in the order they should be built + (the 1st element should be built 1st) + + pacman_deps are dependencies that are required by the AUR/user packages. + """ + + l.print_summary("Resolving AUR / user package dependencies.") + l.print_debug(f"Packages: {foreign_packages}") + + # AUR packages are stored in this 2D list in the following format. + # + # Assume that aur_packages contains package A that should be explicitly installed. + # A depends on B1 and B2 + # B1 depends on C + # + # Then the list will contain the following elements: + # - [A] + # - [B1,A] + # - [B2,A] + # - [C,B1,A] + # + # The list will be processed at the same time it's appended to so an actuality + # all the elements wont be in the list at the same time + packages_with_dependants = [[pkg] for pkg in foreign_packages] + + # Used to solve the build order of packages. + dep_tree_root = DepTreeNode("*", None) + pacman_deps = set() + + while packages_with_dependants: + l.print_info( + f"Packages remaining: {len(packages_with_dependants)}.") + + package_and_parents = packages_with_dependants.pop() + + package = package_and_parents[0] + parents = package_and_parents[1:] + + l.print_debug( + f"Adding package '{package}' with parents {parents} to the dependency tree." + ) + dep_tree_root.add_dependency_package(package, parents) + + info = self._search.get_package_info(package) + if info is None: + raise l.UserFacingError( + f"Failed to find '{package}' from AUR or user provided packages." + ) + + pacman_deps.update(info.pacman_dependencies(self._pacman)) + + for foreign_dep in info.all_foreign_dependencies_stripped( + self._pacman): + pkg = self._search.find_provider(foreign_dep) + if pkg is None: + raise l.UserFacingError( + f"Failed to find '{pkg}' from AUR or user provided packages." + ) + + l.print_info("Found new package to process.") + packages_with_dependants.append([pkg.pkgname] + + package_and_parents) + + l.print_summary("Determining build order.") + + build_order = [] + while True: + to_add = dep_tree_root.get_and_remove_outer_dep_pkgs() + + if len(to_add) == 1 and to_add[0].name == "*": + break + + l.print_debug(f"Adding {to_add} to build_order.") + + for pkg in to_add: + if pkg not in build_order: + build_order.append(pkg) + + l.print_debug(f"Build order is {build_order}") + + return (build_order, pacman_deps) + + def _build_pkg(self, package_base: str, packages: list[ForeignPackage], + force: bool) -> list[str]: + """ + Builds package(s) with the same package base. Returns a list of package files to install. + """ + + package_names = list(map(lambda p: p.name, packages)) + + # Rebuild is only needed if at least one package is not in the cache. + + if self._are_all_pkgs_cached(packages) and not force: + l.print_summary( + f"Skipped building '{' '.join(package_names)}'. Already up to date." + ) + return [] + + l.print_summary(f"To build '{' '.join(package_names)}'.") + + chroot_pacman_pkgs, chroot_pkg_files = self._get_chroot_packages( + packages) + + chroot_dir = os.path.join(conf.build_dir, "chroot") + pkgbuild_dir = os.path.join(conf.build_dir, "pkgbuild") + + l.print_debug( + f"Chroot dir is: '{chroot_dir}', pkgbuild dir is '{pkgbuild_dir}'." + ) + + prev_wd = os.getcwd() + + try: + os.makedirs(conf.pkg_cache_dir, exist_ok=True) + + if os.path.exists(conf.build_dir): + l.print_info("Removing previous build directory.") + shutil.rmtree(conf.build_dir) + + l.print_info("Setting up build directory.") + os.makedirs(pkgbuild_dir) + os.makedirs(chroot_dir) + + os.chdir(pkgbuild_dir) + + git_url = self._search.get_package_info( + package_names[0] + ).git_url # pyright: ignore[reportOptionalMemberAccess] + + l.print_debug(f"Git URL for '{package_base}' is '{git_url}'") + + self.git_clone_and_review_pkgbuild(package_base, git_url) + shutil.chown(pkgbuild_dir, user=conf.makepkg_user) + + l.print_summary(f"Building: '{' '.join(package_names)}'.") + + # Remove GNUPGHOME from mkarchroot environment variables since it may interfere with + # the chroot creation + mkarchroot_env_vars = os.environ.copy() + try: + del mkarchroot_env_vars["GNUPGHOME"] + l.print_debug( + "Removed GNUPGHOME variable from mkarchroot environment.") + except KeyError: + pass + + l.print_info("Creating a new chroot.") + + subprocess.run(conf.commands.make_chroot( + os.path.join(chroot_dir, "root"), + ["base-devel"] + chroot_pacman_pkgs), + env=mkarchroot_env_vars, + check=True, + capture_output=conf.quiet_output) + + l.print_info("Making package.") + + subprocess.run(conf.commands.make_chroot_pkg( + chroot_dir, conf.makepkg_user, chroot_pkg_files), + check=True, + capture_output=conf.quiet_output) + + package_files = [] + + for pkgname in package_names: + file = self._find_pkgfile(pkgname, pkgbuild_dir) + + dest = shutil.copy(file, conf.pkg_cache_dir) + + version = self._search.get_package_info( + pkgname + ).version # pyright: ignore[reportOptionalMemberAccess] + + l.print_debug( + f"Adding '{pkgname}', version: '{version}' to cache as file '{dest}'." + ) + + self._store.add_package_to_cache(pkgname, version, dest) + package_files.append(dest) + + l.print_summary(f"Finished building: '{' '.join(package_names)}'.") + except (subprocess.CalledProcessError, OSError) as error: + raise l.UserFacingError( + f"Failed to build package(s) '{' '.join(map(lambda p: p.name, packages))}'." + ) from error + finally: + os.chdir(prev_wd) + + return package_files + + def _are_all_pkgs_cached(self, pkgs: list[ForeignPackage]) -> bool: + for pkg in pkgs: + cache_entry = self._store.get_package(pkg.name) + if cache_entry is None: + return False + cached_version, _ = cache_entry + # resolve_dependencies gets info for every package so info cannot be None + fetched_version = self._search.get_package_info( + pkg.name + ).version # pyright: ignore[reportOptionalMemberAccess] + + if cached_version != fetched_version or self.is_devel(pkg.name): + return False + return True + + def _get_chroot_packages( + self, pkgs_to_build: list[ForeignPackage] + ) -> tuple[list[str], list[str]]: + """ + Returns a tuple of pacman packages and built foreign pkgs files that are needed in the + chroot before building. pkgs_to_build share the same pkgbase. + """ + chroot_pacman_pkgs = set() + chroot_foreign_pkgs = set() + + for pkg in pkgs_to_build: + info = self._search.get_package_info(pkg.name) + assert info is not None + + chroot_pacman_pkgs.update( + info.all_pacman_dependencies(self._pacman)) + + foreign_deps = pkg.get_all_recursive_foreign_deps() + chroot_foreign_pkgs.update(foreign_deps) + + # Add pacman deps of foreign packages + for dep in foreign_deps: + dep_info = self._search.get_package_info(dep) + assert dep_info is not None + chroot_pacman_pkgs.update( + dep_info.all_pacman_dependencies(self._pacman)) + + # Packages with the same pkgbase might depend on each other, + # but they don't need to be installed for the build to succeed. + for pkg in pkgs_to_build: + if pkg.name in chroot_foreign_pkgs: + chroot_foreign_pkgs.remove(pkg.name) + + chroot_foreign_pkg_files = [] + + for foreign_pkg in chroot_foreign_pkgs: + entry = self._store.get_package(foreign_pkg) + assert entry is not None, "Build order determines that the dependencies are built \ + before and thus are found in the cache." + + _, file = entry + + chroot_foreign_pkg_files.append(file) + + return (list(chroot_pacman_pkgs), chroot_foreign_pkg_files) + + def _find_pkgfile(self, pkgname: str, pkgbuild_dir: str) -> str: + # HACK: Because we don't know the pkgarch we can't be sure what is the build result. + # Instead: we just try with pre- and postfixes. + + matches = [] + + info = self._search.get_package_info(pkgname) + assert info is not None + prefix = info.pkg_file_prefix() + + for file in os.scandir(pkgbuild_dir): + if file.is_file() and file.name.startswith(prefix): + for ext in conf.valid_pkgexts: + if file.name.endswith(ext): + matches.append(file.path) + continue + + if len(matches) != 1: + raise l.UserFacingError( + f"Failed to build package '{pkgname}', because the pkg file cannot be determined." + ) + + return matches[0] + + def git_clone_and_review_pkgbuild(self, pkgbase: str, git_url: str): + """ + Clones an PKGBUILD to the current directory. + + The user is prompted to review the PKGBUILD and confirm if the package should be built. + """ + try: + subprocess.run(conf.commands.git_clone(git_url, "."), check=True) + + latest_reviewed_commit = self._store.pkgbuild_latest_reviewed_commits.get( + pkgbase) + if latest_reviewed_commit is None: + for file in os.scandir("."): + if file.is_file() and not file.name.startswith("."): + subprocess.run(conf.commands.review_file(file.path), + check=True) + else: + subprocess.run(conf.commands.git_diff(latest_reviewed_commit), + check=True) + + if l.prompt_confirm("Proceed with building?", default=True): + commit_id = subprocess.run( + conf.commands.git_get_commit_id(), + check=True, + capture_output=True).stdout.decode().strip() + self._store.pkgbuild_latest_reviewed_commits[ + pkgbase] = commit_id + else: + raise l.UserFacingError("Building aborted.") + + except subprocess.CalledProcessError as error: + raise l.UserFacingError( + f"Failed to clone and review PKGBUILD from {git_url}" + ) from error + + def should_upgrade_package(self, + package: str, + installed_version: str, + fetched_version: str, + upgrade_devel=False) -> bool: + """ + Returns True if a package should be upgraded. + """ + + if upgrade_devel and self.is_devel(package): + return True + + try: + result = int( + subprocess.run(conf.commands.compare_versions( + installed_version, fetched_version), + check=True, + stdout=subprocess.PIPE).stdout.decode()) + return result < 0 + except (ValueError, subprocess.CalledProcessError) as error: + raise l.UserFacingError("Failed to compare versions.") from error + + def is_devel(self, package: str) -> bool: + """ + Returns True if the given package is a devel package. + """ + devel_suffixes = [ + "-git", + "-hg", + "-bzr", + "-svn", + "-cvs", + "-darcs", + ] + for suffix in devel_suffixes: + if package.endswith(suffix): + return True + return False diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..7e068a7 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,6 @@ +import os +import sys + +_SRC_PATH = os.path.join(os.path.dirname(__file__), "../src/") + +sys.path.append(_SRC_PATH) diff --git a/tests/test_package_management.py b/tests/test_package_management.py new file mode 100644 index 0000000..2347a5f --- /dev/null +++ b/tests/test_package_management.py @@ -0,0 +1,80 @@ +# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring + +import unittest +from decman.lib import UserFacingError, Pacman, Store +from decman.lib.aur import ForeignPackageManager, DepTreeNode, ForeignPackage, ExtendedPackageSearch + + +class TestAUR(unittest.TestCase): + + def setUp(self) -> None: + self.aur = ForeignPackageManager(Store(), Pacman(), + ExtendedPackageSearch()) + + def test_should_upgrade_package_returns_true_on_newer_version(self): + self.assertTrue( + self.aur.should_upgrade_package("test", "0.1.9", "0.2.0")) + + def test_should_upgrade_package_returns_false_on_older_version(self): + self.assertFalse( + self.aur.should_upgrade_package("test", "0.1.9", "0.1.8")) + + def test_should_upgrade_package_returns_false_on_same_version(self): + self.assertFalse( + self.aur.should_upgrade_package("test", "0.1.9", "0.1.9")) + + def test_should_upgrade_package_returns_true_on_devel(self): + self.assertTrue( + self.aur.should_upgrade_package("test-git", + "0", + "0", + upgrade_devel=True)) + + +class TestDepTree(unittest.TestCase): + + def test_add_dependency(self): + root = DepTreeNode("", None) + + root.add_dependency_package("A", []) + root.add_dependency_package("B", ["A"]) + root.add_dependency_package("B1", ["A"]) + root.add_dependency_package("C", ["B", "A"]) + + self.assertIn("A", root.children) + self.assertIn("B", root.children["A"].children) + self.assertIn("B1", root.children["A"].children) + self.assertIn("C", root.children["A"].children["B"].children) + + def test_cyclic_dep_fails(self): + root = DepTreeNode("", None) + + root.add_dependency_package("A", []) + root.add_dependency_package("B", ["A"]) + + with self.assertRaises(UserFacingError): + root.add_dependency_package("A", ["B", "A"]) + + def test_get_and_remove_outer_deps(self): + root = DepTreeNode("", None) + + root.add_dependency_package("A", []) + root.add_dependency_package("B", ["A"]) + root.add_dependency_package("B1", ["A"]) + root.add_dependency_package("C", ["B", "A"]) + + a = ForeignPackage("A") + a.add_foreign_dependency_packages(["B", "B1", "C"]) + b = ForeignPackage("B") + b.add_foreign_dependency_packages(["C"]) + b1 = ForeignPackage("B1") + c = ForeignPackage("C") + + self.assertCountEqual(root.get_and_remove_outer_dep_pkgs(), [c, b1]) + self.assertCountEqual(root.get_and_remove_outer_dep_pkgs(), [b]) + self.assertCountEqual(root.get_and_remove_outer_dep_pkgs(), [a]) + + last = root.get_and_remove_outer_dep_pkgs() + + self.assertEqual(len(last), 1) + self.assertEqual(last.pop().name, "")