diff --git a/src/decman/config.py b/src/decman/config.py index 30a8eb4..ca5afa4 100644 --- a/src/decman/config.py +++ b/src/decman/config.py @@ -24,5 +24,4 @@ debug_output: bool = False quiet_output: bool = False color_output: bool = True -pkg_cache_dir: str = "/var/cache/decman" module_on_disable_scripts_dir: str = "/var/lib/decman/scripts/" diff --git a/src/decman/core/command.py b/src/decman/core/command.py index bbcc708..4f26307 100644 --- a/src/decman/core/command.py +++ b/src/decman/core/command.py @@ -32,12 +32,16 @@ def pty_run( user: None | str = None, env_overrides: None | dict[str, str] = None, mimic_login: bool = False, + pass_environment: bool = True, ) -> tuple[int, str]: """ Runs a given command with the given arguments in a pseudo TTY. The command can be ran as the given user and environment variables can be overridden manually. - If mimic_login is True, will set the following environment variables according to the given + By default this will copy the current environment and pass it to the process. To prevent this + set ``pass_environment`` to ``False``. + + If ``mimic_login`` is True, will set the following environment variables according to the given user's passwd file details. This only happens when user is set. - HOME - USER @@ -57,9 +61,11 @@ def pty_run( if not sys.stdin.isatty(): raise OSError(errno.ENOTTY, "Stdin is not a TTY.") + command[0] = shutil.which(command[0]) or command[0] + output.print_debug(f"Running command '{shlex.join(command)}'") - env = _build_env(user=user, env_overrides=env_overrides, mimic_login=mimic_login) + env = _build_env(user, env_overrides, mimic_login, pass_environment) pid, master_fd = pty.fork() if pid == 0: @@ -73,11 +79,15 @@ def run( user: None | str = None, env_overrides: None | dict[str, str] = None, mimic_login: bool = False, + pass_environment: bool = False, ) -> tuple[int, str]: """ Runs a given command with the given arguments. The command can be ran as the given user and environment variables can be overridden manually. + By default this will copy the current environment and pass it to the process. To prevent this + set ``pass_environment`` to ``False``. + If mimic_login is True, will set the following environment variables according to the given user's passwd file details. This only happens when user is set. - HOME @@ -94,9 +104,11 @@ def run( if not command: return 0, "" + command[0] = shutil.which(command[0]) or command[0] + output.print_debug(f"Running command '{shlex.join(command)}'") - env = _build_env(user=user, env_overrides=env_overrides, mimic_login=mimic_login) + env = _build_env(user, env_overrides, mimic_login, pass_environment) uid, gid = None, None if user: @@ -136,8 +148,12 @@ def _build_env( user: None | str, env_overrides: None | dict[str, str], mimic_login: bool, + pass_environment: bool, ) -> dict[str, str]: - env = os.environ.copy() + env = {} + + if pass_environment: + env = os.environ.copy() if mimic_login and user: pw = _get_passwd(user) @@ -184,6 +200,7 @@ def _run_parent(master_fd: int, pid: int) -> tuple[int, str]: # Set PTY window size to match the current terminal size. # We accept that resizing the real terminal causes issues here, it doesn't need to be handeled + # TODO: No actually, it's better to do resizing rows, columns = shutil.get_terminal_size() winsz = struct.pack("HHHH", rows, columns, 0, 0) fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsz) diff --git a/src/decman/plugins/pacman.py b/src/decman/plugins/pacman.py deleted file mode 100644 index 941f500..0000000 --- a/src/decman/plugins/pacman.py +++ /dev/null @@ -1,5 +0,0 @@ -import decman.plugins as plugins - - -class Pacman(plugins.Plugin): - NAME = "pacman" diff --git a/src/decman/plugins/pacman/__init__.py b/src/decman/plugins/pacman/__init__.py new file mode 100644 index 0000000..0d30eb8 --- /dev/null +++ b/src/decman/plugins/pacman/__init__.py @@ -0,0 +1,74 @@ +import shutil + +import decman.core.module as module +import decman.core.store as _store +import decman.plugins as plugins + +# Re-exports +from decman.plugins.pacman.commands import PacmanCommands +from decman.plugins.pacman.package import CustomPackage + +__all__ = [ + "PacmanCommands", + "CustomPackage", + "packages", + "aur_packages", + "custom_packages", + "Pacman", +] + + +def packages(fn): + """ + Annotate that this function returns a set of pacman package names that should be installed. + + Return type of ``fn``: ``set[str]`` + """ + fn.__pacman__packages__ = True + return fn + + +def aur_packages(fn): + """ + Annotate that this function returns a set of AUR package names that should be installed. + + Return type of ``fn``: ``set[str]`` + """ + fn.__aur__packages__ = True + return fn + + +def custom_packages(fn): + """ + Annotate that this function returns a set of ``CustomPackage``s that should be installed. + + Return type of ``fn``: ``set[CustomPackage]`` + """ + fn.__custom__packages__ = True + return fn + + +class Pacman(plugins.Plugin): + """ + Plugin that manages pacman packages added directly to ``packages`` or declared by modules via + @packages. + """ + + NAME = "pacman" + + def __init__(self) -> None: + self.packages: set[str] = set() + self.aur_packages: set[str] = set() + self.commands = PacmanCommands() + + def available(self) -> bool: + return shutil.which("pacman") is not None + + def process_modules(self, store: _store.Store, modules: set[module.Module]): + # This is used to track changes in modules. + store.ensure("packages_for_module", {}) + + def apply( + self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None + ) -> bool: + return True diff --git a/src/decman/plugins/pacman/commands.py b/src/decman/plugins/pacman/commands.py new file mode 100644 index 0000000..d09d2cf --- /dev/null +++ b/src/decman/plugins/pacman/commands.py @@ -0,0 +1,344 @@ +import decman.core.command as command +import decman.core.error as errors +import decman.core.output as output + + +class PacmanCommands: + def list_explicit(self) -> list[str]: + """ + Running this command outputs a newline seperated list of explicitly installed native + packages. + """ + return ["pacman", "-Qeq", "--color=never"] + + def list_orphans(self) -> list[str]: + """ + Running this command outputs a newline seperated list of orphaned packages. + """ + return ["pacman", "-Qdtq", "--color=never"] + + def list_dependants(self, pkg: str) -> list[str]: + """ + Running this command outputs a newline seperated list of packages that depend on the given + package. + """ + return ["pacman", "-Rc", "--print", "--print-format", "%n", pkg] + + def list_foreign_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 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 install(self, pkgs: set[str]) -> list[str]: + """ + Running this command installs the given packages from pacman repositories. + """ + return ["pacman", "-S", "--needed"] + list(pkgs) + + def install_as_dependencies(self, pkgs: set[str]) -> list[str]: + """ + Running this command installs the given packages from pacman repositories. + The packages are installed as dependencies. + """ + return ["pacman", "-S", "--needed", "--asdeps"] + list(pkgs) + + def install_files_as_dependencies(self, pkg_files: list[str]) -> list[str]: + """ + Running this command installs the given packages files as dependencies. + """ + return ["pacman", "-U", "--asdeps"] + pkg_files + + def upgrade(self) -> list[str]: + """ + Running this command upgrades all pacman packages. + """ + return ["pacman", "-Syu"] + + def set_as_dependencies(self, pkgs: set[str]) -> list[str]: + """ + Running this command installs sets the given packages as dependencies. + """ + return ["pacman", "-D", "--asdeps"] + list(pkgs) + + def set_as_explicit(self, pkgs: set[str]) -> list[str]: + """ + Running this command installs sets the given as explicitly installed. + """ + return ["pacman", "-D", "--asexplicit"] + list(pkgs) + + def remove(self, pkgs: set[str]) -> list[str]: + """ + Running this command removes the given packages and their dependencies + (that aren't required by other packages). + """ + return ["pacman", "-Rs"] + list(pkgs) + + 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 git_log_commit_ids(self) -> list[str]: + """ + Running this command outputs commit hashes of the repository. + """ + return ["git", "log", "--format=format:%H"] + + 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_dir: str, with_pkgs: set[str]) -> list[str]: + """ + Running this command creates a new arch chroot to the chroot directory and installs the + given packages there. + """ + return ["mkarchroot", chroot_dir] + list(with_pkgs) + + def install_chroot(self, chroot_dir: str, packages: list[str]): + """ + Running this command installs the given packages to the given chroot. + """ + return [ + "arch-nspawn", + chroot_dir, + "pacman", + "-S", + "--needed", + "--noconfirm", + ] + packages + + def resolve_real_name_chroot(self, chroot_dir: str, pkg: str) -> list[str]: + """ + This command prints a real name of a package. + For example, it prints the package which provides a virtual package. + """ + return [ + "arch-nspawn", + chroot_dir, + "pacman", + "-Sddp", + "--print-format=%n", + pkg, + ] + + def remove_chroot(self, chroot_dir: str, packages: set[str]): + """ + Running this command removes the given packages from the given chroot. + """ + return ["arch-nspawn", chroot_dir, "pacman", "-Rsu", "--noconfirm"] + list(packages) + + def make_chroot_pkg( + self, chroot_wd_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", "-c", "-r", chroot_wd_dir, "-U", user] + + for pkgfile in pkgfiles_to_install: + makechrootpkg_cmd += ["-I", pkgfile] + + return makechrootpkg_cmd + + +class PacmanInterface: + """ + High level interface for running pacman commands. + + On failure methods raise a ``CommandFailedError``. + """ + + def __init__( + self, commands: PacmanCommands, print_highlights: bool, keywords: set[str] + ) -> None: + self._installable: dict[str, bool] = {} + self._commands = commands + self._print_highlights = print_highlights + self._keywords = keywords + + def get_installed(self) -> list[str]: + """ + Returns a list of installed packages. + """ + + returncode, packages_text = command.run(self._commands.list_explicit()) + packages = packages_text.strip().split("\n") + + if returncode != 0: + raise errors.CommandFailedError(self._commands.list_explicit(), packages_text) + + return packages + + def is_installable(self, pkg: str) -> bool: + """ + Returns True if a package can be installed using pacman. + """ + if pkg in self._installable: + return self._installable[pkg] + + returncode, _ = command.run(self._commands.is_installable(pkg)) + result = returncode == 0 + + self._installable[pkg] = result + return result + + 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. + """ + cmd = self._commands.list_foreign_versioned() + returncode, packages_text = command.run(cmd) + packages = [ + (line.split(" ")[0], line.split(" ")[1]) for line in packages_text.strip().split("\n") + ] + + if returncode != 0: + raise errors.CommandFailedError(cmd, packages_text) + + return packages + + def install(self, packages: set[str]): + """ + Installs the given packages. If the packages are already installed, marks them as + explicitly installed. + """ + if not packages: + return + + cmd = self._commands.install(packages) + + returncode, pacman_output = command.pty_run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + self.print_highlighted_pacman_messages(pacman_output) + + cmd = self._commands.set_as_explicit(packages) + + returncode, pacman_output = command.run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + def install_dependencies(self, deps: set[str]): + """ + Installs the given dependencies. + """ + if not deps: + return + + cmd = self._commands.install_as_dependencies(deps) + + returncode, pacman_output = command.pty_run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + self.print_highlighted_pacman_messages(pacman_output) + + def install_files(self, files: list[str], as_explicit: set[str]): + """ + Installs the given files first as dependencies. Then the packages listed in as_explicit are + installed explicitly. + """ + if not files: + return + + cmd = self._commands.install_files_as_dependencies(files) + + returncode, pacman_output = command.pty_run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + self.print_highlighted_pacman_messages(pacman_output) + + if not as_explicit: + return + + cmd = self._commands.set_as_explicit(as_explicit) + + returncode, pacman_output = command.run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + def upgrade(self): + """ + Upgrades all packages. + """ + cmd = self._commands.upgrade() + + returncode, pacman_output = command.pty_run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + self.print_highlighted_pacman_messages(pacman_output) + + def remove(self, packages: set[str]): + """ + Removes the given packages. + """ + if not packages: + return + + cmd = self._commands.remove(packages) + + returncode, pacman_output = command.pty_run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, pacman_output) + + self.print_highlighted_pacman_messages(pacman_output) + + def print_highlighted_pacman_messages(self, pacman_output: str): + """ + Prints lines that contain pacman output keywords. + """ + if not self._print_highlights: + return + + output.print_summary("Pacman output highlights:") + lines = pacman_output.split("\n") + for index, line in enumerate(lines): + for keyword in self._keywords: + if keyword.lower() in line.lower(): + output.print_summary(f"lines: {index}-{index + 2}") + if index >= 1: + output.print_continuation(lines[index - 1]) + output.print_continuation(line) + if index + 1 < len(lines): + output.print_continuation(lines[index + 1]) + output.print_continuation("") + + # Break, as to not print the same line again if it contains multiple keywords + break diff --git a/src/decman/plugins/pacman/error.py b/src/decman/plugins/pacman/error.py new file mode 100644 index 0000000..38d9cc6 --- /dev/null +++ b/src/decman/plugins/pacman/error.py @@ -0,0 +1,40 @@ +class ForeignPackageManagerError(Exception): + """ + Error raised from the ForeignPackageManager + """ + + +class DependencyCycleError(Exception): + """ + Error raised when a dependency cycle is detected involving foreign packages. + """ + + def __init__(self, package1: str, package2: str): + super().__init__( + f"Foreign package dependency cycle detected involving '{package1}' \ + and '{package2}'. Foreign package dependencies are also required \ + during package building and therefore dependency cycles cannot be handled." + ) + + +class PKGBUILDParseError(Exception): + """ + Error raised when parsing a PKGBUILD fails. + """ + + def __init__(self, git_url: str | None, pkgbuild_directory: str | None, message: str) -> None: + # Only one of these should be set + self.pkgbuild_source = git_url or pkgbuild_directory + self.message = message + super().__init__(f"Failed to parse PKGBUILD from '{self.pkgbuild_source}': {message}") + + +class AurRPCError(Exception): + """ + Error raised when accessing AUR RPC fails. + """ + + def __init__(self, message: str, url: str): + self.message = message + self.url = url + super().__init__(f"Failed to complete AUR RPC request to '{url}': {message}") diff --git a/src/decman/plugins/pacman/fpm.py b/src/decman/plugins/pacman/fpm.py new file mode 100644 index 0000000..3d97348 --- /dev/null +++ b/src/decman/plugins/pacman/fpm.py @@ -0,0 +1,832 @@ +import os +import shutil +import time +import typing + +import decman.core.command as command +import decman.core.error as errors +import decman.core.output as output +import decman.core.store as _store +from decman.plugins.pacman.commands import PacmanCommands +from decman.plugins.pacman.error import ForeignPackageManagerError +from decman.plugins.pacman.package import PackageSearch, PacmanInterface +from decman.plugins.pacman.resolver import DepGraph, ForeignPackage + + +def find_latest_cached_package(store: _store.Store, package: str) -> tuple[str, str] | None: + """ + Returns the latest version and path of a package stored in the built packages cache as a + tuple (version, path). + """ + store.ensure("package_file_cache", {}) + entries = store["package_file_cache"].get(package) + + if entries is None: + return None + + latest_version = None + latest_path = None + latest_timestamp = 0 + + for version, path, timestamp in entries: + if latest_timestamp < timestamp and os.path.exists(path): + latest_timestamp = timestamp + latest_version = version + latest_path = path + + output.print_debug(f"Latest file for {package} is '{latest_path}'.") + + if latest_path is None: + return None + + assert latest_version is not None, "If latest_path is set, then latest_version is set." + return (latest_version, latest_path) + + +def add_package_to_cache(store: _store.Store, package: str, version: str, path_to_built_pkg: str): + """ + Adds a built package to the package file cache. Tries to remove excess cached packages. + """ + store.ensure("package_file_cache", {}) + + new_entry = (version, path_to_built_pkg, int(time.time())) + entries = store["package_file_cache"].get(package, []) + for _, already_cached_path, __ in entries: + if already_cached_path == path_to_built_pkg: + output.print_debug( + f"Trying to cache {package} version {version}, but the version is already cached:\ + {already_cached_path}" + ) + return + entries.append(new_entry) + + store["package_file_cache"][package] = entries + clean_package_cache(store, package) + + +def clean_package_cache(store: _store.Store, package: str): + oldest_path = None + oldest_timestamp = None + index_of_oldest = None + + entries = store["package_file_cache"][package] + output.print_debug(f"Package cache has {len(entries)} entries.") + + number_of_packages_stored_in_cache = 3 + + if len(entries) <= number_of_packages_stored_in_cache: + output.print_debug("Old files will not be removed.") + return + + for index, entry in enumerate(entries): + _, path, timestamp = entry + if oldest_timestamp is None or oldest_timestamp > timestamp: + oldest_timestamp = timestamp + oldest_path = path + index_of_oldest = index + + output.print_debug(f"Oldest cached file for {package} is '{oldest_path}'.") + if oldest_path is None: + return + assert index_of_oldest is not None + + entries.pop(index_of_oldest) + if os.path.exists(oldest_path): + output.print_debug(f"Removing '{oldest_path}' from the package cache.") + try: + os.remove(oldest_path) + except OSError as e: + output.print_error(f"Failed to remove file '{oldest_path}' from the package cache.") + output.print_error(f"{e.strerror or e}") + output.print_continuation("You'll have to remove the file manually.") + + store["package_file_cache"][package] = entries + + +def is_devel(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 + + +class ResolvedDependencies: + """ + Result of dependency resolution. + """ + + def __init__(self) -> None: + self.pacman_deps: set[str] = set() + self.foreign_pkgs: set[str] = set() + self.foreign_dep_pkgs: set[str] = set() + self.foreign_build_dep_pkgs: set[str] = set() + self.build_order: list[str] = [] + self.packages: dict[str, ForeignPackage] = {} + self._pkgbases_to_pkgs: dict[str, set[str]] = {} + self._pkgs_to_pkgbases: dict[str, str] = {} + + def add_pkgbase_info(self, pkgname: str, pkgbase: str): + """ + Adds information about a which package belongs in which package base. + """ + pkgs = self._pkgbases_to_pkgs.get(pkgbase, set()) + pkgs.add(pkgname) + self._pkgbases_to_pkgs[pkgbase] = pkgs + self._pkgs_to_pkgbases[pkgname] = pkgbase + + def get_pkgbase(self, pkgname: str) -> str: + """ + Returns the package base of an package. + """ + return self._pkgs_to_pkgbases[pkgname] + + def get_pkgs_with_common_pkgbase(self, pkgname: str) -> set[str]: + """ + Returns all packages that have the same package base as the given package. + """ + pkgbase = self._pkgs_to_pkgbases[pkgname] + return self._pkgbases_to_pkgs[pkgbase] + + def all_pkgbases(self) -> list[str]: + """ + Returns all pkgbases. + """ + return list(self._pkgbases_to_pkgs) + + def get_some_pkgname(self, pkgbase: str) -> str: + """ + Returns some package name that the given pkgbase has. + """ + return list(self._pkgbases_to_pkgs[pkgbase])[0] + + +class ForeignPackageManager: + """ + Class for dealing with foreign packages. + """ + + def __init__( + self, + store: _store.Store, + pacman: PacmanInterface, + search: PackageSearch, + commands: PacmanCommands, + pkg_cache_dir: str, + build_dir: str, + makepkg_user: str, + ): + self._store = store + self._pacman = pacman + self._search = search + self._commands = commands + self._pkg_cache_dir = pkg_cache_dir + self._build_dir = build_dir + self._makepkg_user = makepkg_user + + def upgrade( + self, + upgrade_devel: bool = False, + force: bool = False, + ignored_pkgs: typing.Optional[set[str]] = None, + ): + """ + Upgrades all foreign packages. + """ + if ignored_pkgs is None: + ignored_pkgs = set() + + output.print_summary("Determining foreign packages to upgrade.") + + all_foreign_pkgs = self._pacman.get_versioned_foreign_packages() + all_explicit_pkgs = set(self._pacman.get_installed()) + output.print_debug(f"Foreign packages to check for upgrades: {all_foreign_pkgs}") + + self._search.try_caching_packages(list(map(lambda p: p[0], all_foreign_pkgs))) + + as_explicit = [] + as_deps = [] + for pkg, ver in all_foreign_pkgs: + if pkg in ignored_pkgs: + continue + + info = self._search.get_package_info(pkg) + if info is None: + raise ForeignPackageManagerError( + f"Failed to find '{pkg}' from AUR or user provided packages." + ) + + if self.should_upgrade_package(pkg, ver, info.version, upgrade_devel): + if pkg in all_explicit_pkgs: + as_explicit.append(pkg) + else: + as_deps.append(pkg) + + output.print_debug( + f"The following foreign packages will be upgraded: {' '.join(as_explicit)}" + ) + + self.install(as_explicit, as_deps, force) + + def install( + self, + foreign_pkgs: list[str], + foreign_dep_pkgs: typing.Optional[list[str]] = None, + force: bool = False, + ): + """ + Installs the given foreign packages and their dependencies (both pacman/AUR). + """ + + if foreign_dep_pkgs is None: + foreign_dep_pkgs = [] + + if len(foreign_pkgs) == 0 and len(foreign_dep_pkgs) == 0: + return + + resolved_dependencies = self.resolve_dependencies(foreign_pkgs, foreign_dep_pkgs) + + output.print_list( + "The following foreign packages will be installed explicitly:", + list(resolved_dependencies.foreign_pkgs), + level=output.SUMMARY, + ) + + output.print_list( + "The following foreign packages will be installed as dependencies:", + list(resolved_dependencies.foreign_dep_pkgs), + level=output.SUMMARY, + ) + + output.print_list( + "The following foreign packages will be built in order to install other packages.\ + They will not be installed:", + list(resolved_dependencies.foreign_build_dep_pkgs), + level=output.SUMMARY, + ) + + if not output.prompt_confirm("Proceed?", default=True): + raise ForeignPackageManagerError("Installing aborted.") + + output.print_summary("Installing foreign package dependencies from pacman.") + self._pacman.install_dependencies(resolved_dependencies.pacman_deps) + + try: + with PackageBuilder( + self._search, + self._store, + self._pacman, + resolved_dependencies, + self._commands, + self._pkg_cache_dir, + self._build_dir, + self._makepkg_user, + ) as builder: + while resolved_dependencies.build_order: + to_build = resolved_dependencies.build_order.pop(0) + + pkgbase = resolved_dependencies.get_pkgbase(to_build) + package_names = resolved_dependencies.get_pkgs_with_common_pkgbase(to_build) + + packages = [ + resolved_dependencies.packages[pkgname] for pkgname in package_names + ] + + builder.build_packages(pkgbase, packages, force) + except OSError as e: + raise ForeignPackageManagerError("Failed to build packages.") from e + + packages_to_install = resolved_dependencies.foreign_pkgs + packages_to_install |= resolved_dependencies.foreign_dep_pkgs + + package_files_to_install = [] + for pkg in packages_to_install: + built_pkg = find_latest_cached_package(self._store, pkg) + assert built_pkg is not None + _, path = built_pkg + package_files_to_install.append(path) + + if package_files_to_install or force: + output.print_summary("Installing foreign packages.") + self._pacman.install_files( + package_files_to_install, + as_explicit=resolved_dependencies.foreign_pkgs, + ) + else: + output.print_summary("No packages to install.") + + def resolve_dependencies( + self, + foreign_pkgs: list[str], + foreign_dep_pkgs: typing.Optional[list[str]] = None, + ) -> ResolvedDependencies: + """ + Resolves foreign dependencies of foreign packages. + """ + + output.print_info("Resolving foreign package dependencies.") + output.print_debug(f"Packages: {foreign_pkgs}") + + if foreign_dep_pkgs is None: + foreign_dep_pkgs = [] + + result = ResolvedDependencies() + result.foreign_pkgs = set(foreign_pkgs) + result.foreign_dep_pkgs = set(foreign_dep_pkgs) + + graph = DepGraph() + + for name in foreign_pkgs + foreign_dep_pkgs: + graph.add_requirement(name, None) + + seen_packages = set(foreign_pkgs + foreign_dep_pkgs) + to_process = foreign_pkgs + foreign_dep_pkgs + total_processed = 0 + + self._search.try_caching_packages(to_process) + + def process_dep(pkgname: str, depname: str, add_to: set[str]): + dep_info = self._search.find_provider(depname) + + if dep_info is None: + raise ForeignPackageManagerError( + f"Failed to find '{depname}' from AUR or user provided packages." + ) + + add_to.add(dep_info.pkgname) + + output.print_debug(f"Adding dependency {dep_info.pkgname} to package {pkgname}.") + graph.add_requirement(dep_info.pkgname, pkgname) + if dep_info.pkgname not in seen_packages: + to_process.append(dep_info.pkgname) + seen_packages.add(dep_info.pkgname) + + while to_process: + pkgname = to_process.pop() + + info = self._search.get_package_info(pkgname) + if info is None: + raise ForeignPackageManagerError( + f"Failed to find '{pkgname}' from AUR or user provided packages." + ) + + result.pacman_deps.update(info.native_dependencies(self._pacman)) + result.add_pkgbase_info(pkgname, info.pkgbase) + + build_deps = info.foreign_make_dependencies( + self._pacman + ) + info.foreign_check_dependencies(self._pacman) + + self._search.try_caching_packages(info.foreign_dependencies(self._pacman) + build_deps) + + for depname in info.foreign_dependencies(self._pacman): + process_dep(pkgname, depname, result.foreign_dep_pkgs) + + for depname in build_deps: + process_dep(pkgname, depname, result.foreign_build_dep_pkgs) + + total_processed += 1 + output.print_info(f"Progress: {total_processed}/{len(seen_packages)}.") + + output.print_info("Determining build order.") + + while True: + to_add = graph.get_and_remove_outer_dep_pkgs() + + if len(to_add) == 0: + break + + for pkg in to_add: + if pkg not in result.packages: + output.print_debug(f"Adding {pkg} to build_order.") + result.build_order.append(pkg.name) + result.packages[pkg.name] = pkg + + return result + + 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 is_devel(package): + output.print_debug(f"Package {package} is devel package. It should be upgraded.") + return True + + try: + cmd = self._commands.compare_versions(installed_version, fetched_version) + returncode, vercmp_output = command.run(cmd) + if returncode != 0: + raise errors.CommandFailedError(cmd, vercmp_output) + + should_upgrade = int(vercmp_output) < 0 + + output.print_debug( + f"Installed version is: {installed_version}. \ + Available version is {fetched_version}. Should upgrade: {should_upgrade}" + ) + return should_upgrade + except (ValueError, errors.CommandFailedError) as error: + output.print_error(f"{error}") + raise ForeignPackageManagerError("Failed to compare versions using vercmp.") from error + + +class PackageBuilder: + """ + Used for building packages in a chroot. + """ + + always_included_packages = ["base-devel", "git"] + + def __init__( + self, + search: PackageSearch, + store: _store.Store, + pacman: PacmanInterface, + resolved_deps: ResolvedDependencies, + commands: PacmanCommands, + pkg_cache_dir: str, + build_dir: str, + makepkg_user: str, + ): + self._search = search + self._store = store + self._pacman = pacman + self._resolved_deps = resolved_deps + self._commands = commands + self.pkg_cache_dir = pkg_cache_dir + self.build_dir = build_dir + self.makepkg_user = makepkg_user + self.valid_pkgexts = [ + ".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", + ] + self.chroot_wd_dir = os.path.join(build_dir, "chroot") + self.chroot_dir = os.path.join(self.chroot_wd_dir, "root") + self.pkgbase_dir_map: dict[str, str] = {} + self.original_wd = "" + self._pkgs_in_chroot = set(PackageBuilder.always_included_packages) + self._pkgs_in_chroot.update(resolved_deps.pacman_deps) + + def __enter__(self): + self.store_wd() + self.create_build_environment() + + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.restore_wd() + self.remove_build_environment() + + def store_wd(self): + """ + Remembers the current working directory as the original working directory. + """ + self.original_wd = os.getcwd() + + def restore_wd(self): + """ + Returns to the original working directory. + """ + os.chdir(self.original_wd) + + def create_build_environment(self): + """ + Creates a new chroot and clones all PKGBUILDS. + """ + output.print_info("Creating a build environment..") + + if os.path.exists(self.build_dir): + output.print_info("Removing previous build directory.") + self.remove_build_environment() + + output.print_info("Getting all PKGBUILDS.") + + # Set up PKGBUILDS + for pkgbase in self._resolved_deps.all_pkgbases(): + pkgbuild_dir = os.path.join(self.build_dir, pkgbase) + self.pkgbase_dir_map[pkgbase] = pkgbuild_dir + os.makedirs(pkgbuild_dir) + os.chdir(pkgbuild_dir) + + pkgbase_info = self._search.get_package_info( + self._resolved_deps.get_some_pkgname(pkgbase) + ) + + assert pkgbase_info is not None, ( + "All dependencies and packages should be resolved \ + during the creation of ResolvedDependencies." + ) + + output.print_debug(f"Git URL for '{pkgbase}' is '{pkgbase_info.git_url}'") + output.print_debug( + f"PKGBUILD directory for '{pkgbase}' is '{pkgbase_info.pkgbuild_directory}'" + ) + self._fetch_and_review_pkgbuild( + pkgbase, pkgbase_info.git_url, pkgbase_info.pkgbuild_directory + ) + shutil.chown(pkgbuild_dir, user=self.makepkg_user) + + output.print_info("Creating a new chroot.") + os.makedirs(self.chroot_wd_dir) + + # 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"] + output.print_debug("Removed GNUPGHOME variable from mkarchroot environment.") + except KeyError: + pass + + cmd = self._commands.make_chroot(self.chroot_dir, self._pkgs_in_chroot) + command.check_run_result( + cmd, command.run(cmd, env_overrides=mkarchroot_env_vars, pass_environment=False) + ) + + def remove_build_environment(self): + """ + Deletes the build environment. + """ + shutil.rmtree(self.build_dir) + + def build_packages(self, package_base: str, packages: list[ForeignPackage], force: bool): + """ + Builds package(s) with the same package base. + + Set force to true to force rebuilds of packages that are already cached + """ + + 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: + output.print_info(f"Skipped building '{' '.join(package_names)}'. Already up to date.") + return + + output.print_info(f"Building '{' '.join(package_names)}'.") + + chroot_new_pacman_pkgs, chroot_pkg_files = self._get_chroot_packages(packages) + + pkgbuild_dir = self.pkgbase_dir_map[package_base] + os.chdir(pkgbuild_dir) + + output.print_debug(f"Chroot dir is: '{self.chroot_dir}', pkgbuild dir is '{pkgbuild_dir}'.") + + output.print_info("Installing build dependencies to chroot.") + + cmd = self._commands.install_chroot( + self.chroot_dir, chroot_new_pacman_pkgs + PackageBuilder.always_included_packages + ) + command.check_run_result(cmd, command.run(cmd)) + output.print_info("Making package.") + + cmd = self._commands.make_chroot_pkg( + self.chroot_wd_dir, self.makepkg_user, chroot_pkg_files + ) + command.check_run_result(cmd, command.run(cmd)) + + for pkgname in package_names: + file = self._find_pkgfile(pkgname, pkgbuild_dir) + + dest = shutil.copy(file, self.pkg_cache_dir) + + pkg_info = self._search.get_package_info(pkgname) + + # Because all dependencies and packages should be resolved during the creation + # of ResolvedDependencies. + assert pkg_info is not None + version = pkg_info.version + + output.print_debug( + f"Adding '{pkgname}', version: '{version}' to cache as file '{dest}'." + ) + + add_package_to_cache(self._store, pkgname, version, dest) + + output.print_info("Removing build dependencies from chroot.") + + if len(chroot_new_pacman_pkgs) != 0: + to_remove = set() + for p in chroot_new_pacman_pkgs: + if p not in self._pkgs_in_chroot: + cmd = self._commands.resolve_real_name_chroot(self.chroot_dir, p) + _, cmd_output = command.check_run_result(cmd, command.run(cmd)) + real_pkgname = cmd_output.strip() + to_remove.add(real_pkgname) + cmd = self._commands.remove_chroot(self.chroot_dir, to_remove) + command.check_run_result(cmd, command.run(cmd)) + + output.print_info(f"Finished building: '{' '.join(package_names)}'.") + + def _are_all_pkgs_cached(self, pkgs: list[ForeignPackage]) -> bool: + for pkg in pkgs: + cache_entry = find_latest_cached_package(self._store, pkg.name) + if cache_entry is None: + return False + cached_version, _ = cache_entry + + pkg_info = self._search.get_package_info(pkg.name) + + # Because all dependencies and packages should be resolved during the creation + # of ResolvedDependencies. git_url should not be None. + assert pkg_info is not None + fetched_version = pkg_info.version + + if cached_version != fetched_version or 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 build dependencies and built foreign pkgs files that are needed + in the chroot before building. pkgs_to_build share the same pkgbase. + """ + chroot_pacman_build_deps = set() + chroot_foreign_pkgs = set() + + def add_to_pacman_build_deps(deps: list[str]): + for dep in deps: + if dep not in self._resolved_deps.pacman_deps: + chroot_pacman_build_deps.add(dep) + + for pkg in pkgs_to_build: + info = self._search.get_package_info(pkg.name) + # Because all dependencies and packages should be resolved during the creation + # of ResolvedDependencies. git_url should not be None. + assert info is not None + + add_to_pacman_build_deps(info.native_make_dependencies(self._pacman)) + add_to_pacman_build_deps(info.native_check_dependencies(self._pacman)) + + foreign_deps = pkg.get_all_recursive_foreign_dep_pkgs() + 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) + # Because all dependencies and packages should be resolved during the creation + # of ResolvedDependencies. git_url should not be None. + assert dep_info is not None + + add_to_pacman_build_deps(dep_info.native_make_dependencies(self._pacman)) + add_to_pacman_build_deps(dep_info.native_check_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 = find_latest_cached_package(self._store, 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_build_deps), 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 self.valid_pkgexts: + if file.name.endswith(ext): + matches.append(file.path) + continue + + if len(matches) != 1: + raise ForeignPackageManagerError( + f"Failed to build package '{pkgname}', because the pkg file cannot be determined.\ + Possible files are: {matches}" + ) + + return matches[0] + + def _fetch_and_review_pkgbuild( + self, pkgbase: str, git_url: str | None, pkgbuild_directory: str | None + ): + """ + Fetches a PKGBUILD to the current directory. + + PKGBUILD will be cloned using git if ``git_url`` is set. + PKGBUILD will be copied from ``pkgbuild_directory`` if it is set. + + The user is prompted to review the PKGBUILD and confirm if the package should be built. + """ + + self._store.ensure("pkgbuild_latest_reviewed_commits", {}) + + if git_url: + cmd = self._commands.git_clone(git_url, ".") + rc, git_output = command.run(cmd) + + if rc != 0: + raise ForeignPackageManagerError( + f"Failed to clone PKGBUILD from {git_url}" + ) from errors.CommandFailedError(cmd, git_output) + + if pkgbuild_directory: + pkgbuild_file = os.path.join(pkgbuild_directory, "PKGBUILD") + try: + shutil.copy(pkgbuild_file, "./PKGBUILD") + except OSError as error: + raise ForeignPackageManagerError( + f"Failed to copy PKGBUILD from {pkgbuild_directory}." + ) from error + + if output.prompt_confirm(f"Review PKGBUILD or show diff for {pkgbase}?", default=True): + latest_reviewed_commit = None + git_commit_ids = [] + + if git_url: + latest_reviewed_commit = self._store["pkgbuild_latest_reviewed_commits"].get( + pkgbase + ) + + cmd = self._commands.git_log_commit_ids() + rc, git_output = command.run(cmd) + + if rc != 0: + raise ForeignPackageManagerError( + f"Failed to get git commit ids for {pkgbase}." + ) from errors.CommandFailedError(cmd, git_output) + + git_commit_ids = git_output.strip().split("\n") + + if latest_reviewed_commit is None or latest_reviewed_commit not in git_commit_ids: + try: + for file in os.scandir("."): + if file.is_file() and not file.name.startswith("."): + cmd = self._commands.review_file(file.path) + rc, review_output = command.pty_run(cmd) + if rc != 0: + raise ForeignPackageManagerError( + f"Failed to review file '{file.path}'." + ) from errors.CommandFailedError(cmd, review_output) + except OSError as error: + raise ForeignPackageManagerError( + f"Failed to review files in directory for {pkgbase}." + ) from error + + else: + cmd = self._commands.git_diff(latest_reviewed_commit) + rc, review_output = command.pty_run(cmd) + if rc != 0: + raise ForeignPackageManagerError( + "Failed to review file using git diff." + ) from errors.CommandFailedError(cmd, review_output) + + if output.prompt_confirm("Build this package?", default=True): + cmd = self._commands.git_get_commit_id() + rc, git_output = command.run(cmd) + if rc != 0: + raise ForeignPackageManagerError( + f"Failed to get commit id for {pkgbase}." + ) from errors.CommandFailedError(cmd, git_output) + commit_id = git_output.strip() + self._store["pkgbuild_latest_reviewed_commits"][pkgbase] = commit_id + else: + raise ForeignPackageManagerError("Building aborted.") diff --git a/src/decman/plugins/pacman/package.py b/src/decman/plugins/pacman/package.py new file mode 100644 index 0000000..2546f26 --- /dev/null +++ b/src/decman/plugins/pacman/package.py @@ -0,0 +1,412 @@ +import dataclasses +import re + +import requests # type: ignore + +import decman.core.output as output +from decman.plugins.pacman.commands import PacmanInterface +from decman.plugins.pacman.error import AurRPCError + + +def strip_dependency(dep: str) -> str: + """ + Removes version spefications from a dependency name. + """ + rx = re.compile("(=.*|>.*|<.*)") + return rx.sub("", dep) + + +@dataclasses.dataclass(frozen=True, slots=True) +class PackageInfo: + """ + Immutable description of a package to be built or installed. + + This class represents *resolved* package metadata and is intended to be + passed around as pure data. + + Exactly one source must be specified: + - ``git_url`` for VCS-based (e.g. AUR) packages + - ``pkgbuild_directory`` for local PKGBUILD-based packages + + Invariants: + - ``pkgname`` uniquely identifies the package. + - ``pkgbase`` groups split packages. + - Exactly one of ``git_url`` or ``pkgbuild_directory`` is set. + - All dependency containers are immutable. + + This object is safe for hashing, set membership, and reuse across runs. + """ + + pkgname: str + pkgbase: str + version: str + + git_url: str | None = None + pkgbuild_directory: str | None = None + provides: tuple[str, ...] = dataclasses.field(default_factory=tuple) + dependencies: tuple[str, ...] = dataclasses.field(default_factory=tuple) + make_dependencies: tuple[str, ...] = dataclasses.field(default_factory=tuple) + check_dependencies: tuple[str, ...] = dataclasses.field(default_factory=tuple) + + def __post_init__(self) -> None: + if self.git_url is None and self.pkgbuild_directory is None: + raise ValueError("Both git_url and pkgbuild_directory cannot be None.") + + if self.git_url is not None and self.pkgbuild_directory is not None: + raise ValueError("Both git_url and pkgbuild_directory cannot be set.") + + def pkg_file_prefix(self) -> str: + """ + Returns the beginning of the file created from building this package. + """ + return f"{self.pkgname}-{self.version}" + + def foreign_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of foreign dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.dependencies: + if not pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + def foreign_make_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of foreign make dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.make_dependencies: + if not pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + def foreign_check_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of foreign check dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.check_dependencies: + if not pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + def native_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of native dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.dependencies: + if pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + def native_make_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of native make dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.make_dependencies: + if pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + def native_check_dependencies(self, pacman: PacmanInterface) -> list[str]: + """ + Returns a list of native check dependencies of this package. + + The dependencies are stripped of their version constrainst if there are any. + """ + result = [] + for dependency in self.check_dependencies: + if pacman.is_installable(dependency): + result.append(strip_dependency(dependency)) + return result + + +class CustomPackage: + """ + Custom package installed from some other location than the official repos or the AUR. + + Exactly one of ``git_url`` or ``pkgbuild_directory`` must be provided. + + Parameters: + ``git_url``: + URL to a git repository containing the PKGBUILD. + + ``pkgbuild_directory``: + Path to the directory containing the PKGBUILD. + """ + + def __init__(self, git_url: str | None, pkgbuild_directory: str | None) -> None: + if git_url is None and pkgbuild_directory is None: + raise ValueError("Both git_url and pkgbuild_directory cannot be None.") + + if git_url is not None and pkgbuild_directory is not None: + raise ValueError("Both git_url and pkgbuild_directory cannot be set.") + + self.git_url = git_url + self.pkgbuild_directory = pkgbuild_directory + + def parse(self) -> PackageInfo: + """ + Parses this package's PKGBUILD to ``PackageInfo``. + + If this fails, raises a ``PKGBUILDParseError``. + """ + raise NotImplementedError + + +class PackageSearch: + """ + Allows searcing for packages / providers from the AUR as well as user defined sources. + + Results are cached and custom packages are preferred. + """ + + def __init__(self, aur_rpc_timeout: int = 30) -> None: + self._package_cache: dict[str, PackageInfo] = {} + self._selected_providers_cache: dict[str, PackageInfo] = {} + self._all_providers_cache: dict[str, list[str]] = {} + self._custom_packages: list[PackageInfo] = [] + self._timeout = aur_rpc_timeout + + def add_custom_pkg(self, user_pkg: PackageInfo): + """ + Adds the given package to custom packages. + """ + self._custom_packages.append(user_pkg) + self._cache_pkg(user_pkg) + + def _cache_pkg(self, pkg: PackageInfo): + for provided_pkg in pkg.provides: + self._all_providers_cache.setdefault(provided_pkg, []).append(pkg.pkgname) + + self._package_cache[pkg.pkgname] = pkg + + def try_caching_packages(self, packages: list[str]): + """ + Tries caching the given packages. Virtual packages may not be cached. + + This can be used before calling get_package_info or find_provider multiple individual + times, because then those methods don't have to make new AUR RPC requests. + """ + + uncached_packages = list(filter(lambda p: p not in self._package_cache, packages)) + + if len(uncached_packages) == 0: + return + + output.print_debug(f"Trying to cache {uncached_packages}.") + + max_pkgs_per_request = 200 + + while uncached_packages: + to_request = map(lambda p: f"arg[]={p}", uncached_packages[:max_pkgs_per_request]) + uncached_packages = uncached_packages[max_pkgs_per_request:] + + url = f"https://aur.archlinux.org/rpc/v5/info?{'&'.join(to_request)}" + output.print_debug(f"Request URL = {url}") + + try: + request = requests.get(url, timeout=self._timeout) + d = request.json() + + if d["type"] == "error": + raise AurRPCError(f"AUR RPC returned error: {d['error']}", url) + + for result in d["results"]: + pkgname = result["Name"] + + if pkgname in self._package_cache: + continue + + for user_package in self._custom_packages: + if user_package.pkgname == pkgname: + output.print_debug(f"'{pkgname}' found in custom packages.") + self._cache_pkg(user_package) + break + else: # if not in user_packages then: + info = PackageInfo( + pkgname=result["Name"], + pkgbase=result["PackageBase"], + version=result["Version"], + dependencies=result.get("Depends", []), + make_dependencies=result.get("MakeDepends", []), + check_dependencies=result.get("CheckDepends", []), + provides=result.get("Provides", []), + git_url=f"https://aur.archlinux.org/{result['PackageBase']}.git", + ) + self._cache_pkg(info) + + output.print_debug("Request completed.") + except (requests.RequestException, KeyError) as e: + raise AurRPCError( + f"Failed to fetch package information for {uncached_packages} from AUR RPC.", + url, + ) from e + + def get_package_info(self, package: str) -> PackageInfo | None: + """ + Returns information about a package. + + If the package is not custom, fetches information from the AUR. + Returns None if no such AUR package exists. + """ + output.print_debug(f"Getting info for package '{package}'.") + + if package in self._package_cache: + output.print_debug(f"'{package}' found in cache.") + return self._package_cache[package] + + # This code is probably not needed since all user packages should be cached + for user_package in self._custom_packages: + if user_package.pkgname == package: + output.print_debug(f"'{package}' found in custom packages.") + self._cache_pkg(user_package) + return user_package + + url = f"https://aur.archlinux.org/rpc/v5/info/{package}" + output.print_debug(f"Requesting info for '{package}' from AUR. URL = {url}") + try: + request = requests.get(url, timeout=self._timeout) + d = request.json() + + if d["type"] == "error": + raise AurRPCError(f"AUR RPC returned error: {d['error']}", url) + + if d["resultcount"] == 0: + output.print_debug(f"'{package}' not found.") + return None + + output.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_dependencies=result.get("MakeDepends", []), + check_dependencies=result.get("CheckDepends", []), + provides=result.get("Provides", []), + git_url=f"https://aur.archlinux.org/{result['PackageBase']}.git", + ) + + self._cache_pkg(info) + + return info + except (requests.RequestException, KeyError) as e: + raise AurRPCError( + f"Failed to fetch package information for {package} from AUR RPC.", + url, + ) from e + + def find_provider(self, stripped_dependency: str) -> PackageInfo | None: + """ + Finds a provider for a dependency. The dependency should not contain version constraints. + + May prompt the user to select if multiple are available. + """ + output.print_debug(f"Finding provider for '{stripped_dependency}'.") + + if stripped_dependency in self._selected_providers_cache: + output.print_debug(f"'{stripped_dependency}' found in cache.") + return self._selected_providers_cache[stripped_dependency] + + output.print_debug("Are there exact name matches?") + + exact_name_match = self.get_package_info(stripped_dependency) + + if exact_name_match is not None: + output.print_debug("Exact name match found.") + self._selected_providers_cache[stripped_dependency] = exact_name_match + return exact_name_match + + output.print_debug("No exact name matches found. Finding providers.") + + known_pkg_results = self._all_providers_cache.get(stripped_dependency, []) + for user_package in self._custom_packages: + if ( + stripped_dependency in user_package.provides + and stripped_dependency not in known_pkg_results + ): + known_pkg_results.append(user_package.pkgname) + + if len(known_pkg_results) == 1: + pkg = self.get_package_info(known_pkg_results[0]) + assert pkg is not None + output.print_debug( + f"Single provider for '{stripped_dependency}' found in known packages: '{pkg}'." + ) + self._selected_providers_cache[stripped_dependency] = pkg + return pkg + + if len(known_pkg_results) > 1: + return self._choose_provider(stripped_dependency, known_pkg_results, "user packages") + + url = f"https://aur.archlinux.org/rpc/v5/search/{stripped_dependency}?by=provides" + output.print_debug( + f"Requesting providers for '{stripped_dependency}' from AUR. URL = {url}" + ) + try: + request = requests.get(url, timeout=self._timeout) + d = request.json() + + if d["type"] == "error": + raise AurRPCError(f"AUR RPC returned error: {d['error']}", url) + + if d["resultcount"] == 0: + output.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] + output.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 AurRPCError( + f"Failed to search for {stripped_dependency} from AUR RPC.", + url, + ) from e + + def _choose_provider( + self, dep: str, possible_providers: list[str], where: str + ) -> PackageInfo | None: + min_selection = 1 + max_selection = len(possible_providers) + output.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} " + output.print_summary(providers) + + selection = output.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._selected_providers_cache[dep] = info + return info diff --git a/src/decman/plugins/pacman/resolver.py b/src/decman/plugins/pacman/resolver.py new file mode 100644 index 0000000..1bf0ce3 --- /dev/null +++ b/src/decman/plugins/pacman/resolver.py @@ -0,0 +1,119 @@ +import typing + +from decman.plugins.pacman.error import DependencyCycleError + + +class ForeignPackage: + """ + Class used to keep track of foreign recursive dependency packages of an foreign package. + """ + + def __init__(self, name: str): + self.name = name + self._all_recursive_foreign_deps: set[str] = 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 __str__(self) -> str: + return f"{self.name}" + + 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_dep_pkgs(self) -> set[str]: + """ + Returns all dependencies and sub-dependencies of the package. + """ + return set(self._all_recursive_foreign_deps) + + +class DepNode: + """ + A Node of the DepGraph + """ + + def __init__(self, package: ForeignPackage) -> None: + self.parents: dict[str, DepNode] = {} + self.children: dict[str, DepNode] = {} + self.pkg = package + + def is_pkgname_in_parents_recursive(self, pkgname: str) -> bool: + """ + Returns True if the given package name is in the parents of this DepNode. + """ + for name, parent in self.parents.items(): + if name == pkgname or parent.is_pkgname_in_parents_recursive(pkgname): + return True + return False + + +class DepGraph: + """ + Represents a graph between foreign packages + """ + + def __init__(self) -> None: + self.package_nodes: dict[str, DepNode] = {} + self._childless_node_names: set[str] = set() + + def add_requirement(self, child_pkgname: str, parent_pkgname: typing.Optional[str]): + """ + Adds a connection between two packages, creating the child package if it doesn't exist. + + The parent is the package that requires the child package. + """ + child_node = self.package_nodes.get(child_pkgname, DepNode(ForeignPackage(child_pkgname))) + self.package_nodes[child_pkgname] = child_node + + if len(child_node.children) == 0: + self._childless_node_names.add(child_pkgname) + + if parent_pkgname is None: + return + + parent_node = self.package_nodes[parent_pkgname] + + if parent_node.is_pkgname_in_parents_recursive(child_pkgname): + raise DependencyCycleError(child_pkgname, parent_pkgname) + + parent_node.children[child_pkgname] = child_node + child_node.parents[parent_pkgname] = parent_node + + if parent_pkgname in self._childless_node_names: + self._childless_node_names.remove(parent_pkgname) + + def get_and_remove_outer_dep_pkgs(self) -> list[ForeignPackage]: + """ + Returns all childless nodes of the dependency package graph and removes them. + """ + new_childless_node_names = set() + result = [] + for childless_node_name in self._childless_node_names: + childless_node = self.package_nodes[childless_node_name] + + for parent in childless_node.parents.values(): + new_deps = childless_node.pkg.get_all_recursive_foreign_dep_pkgs() + new_deps.add(childless_node.pkg.name) + parent.pkg.add_foreign_dependency_packages(new_deps) + del parent.children[childless_node_name] + if len(parent.children) == 0: + new_childless_node_names.add(parent.pkg.name) + + result.append(childless_node.pkg) + self._childless_node_names = new_childless_node_names + return result diff --git a/src/decman/plugins/systemd.py b/src/decman/plugins/systemd.py index 70d1f40..9ab0f95 100644 --- a/src/decman/plugins/systemd.py +++ b/src/decman/plugins/systemd.py @@ -10,6 +10,8 @@ import decman.plugins as plugins def units(fn): """ Annotate that this function returns a set of systemd unit names that should be enabled. + + Return type of ``fn``: ``set[str]`` """ fn.__systemd__units__ = True return fn @@ -19,6 +21,8 @@ def user_units(fn): """ Annotate that this function returns a dict of users and systemd user unit names that should be enabled. + + Return type of ``fn``: ``dict[str, set[str]]`` """ fn.__systemd__user__units__ = True return fn diff --git a/tests/test_decman_plugins_pacman_packages.py b/tests/test_decman_plugins_pacman_packages.py new file mode 100644 index 0000000..16d6f78 --- /dev/null +++ b/tests/test_decman_plugins_pacman_packages.py @@ -0,0 +1,519 @@ + +import pytest + +from decman.plugins.pacman import package as pkg_mod +from decman.plugins.pacman.error import AurRPCError +from decman.plugins.pacman.package import ( + CustomPackage, + PackageInfo, + PackageSearch, + strip_dependency, +) + + +@pytest.fixture(autouse=True) +def silence_output(monkeypatch): + # Avoid real I/O / prompts in tests by default + monkeypatch.setattr(pkg_mod.output, "print_debug", lambda *a, **k: None) + monkeypatch.setattr(pkg_mod.output, "print_summary", lambda *a, **k: None) + monkeypatch.setattr( + pkg_mod.output, + "prompt_number", + lambda *a, **k: 1, # safe default + ) + + +# --- strip_dependency ------------------------------------------------------ + + +@pytest.mark.parametrize( + "dep,expected", + [ + ("foo", "foo"), + ("foo=1.0", "foo"), + ("bar>=2", "bar"), + ("baz<3", "baz"), + ("multi=1.0-2", "multi"), + ], +) +def test_strip_dependency(dep, expected): + assert strip_dependency(dep) == expected + + +# --- PackageInfo ----------------------------------------------------------- + + +def test_packageinfo_requires_exactly_one_source(): + with pytest.raises(ValueError, match="cannot be None"): + PackageInfo(pkgname="a", pkgbase="a", version="1.0") + + with pytest.raises(ValueError, match="cannot be set"): + PackageInfo( + pkgname="a", + pkgbase="a", + version="1.0", + git_url="git://example", + pkgbuild_directory="/tmp", + ) + + +class DummyPacman: + def __init__(self, installable: set[str]): + self._installable = installable + self.calls: list[str] = [] + + def is_installable(self, name: str) -> bool: + self.calls.append(name) + return name in self._installable + + +def _make_pkg_for_deps() -> PackageInfo: + return PackageInfo( + pkgname="pkg", + pkgbase="pkg", + version="1.0", + git_url="git://example", + dependencies=("native>=1", "foreign=2"), + make_dependencies=("make-native", "make-foreign>=3"), + check_dependencies=("check-foreign<4", "check-native"), + ) + + +def test_packageinfo_foreign_and_native_dependencies_are_split_and_stripped(): + pacman = DummyPacman( + { + "native>=1", + "make-native", + "check-native", + } + ) + pkg = _make_pkg_for_deps() + + assert pkg.native_dependencies(pacman) == ["native"] + assert pkg.foreign_dependencies(pacman) == ["foreign"] + assert pkg.native_make_dependencies(pacman) == ["make-native"] + assert pkg.foreign_make_dependencies(pacman) == ["make-foreign"] + assert pkg.native_check_dependencies(pacman) == ["check-native"] + assert pkg.foreign_check_dependencies(pacman) == ["check-foreign"] + + +# --- CustomPackage --------------------------------------------------------- + + +def test_custompackage_requires_exactly_one_source(): + with pytest.raises(ValueError, match="cannot be None"): + CustomPackage(git_url=None, pkgbuild_directory=None) + + with pytest.raises(ValueError, match="cannot be set"): + CustomPackage(git_url="git://example", pkgbuild_directory="/tmp") + + +# --- PackageSearch: caching ------------------------------------------------ + + +def _make_pkg(name: str = "pkg") -> PackageInfo: + return PackageInfo( + pkgname=name, + pkgbase=name, + version="1.0", + git_url=f"git://example/{name}", + provides=("virt-" + name,), + dependencies=("dep",), + make_dependencies=(), + check_dependencies=(), + ) + + +def test_add_custom_pkg_caches_package(): + search = PackageSearch() + pkg = _make_pkg("foo") + + search.add_custom_pkg(pkg) + + assert pkg in search._custom_packages + assert search._package_cache["foo"] is pkg + assert search._all_providers_cache["virt-foo"] == ["foo"] + + +def test_try_caching_packages_skips_already_cached(monkeypatch): + search = PackageSearch() + pkg = _make_pkg("foo") + search._cache_pkg(pkg) + + calls = [] + + def fake_get(*args, **kwargs): + calls.append((args, kwargs)) + raise AssertionError("requests.get should not be called") + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + search.try_caching_packages(["foo"]) + assert calls == [] + + +def test_try_caching_packages_caches_from_aur(monkeypatch): + search = PackageSearch() + + def fake_get(url, timeout): + class Resp: + def json(self): + return { + "type": "success", + "results": [ + { + "Name": "bar", + "PackageBase": "bar-base", + "Version": "2.0", + "Depends": ["dep1"], + "MakeDepends": ["make1"], + "CheckDepends": ["check1"], + "Provides": ["virt-bar"], + } + ], + } + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + search.try_caching_packages(["bar"]) + + assert "bar" in search._package_cache + info = search._package_cache["bar"] + assert isinstance(info, PackageInfo) + assert search._all_providers_cache["virt-bar"] == ["bar"] + + +def test_try_caching_packages_aur_returns_error(monkeypatch): + search = PackageSearch() + + def fake_get(url, timeout): + class Resp: + def json(self): + return {"type": "error", "error": "boom"} + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.try_caching_packages(["bar"]) + + +def test_try_caching_packages_request_exception_raises_aur_error(monkeypatch): + search = PackageSearch() + + class DummyError(pkg_mod.requests.RequestException): + pass + + def fake_get(url, timeout): + raise DummyError("boom") + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.try_caching_packages(["bar"]) + + +# --- PackageSearch: get_package_info -------------------------------------- + + +def test_get_package_info_returns_from_cache(): + search = PackageSearch() + pkg = _make_pkg("foo") + search._cache_pkg(pkg) + + result = search.get_package_info("foo") + assert result is pkg + + +def test_get_package_info_returns_custom_package_if_not_cached(): + search = PackageSearch() + pkg = _make_pkg("foo") + search._custom_packages.append(pkg) + + result = search.get_package_info("foo") + + assert result is pkg + assert search._package_cache["foo"] is pkg + + +def test_get_package_info_aur_not_found_returns_none(monkeypatch): + search = PackageSearch() + + def fake_get(url, timeout): + class Resp: + def json(self): + return {"type": "success", "resultcount": 0, "results": []} + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + result = search.get_package_info("foo") + assert result is None + assert "foo" not in search._package_cache + + +def test_get_package_info_aur_success_caches_and_returns(monkeypatch): + search = PackageSearch() + + def fake_get(url, timeout): + class Resp: + def json(self): + return { + "type": "success", + "resultcount": 1, + "results": [ + { + "Name": "foo", + "PackageBase": "foo-base", + "Version": "1.2", + "Depends": ["dep1"], + "MakeDepends": ["make1"], + "CheckDepends": ["check1"], + "Provides": ["virt-foo"], + } + ], + } + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + result = search.get_package_info("foo") + assert isinstance(result, PackageInfo) + assert result.pkgname == "foo" + assert search._package_cache["foo"] is result + + +def test_get_package_info_aur_returns_error(monkeypatch): + search = PackageSearch() + + def fake_get(url, timeout): + class Resp: + def json(self): + return {"type": "error", "error": "boom"} + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.get_package_info("foo") + + +def test_get_package_info_request_exception_raises_aur_error(monkeypatch): + search = PackageSearch() + + class DummyError(pkg_mod.requests.RequestException): + pass + + def fake_get(url, timeout): + raise DummyError("boom") + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.get_package_info("foo") + + +# --- PackageSearch: find_provider ----------------------------------------- + + +def test_find_provider_uses_selected_providers_cache(): + search = PackageSearch() + pkg = _make_pkg("foo") + search._selected_providers_cache["dep"] = pkg + + result = search.find_provider("dep") + assert result is pkg + + +def test_find_provider_exact_name_match(monkeypatch): + search = PackageSearch() + pkg = _make_pkg("dep") + + def fake_get_package_info(name: str): + assert name == "dep" + return pkg + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + result = search.find_provider("dep") + assert result is pkg + assert search._selected_providers_cache["dep"] is pkg + + +def test_find_provider_single_known_provider(monkeypatch): + search = PackageSearch() + pkg = _make_pkg("provider") + search._all_providers_cache["dep"] = ["provider"] + + def fake_get_package_info(name: str): + if name == "dep": + return None + assert name == "provider" + return pkg + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + result = search.find_provider("dep") + assert result is pkg + assert search._selected_providers_cache["dep"] is pkg + + +def test_find_provider_aur_search_not_found(monkeypatch): + search = PackageSearch() + + def fake_get_package_info(name: str): + # Exact name match should fail + return None + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + def fake_get(url, timeout): + class Resp: + def json(self): + return {"type": "success", "resultcount": 0, "results": []} + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + result = search.find_provider("dep") + assert result is None + + +def test_find_provider_aur_search_single_result(monkeypatch): + search = PackageSearch() + pkg = _make_pkg("provider") + + def fake_get_package_info(name: str): + # first call for stripped_dependency -> None + if name == "dep": + return None + assert name == "provider" + return pkg + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + def fake_get(url, timeout): + class Resp: + def json(self): + return { + "type": "success", + "resultcount": 1, + "results": [{"Name": "provider"}], + } + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + result = search.find_provider("dep") + assert result is pkg + + +def test_find_provider_aur_search_multiple_results_calls_choose_provider(monkeypatch): + search = PackageSearch() + + def fake_get_package_info(name: str): + # no exact match + return None + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + def fake_get(url, timeout): + class Resp: + def json(self): + return { + "type": "success", + "resultcount": 2, + "results": [{"Name": "a"}, {"Name": "b"}], + } + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + sentinel = object() + + def fake_choose(dep, providers, where): + assert dep == "dep" + assert providers == ["a", "b"] + assert where == "AUR" + return sentinel + + monkeypatch.setattr(search, "_choose_provider", fake_choose) + + result = search.find_provider("dep") + assert result is sentinel + + +def test_find_provider_aur_search_error(monkeypatch): + search = PackageSearch() + + def fake_get_package_info(name: str): + return None + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + def fake_get(url, timeout): + class Resp: + def json(self): + return {"type": "error", "error": "boom"} + + return Resp() + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.find_provider("dep") + + +def test_find_provider_aur_search_request_exception_raises_aur_error(monkeypatch): + search = PackageSearch() + + def fake_get_package_info(name: str): + return None + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + class DummyError(pkg_mod.requests.RequestException): + pass + + def fake_get(url, timeout): + raise DummyError("boom") + + monkeypatch.setattr(pkg_mod.requests, "get", fake_get) + + with pytest.raises(AurRPCError): + search.find_provider("dep") + + +# --- PackageSearch: _choose_provider -------------------------------------- + + +def test_choose_provider_prompts_and_caches(monkeypatch): + search = PackageSearch() + providers = ["a", "b", "c"] + selected_pkg = _make_pkg("b") + + # override prompt to select "2" (provider "b") + monkeypatch.setattr( + pkg_mod.output, + "prompt_number", + lambda *a, **k: 2, + ) + + def fake_get_package_info(name: str): + assert name == "b" + return selected_pkg + + monkeypatch.setattr(search, "get_package_info", fake_get_package_info) + + result = search._choose_provider("dep", providers, "AUR") + assert result is selected_pkg + assert search._selected_providers_cache["dep"] is selected_pkg diff --git a/tests/test_decman_plugins_pacman_resolver.py b/tests/test_decman_plugins_pacman_resolver.py new file mode 100644 index 0000000..7fb923e --- /dev/null +++ b/tests/test_decman_plugins_pacman_resolver.py @@ -0,0 +1,88 @@ +import pytest + +from decman.plugins.pacman.error import DependencyCycleError +from decman.plugins.pacman.resolver import DepGraph, ForeignPackage + + +def test_add_dependency(): + graph = DepGraph() + + graph.add_requirement("A", None) + graph.add_requirement("B1", "A") + graph.add_requirement("B2", "A") + graph.add_requirement("C", "B1") + + assert "B1" in graph.package_nodes["A"].children + assert "B2" in graph.package_nodes["A"].children + assert "C" in graph.package_nodes["B1"].children + + +def test_cyclic_dependency_raises(): + graph = DepGraph() + + graph.add_requirement("A", None) + graph.add_requirement("B", "A") + graph.add_requirement("C", "B") + + with pytest.raises(DependencyCycleError): + graph.add_requirement("A", "C") + + +def _build_graph_for_outer_deps() -> DepGraph: + graph = DepGraph() + + # Roots + graph.add_requirement("A", None) + graph.add_requirement("V", None) + + # Level B + graph.add_requirement("B1", "A") + graph.add_requirement("B2", "A") + graph.add_requirement("B3", "A") + + # Extra dependency B1 -> B2 + graph.add_requirement("B1", "B2") + + # Level C + graph.add_requirement("C1", "B1") + graph.add_requirement("C2", "B1") + + # Level D + cycle-ish edges + graph.add_requirement("D", "C1") + graph.add_requirement("C2", "D") + + # Foreign packages and their foreign deps + defs = { + "V": [], + "A": ["B1", "B2", "B3", "C1", "C2", "D"], + "B1": ["C1", "C2", "D"], + "B2": ["B1", "C1", "C2", "D"], + "B3": [], + "C1": ["D", "C2"], + "C2": [], + "D": ["C2"], + } + + for name, deps in defs.items(): + pkg = ForeignPackage(name) + pkg.add_foreign_dependency_packages(deps) + + return graph + + +def _assert_outer_dep_names(graph: DepGraph, expected: set[str]) -> None: + result = graph.get_and_remove_outer_dep_pkgs() + names = {pkg.name for pkg in result} + assert names == expected + + +def test_get_and_remove_outer_deps_sequence(): + graph = _build_graph_for_outer_deps() + + _assert_outer_dep_names(graph, {"C2", "B3", "V"}) + _assert_outer_dep_names(graph, {"D"}) + _assert_outer_dep_names(graph, {"C1"}) + _assert_outer_dep_names(graph, {"B1"}) + _assert_outer_dep_names(graph, {"B2"}) + _assert_outer_dep_names(graph, {"A"}) + _assert_outer_dep_names(graph, set())