From 6fe1f4cb6fb2cdda79fc82337f7df9d801b7000c Mon Sep 17 00:00:00 2001 From: Kivi Kaitaniemi Date: Sun, 14 Dec 2025 22:28:08 +0200 Subject: [PATCH] Add AUR plugin --- pyproject.toml | 2 +- src/decman/__init__.py | 6 + src/decman/app.py | 2 +- src/decman/config.py | 1 + src/decman/plugins/aur/__init__.py | 228 +++++++++++++ .../plugins/{pacman => aur}/commands.py | 183 ++--------- src/decman/plugins/{pacman => aur}/error.py | 0 src/decman/plugins/{pacman => aur}/fpm.py | 20 +- src/decman/plugins/{pacman => aur}/package.py | 32 +- .../plugins/{pacman => aur}/resolver.py | 2 +- src/decman/plugins/pacman.py | 311 ++++++++++++++++++ src/decman/plugins/pacman/__init__.py | 104 ------ tests/test_decman_plugins_aur.py | 302 +++++++++++++++++ ....py => test_decman_plugins_aur_package.py} | 6 +- ...py => test_decman_plugins_aur_resolver.py} | 4 +- tests/test_decman_plugins_pacman.py | 291 ++++++++++++++++ 16 files changed, 1203 insertions(+), 291 deletions(-) create mode 100644 src/decman/plugins/aur/__init__.py rename src/decman/plugins/{pacman => aur}/commands.py (53%) rename src/decman/plugins/{pacman => aur}/error.py (100%) rename src/decman/plugins/{pacman => aur}/fpm.py (98%) rename src/decman/plugins/{pacman => aur}/package.py (95%) rename src/decman/plugins/{pacman => aur}/resolver.py (98%) create mode 100644 src/decman/plugins/pacman.py delete mode 100644 src/decman/plugins/pacman/__init__.py create mode 100644 tests/test_decman_plugins_aur.py rename tests/{test_decman_plugins_pacman_package.py => test_decman_plugins_aur_package.py} (99%) rename tests/{test_decman_plugins_pacman_resolver.py => test_decman_plugins_aur_resolver.py} (94%) create mode 100644 tests/test_decman_plugins_pacman.py diff --git a/pyproject.toml b/pyproject.toml index 95821b8..3f111fb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ decman = "decman.app:main" [project.entry-points."decman.plugins"] systemd = "decman.plugins.systemd:Systemd" pacman = "decman.plugins.pacman:Pacman" -aur = "decman.plugins.pacman:AUR" +aur = "decman.plugins.aur:AUR" [dependency-groups] dev = [ diff --git a/src/decman/__init__.py b/src/decman/__init__.py index 4cb2e0b..0874660 100644 --- a/src/decman/__init__.py +++ b/src/decman/__init__.py @@ -9,6 +9,7 @@ from decman.core.error import SourceError from decman.core.fs import Directory, File from decman.core.module import Module from decman.plugins import Plugin, available_plugins +from decman.plugins.aur import AUR # Plugin types from decman.plugins.pacman import Pacman @@ -41,12 +42,17 @@ execution_order: list[str] = [ # Default plugins get quick access pacman: None | Pacman = None +aur: None | AUR = None systemd: None | Systemd = None _pacman = plugins.get("pacman", None) if isinstance(_pacman, Pacman): pacman = _pacman +_aur = plugins.get("aur", None) +if isinstance(_aur, AUR): + aur = _aur + _systemd = plugins.get("systemd", None) if isinstance(_systemd, Systemd): systemd = _systemd diff --git a/src/decman/app.py b/src/decman/app.py index 03005b9..166ddd2 100644 --- a/src/decman/app.py +++ b/src/decman/app.py @@ -18,7 +18,7 @@ def main(): Main entry for the CLI app """ - sys.pycache_prefix = os.path.join(conf.pkg_cache_dir, "python/") + sys.pycache_prefix = os.path.join(conf.cache_dir, "python/") parser = argparse.ArgumentParser( prog="decman", diff --git a/src/decman/config.py b/src/decman/config.py index bfdda4d..fb343b2 100644 --- a/src/decman/config.py +++ b/src/decman/config.py @@ -25,5 +25,6 @@ quiet_output: bool = False color_output: bool = True module_on_disable_scripts_dir: str = "/var/lib/decman/scripts/" +cache_dir: str = "/var/cache/decman" arch: str = "x86_64" diff --git a/src/decman/plugins/aur/__init__.py b/src/decman/plugins/aur/__init__.py new file mode 100644 index 0000000..b1064d0 --- /dev/null +++ b/src/decman/plugins/aur/__init__.py @@ -0,0 +1,228 @@ +import os +import shutil + +import decman.config as config +import decman.core.error as errors +import decman.core.module as module +import decman.core.output as output +import decman.core.store as _store +import decman.plugins as plugins +from decman.plugins.aur.commands import AurCommands, AurPacmanInterface +from decman.plugins.aur.error import ( + AurRPCError, + DependencyCycleError, + ForeignPackageManagerError, + PKGBUILDParseError, +) +from decman.plugins.aur.fpm import ForeignPackageManager +from decman.plugins.aur.package import CustomPackage, PackageSearch + +# Re-exports +__all__ = [ + "AUR", + "AurCommands", + "CustomPackage", + "aur_packages", + "custom_packages", +] + + +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 AUR(plugins.Plugin): + """ + Plugin that manages additional pacman packages installed outside the pacman repos. + + AUR packages are added directly to ``packages`` or declared by modules via ``@aur_packages``. + + Custom packages are added directly to ``custom_packages`` or declared by modules via + ``@custom_packages``. + """ + + NAME = "aur" + + def __init__(self) -> None: + self.packages: set[str] = set() + self.custom_packages: set[CustomPackage] = set() + self.ignored_packages: set[str] = set() + self.commands: AurCommands = AurCommands() + + self.aur_rpc_timeout: int = 30 + self.print_highlights: bool = True + self.keywords: set[str] = { + "pacsave", + "pacnew", + # These cause too many false positives IMO + # "warning", + # "error", + # "note", + } + self.build_dir: str = "/tmp/decman/build" + self.makepkg_user: str = "nobody" + + def available(self) -> bool: + return ( + shutil.which("pacman") is not None + and shutil.which("git") is not None + and shutil.which("mkarchroot") is not None + ) + + def process_modules(self, store: _store.Store, modules: set[module.Module]): + # This is used to track changes in modules. + store.ensure("aur_packages_for_module", {}) + store.ensure("custom_packages_for_module", {}) + + for mod in modules: + store["aur_packages_for_module"].setdefault(mod.name, set()) + store["custom_packages_for_module"].setdefault(mod.name, set()) + + aur_packages = plugins.run_method_with_attribute(mod, "__aur__packages__") or set() + custom_packages = ( + plugins.run_method_with_attribute(mod, "__custom__packages__") or set() + ) + + if store["aur_packages_for_module"][mod.name] != aur_packages: + mod._changed = True + + if store["custom_packages_for_module"][mod.name] != custom_packages: + mod._changed = True + + self.packages |= aur_packages + self.custom_packages |= custom_packages + + store["aur_packages_for_module"][mod.name] = aur_packages + store["custom_packages_for_module"][mod.name] = custom_packages + + def apply( + self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None + ) -> bool: + params = params or [] + upgrade_devel = "aur-upgrade-devel" in params + force = "aur-force" in params + pkg_cache_dir = os.path.join(config.cache_dir, "aur/") + + if not dry_run: + try: + os.makedirs(pkg_cache_dir, exist_ok=True) + except OSError as error: + output.print_error("Failed to ensure AUR package cache directory exists.") + output.print_continuation(f"{error.strerror or error}") + output.print_traceback() + + return False + + try: + package_search = PackageSearch(self.aur_rpc_timeout) + for custom_package in self.custom_packages: + package_search.add_custom_pkg(custom_package.parse(self.commands)) + pm = AurPacmanInterface(self.commands, self.print_highlights, self.keywords) + fpm = ForeignPackageManager( + store, + pm, + package_search, + self.commands, + pkg_cache_dir, + self.build_dir, + self.makepkg_user, + ) + + custom_package_names = {p.pkgname for p in self.custom_packages} + currently_installed_native = pm.get_native_explicit() + currently_installed_foreign = pm.get_foreign_explicit() + orphans = pm.get_foreign_orphans() + + to_remove = ( + (currently_installed_foreign | orphans) + - self.packages + - custom_package_names + - self.ignored_packages + ) + + actually_to_remove = set() + to_set_as_dependencies = set() + + dependants_to_keep = ( + self.packages + | custom_package_names + | currently_installed_native + # don't remove ignored packages' dependencies + | (self.ignored_packages & currently_installed_foreign) + ) + for package in to_remove: + dependants = pm.get_dependants(package) + if any(dependant in dependants_to_keep for dependant in dependants): + to_set_as_dependencies.add(package) + else: + actually_to_remove.add(package) + + if actually_to_remove: + output.print_list("Removing foreign packages:", sorted(actually_to_remove)) + if not dry_run: + pm.remove(actually_to_remove) + + if to_set_as_dependencies: + output.print_list( + "Setting previously explicitly installed foreign packages as dependencies:", + sorted(to_set_as_dependencies), + ) + if not dry_run: + pm.set_as_dependencies(to_set_as_dependencies) + + output.print_summary("Upgrading foreign packages.") + if not dry_run: + fpm.upgrade(upgrade_devel, force, self.ignored_packages) + + to_install = ( + (self.packages | custom_package_names) + - currently_installed_foreign + - self.ignored_packages + ) + output.print_list("Installing foreign packages:", sorted(to_install)) + + if not dry_run: + fpm.install(list(to_install), force=force) + except AurRPCError as error: + output.print_error("Failed to fetch data from AUR RPC.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + except DependencyCycleError as error: + output.print_error("Foreign package dependency cycle detected.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + except PKGBUILDParseError as error: + output.print_error("Failed to parse a CustomPackage PKGBUILD.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + except ForeignPackageManagerError as error: + output.print_error("Foreign package manager failed.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + except errors.CommandFailedError as error: + output.print_error("Running a command failed.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + + return True diff --git a/src/decman/plugins/pacman/commands.py b/src/decman/plugins/aur/commands.py similarity index 53% rename from src/decman/plugins/pacman/commands.py rename to src/decman/plugins/aur/commands.py index b202489..4acfb76 100644 --- a/src/decman/plugins/pacman/commands.py +++ b/src/decman/plugins/aur/commands.py @@ -1,28 +1,14 @@ import decman.core.command as command import decman.core.error as errors -import decman.core.output as output +import decman.plugins.pacman as pacman -class PacmanCommands: - def list_explicit(self) -> list[str]: +class AurCommands(pacman.PacmanCommands): + def list_orphans_foreign(self) -> list[str]: """ - Running this command outputs a newline seperated list of explicitly installed native - packages. + Running this command outputs a newline seperated list of orphaned foreign 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] + return ["pacman", "-Qmdtq", "--color=never"] def list_foreign_versioned(self) -> list[str]: """ @@ -37,12 +23,6 @@ class PacmanCommands: """ 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. @@ -56,31 +36,6 @@ class PacmanCommands: """ 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. @@ -181,31 +136,32 @@ class PacmanCommands: return ["makepkg", "--printsrcinfo"] -class PacmanInterface: +class AurPacmanInterface(pacman.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: + def __init__(self, commands: AurCommands, print_highlights: bool, keywords: set[str]) -> None: + super().__init__(commands, print_highlights, keywords) self._installable: dict[str, bool] = {} - self._commands = commands - self._print_highlights = print_highlights - self._keywords = keywords + self._aur_commands = commands - def get_installed(self) -> list[str]: + def get_foreign_orphans(self) -> set[str]: """ - Returns a list of installed packages. + Returns a set of orphaned foreign packages. """ - returncode, packages_text = command.run(self._commands.list_explicit()) - packages = packages_text.strip().split("\n") + cmd = self._aur_commands.list_orphans_foreign() + rc, packages_text = command.run(cmd) + # returncode 1 means no packages exist + if rc == 1: + return set() + if rc != 0: + raise errors.CommandFailedError(cmd, packages_text) - if returncode != 0: - raise errors.CommandFailedError(self._commands.list_explicit(), packages_text) + packages = set(packages_text.strip().split("\n")) return packages @@ -216,7 +172,7 @@ class PacmanInterface: if pkg in self._installable: return self._installable[pkg] - returncode, _ = command.run(self._commands.is_installable(pkg)) + returncode, _ = command.run(self._aur_commands.is_installable(pkg)) result = returncode == 0 self._installable[pkg] = result @@ -227,39 +183,21 @@ class PacmanInterface: Returns a list of installed packages and their versions that aren't from pacman databases, basically AUR packages. """ - cmd = self._commands.list_foreign_versioned() + cmd = self._aur_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") ] + # returncode 1 means no packages exist + if returncode == 1: + return [] + 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. @@ -267,12 +205,8 @@ class PacmanInterface: 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) - + cmd = self._aur_commands.install_as_dependencies(deps) + _, pacman_output = command.check_run_result(cmd, command.pty_run(cmd)) self.print_highlighted_pacman_messages(pacman_output) def install_files(self, files: list[str], as_explicit: set[str]): @@ -283,69 +217,12 @@ class PacmanInterface: 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) - + cmd = self._aur_commands.install_files_as_dependencies(files) + _, pacman_output = command.check_run_result(cmd, command.pty_run(cmd)) 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 + _, pacman_output = command.check_run_result(cmd, command.run(cmd)) diff --git a/src/decman/plugins/pacman/error.py b/src/decman/plugins/aur/error.py similarity index 100% rename from src/decman/plugins/pacman/error.py rename to src/decman/plugins/aur/error.py diff --git a/src/decman/plugins/pacman/fpm.py b/src/decman/plugins/aur/fpm.py similarity index 98% rename from src/decman/plugins/pacman/fpm.py rename to src/decman/plugins/aur/fpm.py index 3d97348..b741dde 100644 --- a/src/decman/plugins/pacman/fpm.py +++ b/src/decman/plugins/aur/fpm.py @@ -7,10 +7,10 @@ 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 +from decman.plugins.aur.commands import AurCommands +from decman.plugins.aur.error import ForeignPackageManagerError +from decman.plugins.aur.package import AurPacmanInterface, PackageSearch +from decman.plugins.aur.resolver import DepGraph, ForeignPackage def find_latest_cached_package(store: _store.Store, package: str) -> tuple[str, str] | None: @@ -179,9 +179,9 @@ class ForeignPackageManager: def __init__( self, store: _store.Store, - pacman: PacmanInterface, + pacman: AurPacmanInterface, search: PackageSearch, - commands: PacmanCommands, + commands: AurCommands, pkg_cache_dir: str, build_dir: str, makepkg_user: str, @@ -209,7 +209,7 @@ class ForeignPackageManager: 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()) + all_explicit_foreign_pkgs = set(self._pacman.get_foreign_explicit()) 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))) @@ -227,7 +227,7 @@ class ForeignPackageManager: ) if self.should_upgrade_package(pkg, ver, info.version, upgrade_devel): - if pkg in all_explicit_pkgs: + if pkg in all_explicit_foreign_pkgs: as_explicit.append(pkg) else: as_deps.append(pkg) @@ -458,9 +458,9 @@ class PackageBuilder: self, search: PackageSearch, store: _store.Store, - pacman: PacmanInterface, + pacman: AurPacmanInterface, resolved_deps: ResolvedDependencies, - commands: PacmanCommands, + commands: AurCommands, pkg_cache_dir: str, build_dir: str, makepkg_user: str, diff --git a/src/decman/plugins/pacman/package.py b/src/decman/plugins/aur/package.py similarity index 95% rename from src/decman/plugins/pacman/package.py rename to src/decman/plugins/aur/package.py index 3f46e65..dae9634 100644 --- a/src/decman/plugins/pacman/package.py +++ b/src/decman/plugins/aur/package.py @@ -10,8 +10,8 @@ import decman.config as config import decman.core.command as command import decman.core.error as errors import decman.core.output as output -from decman.plugins.pacman.commands import PacmanCommands, PacmanInterface -from decman.plugins.pacman.error import AurRPCError, PKGBUILDParseError +from decman.plugins.aur.commands import AurCommands, AurPacmanInterface +from decman.plugins.aur.error import AurRPCError, PKGBUILDParseError def strip_dependency(dep: str) -> str: @@ -91,7 +91,7 @@ class PackageInfo: # --- public API --------------------------------------------------------- - def foreign_dependencies(self, pacman: PacmanInterface) -> list[str]: + def foreign_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of foreign dependencies of this package. @@ -101,7 +101,7 @@ class PackageInfo: assert self._foreign_dependencies is not None return list(self._foreign_dependencies) - def foreign_make_dependencies(self, pacman: PacmanInterface) -> list[str]: + def foreign_make_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of foreign make dependencies of this package. @@ -111,7 +111,7 @@ class PackageInfo: assert self._foreign_make_dependencies is not None return list(self._foreign_make_dependencies) - def foreign_check_dependencies(self, pacman: PacmanInterface) -> list[str]: + def foreign_check_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of foreign check dependencies of this package. @@ -121,7 +121,7 @@ class PackageInfo: assert self._foreign_check_dependencies is not None return list(self._foreign_check_dependencies) - def native_dependencies(self, pacman: PacmanInterface) -> list[str]: + def native_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of native dependencies of this package. @@ -131,7 +131,7 @@ class PackageInfo: assert self._native_dependencies is not None return list(self._native_dependencies) - def native_make_dependencies(self, pacman: PacmanInterface) -> list[str]: + def native_make_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of native make dependencies of this package. @@ -141,7 +141,7 @@ class PackageInfo: assert self._native_make_dependencies is not None return list(self._native_make_dependencies) - def native_check_dependencies(self, pacman: PacmanInterface) -> list[str]: + def native_check_dependencies(self, pacman: AurPacmanInterface) -> list[str]: """ Returns a list of native check dependencies of this package. @@ -155,7 +155,7 @@ class PackageInfo: @staticmethod def _classify_dependencies( - deps: tuple[str, ...], pacman: PacmanInterface + deps: tuple[str, ...], pacman: AurPacmanInterface ) -> tuple[tuple[str, ...], tuple[str, ...]]: native: list[str] = [] foreign: list[str] = [] @@ -169,7 +169,7 @@ class PackageInfo: return tuple(native), tuple(foreign) - def _ensure_dependencies_cached(self, pacman: PacmanInterface) -> None: + def _ensure_dependencies_cached(self, pacman: AurPacmanInterface) -> None: if self._native_dependencies is not None: return @@ -177,7 +177,7 @@ class PackageInfo: object.__setattr__(self, "_native_dependencies", native) object.__setattr__(self, "_foreign_dependencies", foreign) - def _ensure_make_dependencies_cached(self, pacman: PacmanInterface) -> None: + def _ensure_make_dependencies_cached(self, pacman: AurPacmanInterface) -> None: if self._native_make_dependencies is not None: return @@ -185,7 +185,7 @@ class PackageInfo: object.__setattr__(self, "_native_make_dependencies", native) object.__setattr__(self, "_foreign_make_dependencies", foreign) - def _ensure_check_dependencies_cached(self, pacman: PacmanInterface) -> None: + def _ensure_check_dependencies_cached(self, pacman: AurPacmanInterface) -> None: if self._native_check_dependencies is not None: return @@ -224,7 +224,7 @@ class CustomPackage: self.git_url = git_url self.pkgbuild_directory = pkgbuild_directory - def parse(self, commands: PacmanCommands) -> PackageInfo: + def parse(self, commands: AurCommands) -> PackageInfo: """ Parses this package's PKGBUILD to ``PackageInfo``. @@ -256,7 +256,7 @@ class CustomPackage: f"CustomPackage(pkgname={self.pkgname}, pkgbuild_directory={self.pkgbuild_directory})" ) - def _srcinfo_from_pkgbuild_directory(self, commands: PacmanCommands) -> str: + def _srcinfo_from_pkgbuild_directory(self, commands: AurCommands) -> str: assert self.pkgbuild_directory is not None, ( "This will not get called if pkgbuild_directory is unset." ) @@ -276,7 +276,7 @@ class CustomPackage: return self._run_makepkg_printsrcinfo(path, commands) - def _srcinfo_from_git(self, commands: PacmanCommands) -> str: + def _srcinfo_from_git(self, commands: AurCommands) -> str: assert self.git_url is not None, "This will not get called if git_url is unset." with tempfile.TemporaryDirectory(prefix="decman-pkgbuild-") as tmpdir: tmp_path = pathlib.Path(tmpdir) @@ -297,7 +297,7 @@ class CustomPackage: return self._run_makepkg_printsrcinfo(tmp_path, commands) - def _run_makepkg_printsrcinfo(self, path: pathlib.Path, commands: PacmanCommands) -> str: + def _run_makepkg_printsrcinfo(self, path: pathlib.Path, commands: AurCommands) -> str: orig_wd = os.getcwd() try: os.chdir(path) diff --git a/src/decman/plugins/pacman/resolver.py b/src/decman/plugins/aur/resolver.py similarity index 98% rename from src/decman/plugins/pacman/resolver.py rename to src/decman/plugins/aur/resolver.py index 1bf0ce3..754c297 100644 --- a/src/decman/plugins/pacman/resolver.py +++ b/src/decman/plugins/aur/resolver.py @@ -1,6 +1,6 @@ import typing -from decman.plugins.pacman.error import DependencyCycleError +from decman.plugins.aur.error import DependencyCycleError class ForeignPackage: diff --git a/src/decman/plugins/pacman.py b/src/decman/plugins/pacman.py new file mode 100644 index 0000000..f363938 --- /dev/null +++ b/src/decman/plugins/pacman.py @@ -0,0 +1,311 @@ +import shutil + +import decman.core.command as command +import decman.core.error as errors +import decman.core.module as module +import decman.core.output as output +import decman.core.store as _store +import decman.plugins as plugins + + +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 + + +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.ignored_packages: set[str] = set() + self.commands = PacmanCommands() + self.print_highlights = True + self.keywords = { + "pacsave", + "pacnew", + # These cause too many false positives IMO + # "warning", + # "error", + # "note", + } + + 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", {}) + + for mod in modules: + store["packages_for_module"].setdefault(mod.name, set()) + + packages = plugins.run_method_with_attribute(mod, "__pacman__packages__") or set() + + if store["packages_for_module"][mod.name] != packages: + mod._changed = True + + self.packages |= packages + + store["packages_for_module"][mod.name] = packages + + def apply( + self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None + ) -> bool: + pm = PacmanInterface(self.commands, self.print_highlights, self.keywords) + + try: + currently_installed_native = pm.get_native_explicit() + currently_installed_foreign = pm.get_foreign_explicit() + orphans = pm.get_native_orphans() + to_remove = ( + (currently_installed_native | orphans) - self.packages - self.ignored_packages + ) + + actually_to_remove = set() + to_set_as_dependencies = set() + + dependants_to_keep = self.packages | currently_installed_foreign + for package in to_remove: + dependants = pm.get_dependants(package) + if any(dependant in dependants_to_keep for dependant in dependants): + to_set_as_dependencies.add(package) + else: + actually_to_remove.add(package) + + if actually_to_remove: + output.print_list("Removing pacman packages:", sorted(actually_to_remove)) + if not dry_run: + pm.remove(actually_to_remove) + + if to_set_as_dependencies: + output.print_list( + "Setting previously explicitly installed packages as dependencies:", + sorted(to_set_as_dependencies), + ) + if not dry_run: + pm.set_as_dependencies(to_set_as_dependencies) + + output.print_summary("Upgrading packages.") + if not dry_run: + pm.upgrade() + + to_install = self.packages - currently_installed_native - self.ignored_packages + output.print_list("Installing pacman packages:", sorted(to_install)) + + if not dry_run: + pm.install(to_install) + except errors.CommandFailedError as error: + output.print_error("Running a pacman command failed.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + return True + + +class PacmanCommands: + def list_explicit_native(self) -> list[str]: + """ + Running this command outputs a newline seperated list of explicitly installed native + packages. + """ + return ["pacman", "-Qeqn", "--color=never"] + + def list_explicit_foreign(self) -> list[str]: + """ + Running this command outputs a newline seperated list of explicitly installed foreign + packages. + """ + return ["pacman", "-Qeqm", "--color=never"] + + def list_orphans_native(self) -> list[str]: + """ + Running this command outputs a newline seperated list of orphaned native packages. + """ + return ["pacman", "-Qndtq", "--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 install(self, pkgs: set[str]) -> list[str]: + """ + Running this command installs the given packages from pacman repositories. + """ + return ["pacman", "-S", "--needed"] + list(pkgs) + + def upgrade(self) -> list[str]: + """ + Running this command upgrades all pacman packages from pacman repositories. + """ + return ["pacman", "-Syu"] + + def set_as_dependencies(self, pkgs: set[str]) -> list[str]: + """ + Running this command 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 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) + + +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._commands = commands + self._print_highlights = print_highlights + self._keywords = keywords + + def get_native_explicit(self) -> set[str]: + """ + Returns a set of explicitly installed native packages. + """ + + cmd = self._commands.list_explicit_native() + _, packages_text = command.check_run_result(cmd, command.run(cmd)) + packages = set(packages_text.strip().split("\n")) + + return packages + + def get_native_orphans(self) -> set[str]: + """ + Returns a set of orphaned native packages. + """ + + cmd = self._commands.list_orphans_native() + rc, packages_text = command.run(cmd) + # returncode 1 means no packages exist + if rc == 1: + return set() + if rc != 0: + raise errors.CommandFailedError(cmd, packages_text) + + packages = set(packages_text.strip().split("\n")) + + return packages + + def get_foreign_explicit(self) -> set[str]: + """ + Returns a set of explicitly installed foreign packages. + """ + cmd = self._commands.list_explicit_foreign() + rc, packages_text = command.run(cmd) + # returncode 1 means no packages exist + if rc == 1: + return set() + if rc != 0: + raise errors.CommandFailedError(cmd, packages_text) + + packages = set(packages_text.strip().split("\n")) + + return packages + + def get_dependants(self, package: str) -> set[str]: + """ + Returns a set of packages that depend on the given package. + """ + + cmd = self._commands.list_dependants(package) + _, packages_text = command.check_run_result(cmd, command.run(cmd)) + packages = set(packages_text.strip().split("\n")) + + return packages + + def set_as_dependencies(self, packages: set[str]): + """ + Marks the given packages as dependency packages. + """ + if not packages: + return + + cmd = self._commands.set_as_dependencies(packages) + command.check_run_result(cmd, command.run(cmd)) + + 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) + + _, pacman_output = command.check_run_result(cmd, command.pty_run(cmd)) + self.print_highlighted_pacman_messages(pacman_output) + + cmd = self._commands.set_as_explicit(packages) + command.check_run_result(cmd, command.run(cmd)) + + def upgrade(self): + """ + Upgrades all packages. + """ + cmd = self._commands.upgrade() + _, pacman_output = command.check_run_result(cmd, command.pty_run(cmd)) + 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) + _, pacman_output = command.check_run_result(cmd, command.pty_run(cmd)) + 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/__init__.py b/src/decman/plugins/pacman/__init__.py deleted file mode 100644 index c862814..0000000 --- a/src/decman/plugins/pacman/__init__.py +++ /dev/null @@ -1,104 +0,0 @@ -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 AUR(plugins.Plugin): - """ - Plugin that manages additional pacman packages installed outside the pacman repos. - - AUR packages are added directly to ``packages`` or declared by modules via ``@aur_packages``. - - Custom packages are added directly to ``custom_packages`` or declared by modules via - ``@custom_packages``. - """ - - NAME = "aur" - - def __init__(self) -> None: - self.packages: set[str] = set() - self.custom_packages: set[CustomPackage] = 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("aur_packages_for_module", {}) - store.ensure("custom_packages_for_module", {}) - - def apply( - self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None - ) -> bool: - return True - - -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.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/tests/test_decman_plugins_aur.py b/tests/test_decman_plugins_aur.py new file mode 100644 index 0000000..fc2e6a9 --- /dev/null +++ b/tests/test_decman_plugins_aur.py @@ -0,0 +1,302 @@ +from typing import Any + +import pytest + +from decman.plugins import aur as aur_plugin + + +class FakeStore(dict): + def ensure(self, key: str, default: Any) -> None: + if key not in self: + self[key] = default + + +class FakeModule: + def __init__(self, name: str, aur_pkgs: set[str], custom_pkgs: set[Any]) -> None: + self.name = name + self._changed = False + self._aur_pkgs = aur_pkgs + self._custom_pkgs = custom_pkgs + + +class FakeCustomPackage: + def __init__(self, pkgname: str) -> None: + self.pkgname = pkgname + + def __hash__(self) -> int: # needed because instances go into sets + return hash(self.pkgname) + + def __eq__(self, other: object) -> bool: + return isinstance(other, FakeCustomPackage) and self.pkgname == other.pkgname + + def parse(self, commands: Any) -> str: + # Whatever ForeignPackageManager expects; we just need something to feed into add_custom_pkg + return f"parsed-{self.pkgname}" + + +def test_process_modules_collects_aur_and_custom_packages_and_marks_changed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + aur = aur_plugin.AUR() + store = FakeStore() + + cp1 = FakeCustomPackage("custom1") + cp2 = FakeCustomPackage("custom2") + + mod1 = FakeModule("mod1", {"aur1", "aur2"}, {cp1}) + mod2 = FakeModule("mod2", {"aur3"}, {cp2}) + + def fake_run_method_with_attribute(mod: FakeModule, attr: str): + if attr == "__aur__packages__": + return mod._aur_pkgs + if attr == "__custom__packages__": + return mod._custom_pkgs + return None + + monkeypatch.setattr( + aur_plugin.plugins, "run_method_with_attribute", fake_run_method_with_attribute + ) + + aur.process_modules(store, {mod1, mod2}) + + # union of all aur/custom packages collected + assert aur.packages == {"aur1", "aur2", "aur3"} + assert aur.custom_packages == {cp1, cp2} + + # stored per-module + assert store["aur_packages_for_module"]["mod1"] == {"aur1", "aur2"} + assert store["aur_packages_for_module"]["mod2"] == {"aur3"} + assert store["custom_packages_for_module"]["mod1"] == {cp1} + assert store["custom_packages_for_module"]["mod2"] == {cp2} + + # first run: modules marked changed + assert mod1._changed is True + assert mod2._changed is True + + +def test_apply_respects_ignored_packages_and_protects_their_dependencies( + monkeypatch: pytest.MonkeyPatch, +) -> None: + aur = aur_plugin.AUR() + store = FakeStore() + + # Desired AUR/custom state + aur.packages = {"desired-aur"} + cp = FakeCustomPackage("custom-aur") + aur.custom_packages = {cp} + + # Ignored foreign package (installed) and an ignored but *uninstalled* package + aur.ignored_packages = {"ignored-aur", "ignored-not-installed"} + + # Fake PackageSearch + class FakePackageSearch: + def __init__(self, timeout: int) -> None: + self.timeout = timeout + self.added: list[Any] = [] + + def add_custom_pkg(self, parsed: Any) -> None: + self.added.append(parsed) + + monkeypatch.setattr(aur_plugin, "PackageSearch", FakePackageSearch) + monkeypatch.setattr(aur_plugin.os, "makedirs", lambda *x, **kw: None) + + # Fake pacman interface for foreign/native info + class FakePM: + def __init__(self, commands, print_highlights, keywords) -> None: + self.commands = commands + self.print_highlights = print_highlights + self.keywords = keywords + + self.remove_called_with: set[str] | None = None + self.set_as_deps_called_with: set[str] | None = None + + def get_native_explicit(self) -> set[str]: + # no natives needed for this scenario + return set() + + def get_foreign_explicit(self) -> set[str]: + # All explicitly installed foreign packages: + # - ignored-aur (ignored, must stay and protect deps) + # - dep-of-ignored (candidate; has ignored dependant) + # - orphan-foreign (candidate; no dependants) + return {"ignored-aur", "dep-of-ignored", "orphan-foreign"} + + def get_foreign_orphans(self) -> set[str]: + # orphan-foreign also considered orphan + return {"orphan-foreign"} + + def get_dependants(self, pkg: str) -> set[str]: + if pkg == "dep-of-ignored": + # ignored-aur depends on dep-of-ignored -> must demote, not remove + return {"ignored-aur"} + if pkg == "orphan-foreign": + return set() + return set() + + def remove(self, pkgs: set[str]) -> None: + self.remove_called_with = pkgs + + def set_as_dependencies(self, pkgs: set[str]) -> None: + self.set_as_deps_called_with = pkgs + + fake_pm = FakePM(None, None, None) + + def fake_pm_ctor(commands, print_highlights, keywords) -> FakePM: + fake_pm.commands = commands + fake_pm.print_highlights = print_highlights + fake_pm.keywords = keywords + return fake_pm + + monkeypatch.setattr(aur_plugin, "AurPacmanInterface", fake_pm_ctor) + + # Fake ForeignPackageManager + class FakeFPM: + def __init__( + self, + store_arg, + pm_arg, + package_search_arg, + commands_arg, + cache_dir, + build_dir, + makepkg_user, + ) -> None: + self.store = store_arg + self.pm = pm_arg + self.package_search = package_search_arg + self.commands = commands_arg + self.cache_dir = cache_dir + self.build_dir = build_dir + self.makepkg_user = makepkg_user + + self.upgrade_args: tuple[bool, bool, set[str]] | None = None + self.install_called_with: list[str] | None = None + + def upgrade(self, upgrade_devel: bool, force: bool, ignored: set[str]) -> None: + self.upgrade_args = (upgrade_devel, force, ignored) + + def install(self, pkgs: list[str], force: bool = False) -> None: + # store as set to ignore ordering + self.install_called_with = pkgs + + fake_fpm = FakeFPM(None, None, None, None, None, None, None) + + def fake_fpm_ctor( + store_arg, pm_arg, package_search_arg, commands_arg, cache_dir, build_dir, makepkg_user + ): + fake_fpm.store = store_arg + fake_fpm.pm = pm_arg + fake_fpm.package_search = package_search_arg + fake_fpm.commands = commands_arg + fake_fpm.cache_dir = cache_dir + fake_fpm.build_dir = build_dir + fake_fpm.makepkg_user = makepkg_user + return fake_fpm + + monkeypatch.setattr(aur_plugin, "ForeignPackageManager", fake_fpm_ctor) + + printed_lists: list[tuple[str, list[str]]] = [] + printed_summaries: list[str] = [] + + def fake_print_list(title: str, items: list[str]) -> None: + printed_lists.append((title, items)) + + def fake_print_summary(msg: str) -> None: + printed_summaries.append(msg) + + monkeypatch.setattr(aur_plugin.output, "print_list", fake_print_list) + monkeypatch.setattr(aur_plugin.output, "print_summary", fake_print_summary) + + # Use params to test flag propagation into upgrade/install + ok = aur.apply(store, dry_run=False, params=["aur-upgrade-devel", "aur-force"]) + + assert ok is True + + # Removal / demotion logic: + # + # custom_package_names = {"custom-aur"} + # currently_installed_foreign = {"ignored-aur", "dep-of-ignored", "orphan-foreign"} + # orphans = {"orphan-foreign"} + # + # to_remove candidates: + # (foreign | orphans) - desired - custom - ignored + # = {"ignored-aur", "dep-of-ignored", "orphan-foreign"} ∪ {"orphan-foreign"} + # - {"desired-aur"} - {"custom-aur"} - {"ignored-aur"} + # = {"dep-of-ignored", "orphan-foreign"} + # + # dependants_to_keep includes ignored installed foreign -> dep-of-ignored is demoted, orphan-foreign removed. + + assert fake_pm.remove_called_with == {"orphan-foreign"} + assert fake_pm.set_as_deps_called_with == {"dep-of-ignored"} + + # Ensure ignored packages were not removed + assert "ignored-aur" not in (fake_pm.remove_called_with or set()) + + # Upgrade called with flags and ignored set + assert fake_fpm.upgrade_args == (True, True, aur.ignored_packages) + + # to_install = (packages | custom_names) - installed_foreign - ignored + # = {"desired-aur", "custom-aur"} - {"ignored-aur", "dep-of-ignored", "orphan-foreign"} + # - {"ignored-aur", "ignored-not-installed"} + # = {"desired-aur", "custom-aur"} + assert set(fake_fpm.install_called_with or []) == {"desired-aur", "custom-aur"} + # ignored packages must not be installed + assert "ignored-aur" not in (fake_fpm.install_called_with or []) + assert "ignored-not-installed" not in (fake_fpm.install_called_with or []) + + # Also check the printed lists mirror this + titles = [t for t, _ in printed_lists] + assert "Removing foreign packages:" in titles + assert "Setting previously explicitly installed foreign packages as dependencies:" in titles + assert "Installing foreign packages:" in titles + + remove_list = next(items for t, items in printed_lists if "Removing foreign packages:" in t) + demote_list = next( + items + for t, items in printed_lists + if "Setting previously explicitly installed foreign packages as dependencies:" in t + ) + install_list = next(items for t, items in printed_lists if "Installing foreign packages:" in t) + + assert remove_list == ["orphan-foreign"] + assert demote_list == ["dep-of-ignored"] + # Order of install_list is deterministic because sorted() is used + assert install_list == ["custom-aur", "desired-aur"] + assert any("Upgrading foreign packages." in s for s in printed_summaries) + + +def test_apply_returns_false_on_aur_rpc_error(monkeypatch: pytest.MonkeyPatch) -> None: + aur = aur_plugin.AUR() + store = FakeStore() + + # Force PackageSearch to fail immediately + class FailingPackageSearch: + def __init__(self, timeout: int) -> None: + raise aur_plugin.AurRPCError("RPC down", "url") + + monkeypatch.setattr(aur_plugin, "PackageSearch", FailingPackageSearch) + monkeypatch.setattr(aur_plugin.os, "makedirs", lambda *x, **kw: None) + + errors_logged: list[str] = [] + continuations: list[str] = [] + traceback_called: list[bool] = [] + + def fake_print_error(msg: str) -> None: + errors_logged.append(msg) + + def fake_print_continuation(msg: str) -> None: + continuations.append(msg) + + def fake_print_traceback() -> None: + traceback_called.append(True) + + monkeypatch.setattr(aur_plugin.output, "print_error", fake_print_error) + monkeypatch.setattr(aur_plugin.output, "print_continuation", fake_print_continuation) + monkeypatch.setattr(aur_plugin.output, "print_traceback", fake_print_traceback) + + ok = aur.apply(store, dry_run=False) + + assert ok is False + assert any("AUR RPC" in msg or "fetch data from AUR RPC" in msg for msg in errors_logged) + assert any("RPC down" in msg for msg in continuations) + assert traceback_called diff --git a/tests/test_decman_plugins_pacman_package.py b/tests/test_decman_plugins_aur_package.py similarity index 99% rename from tests/test_decman_plugins_pacman_package.py rename to tests/test_decman_plugins_aur_package.py index 7dc3bed..cc1db5f 100644 --- a/tests/test_decman_plugins_pacman_package.py +++ b/tests/test_decman_plugins_aur_package.py @@ -2,9 +2,9 @@ import pathlib import pytest -from decman.plugins.pacman import package as pkg_mod -from decman.plugins.pacman.error import AurRPCError, PKGBUILDParseError -from decman.plugins.pacman.package import ( +from decman.plugins.aur import package as pkg_mod +from decman.plugins.aur.error import AurRPCError, PKGBUILDParseError +from decman.plugins.aur.package import ( CustomPackage, PackageInfo, PackageSearch, diff --git a/tests/test_decman_plugins_pacman_resolver.py b/tests/test_decman_plugins_aur_resolver.py similarity index 94% rename from tests/test_decman_plugins_pacman_resolver.py rename to tests/test_decman_plugins_aur_resolver.py index 7fb923e..64d1731 100644 --- a/tests/test_decman_plugins_pacman_resolver.py +++ b/tests/test_decman_plugins_aur_resolver.py @@ -1,7 +1,7 @@ import pytest -from decman.plugins.pacman.error import DependencyCycleError -from decman.plugins.pacman.resolver import DepGraph, ForeignPackage +from decman.plugins.aur.error import DependencyCycleError +from decman.plugins.aur.resolver import DepGraph, ForeignPackage def test_add_dependency(): diff --git a/tests/test_decman_plugins_pacman.py b/tests/test_decman_plugins_pacman.py new file mode 100644 index 0000000..d2713e1 --- /dev/null +++ b/tests/test_decman_plugins_pacman.py @@ -0,0 +1,291 @@ +from typing import Any + +import pytest + +from decman.plugins import pacman as pacman_plugin + + +class FakeStore(dict): + def ensure(self, key: str, default: Any) -> None: + if key not in self: + self[key] = default + + +class FakeModule: + def __init__(self, name: str, packages: set[str]) -> None: + self.name = name + self._changed = False + self._packages = packages + + +def test_process_modules_collects_packages_and_marks_changed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + pacman = pacman_plugin.Pacman() + store = FakeStore() + mod1 = FakeModule("mod1", {"pkg1", "pkg2"}) + mod2 = FakeModule("mod2", {"pkg3"}) + + def fake_run_method_with_attribute(mod: FakeModule, attr: str) -> set[str]: + assert attr == "__pacman__packages__" + return mod._packages + + monkeypatch.setattr( + pacman_plugin.plugins, + "run_method_with_attribute", + fake_run_method_with_attribute, + ) + + pacman.process_modules(store, {mod1, mod2}) + + # packages collected + assert pacman.packages == {"pkg1", "pkg2", "pkg3"} + # stored mapping per module + assert store["packages_for_module"]["mod1"] == {"pkg1", "pkg2"} + assert store["packages_for_module"]["mod2"] == {"pkg3"} + # modules marked changed (first run) + assert mod1._changed is True + assert mod2._changed is True + + +def test_apply_dry_run_computes_sets_and_does_not_call_pacman( + monkeypatch: pytest.MonkeyPatch, +) -> None: + pacman = pacman_plugin.Pacman() + store = FakeStore() + + # Desired state + pacman.packages = {"keep-explicit", "new-pkg"} + + # Fake PacmanInterface returned by plugin module + class FakePM: + def __init__(self, commands, print_highlights, keywords) -> None: # noqa: D401 + self.commands = commands + self.print_highlights = print_highlights + self.keywords = keywords + self.remove_called_with: set[str] | None = None + self.set_as_deps_called_with: set[str] | None = None + self.upgrade_called = False + self.install_called_with: set[str] | None = None + + def get_native_explicit(self) -> set[str]: + # keep-explicit (in desired), old-explicit (to demote/remove) + return {"keep-explicit", "old-explicit"} + + def get_foreign_explicit(self) -> set[str]: + # foreign-package protects its deps + return {"foreign-pkg"} + + def get_native_orphans(self) -> set[str]: + # orphan-explicit is also candidate + return {"orphan-explicit"} + + def get_dependants(self, pkg: str) -> set[str]: + # old-explicit has a foreign dependant -> demote to dep + # orphan-explicit has no dependants -> remove + if pkg == "old-explicit": + return {"foreign-pkg"} + if pkg == "orphan-explicit": + return set() + return set() + + def remove(self, pkgs: set[str]) -> None: + self.remove_called_with = pkgs + + def set_as_dependencies(self, pkgs: set[str]) -> None: + self.set_as_deps_called_with = pkgs + + def upgrade(self) -> None: + self.upgrade_called = True + + def install(self, pkgs: set[str]) -> None: + self.install_called_with = pkgs + + fake_pm = FakePM(None, None, None) + + def fake_pm_ctor(commands, print_highlights, keywords) -> FakePM: + # constructor used in Pacman.apply + fake_pm.commands = commands + fake_pm.print_highlights = print_highlights + fake_pm.keywords = keywords + return fake_pm + + monkeypatch.setattr(pacman_plugin, "PacmanInterface", fake_pm_ctor) + + printed_lists: list[tuple[str, list[str]]] = [] + printed_summaries: list[str] = [] + + def fake_print_list(title: str, items: list[str]) -> None: + printed_lists.append((title, items)) + + def fake_print_summary(msg: str) -> None: + printed_summaries.append(msg) + + monkeypatch.setattr(pacman_plugin.output, "print_list", fake_print_list) + monkeypatch.setattr(pacman_plugin.output, "print_summary", fake_print_summary) + + ok = pacman.apply(store, dry_run=True) + + assert ok is True + + # to_remove = (native | orphans) - desired + # = {keep-explicit, old-explicit} ∪ {orphan-explicit} - {keep-explicit, new-pkg} + # = {old-explicit, orphan-explicit} + # + # old-explicit has foreign dependant -> demoted to dep + # orphan-explicit has no dependants -> removed + + # printed lists (titles and contents) + titles = [t for t, _ in printed_lists] + assert "Removing pacman packages:" in titles + assert "Setting previously explicitly installed packages as dependencies:" in titles + assert "Installing pacman packages:" in titles + + # find lists by title + remove_list = next(items for t, items in printed_lists if "Removing pacman packages:" in t) + demote_list = next( + items + for t, items in printed_lists + if "Setting previously explicitly installed packages as dependencies:" in t + ) + install_list = next(items for t, items in printed_lists if "Installing pacman packages:" in t) + + assert remove_list == ["orphan-explicit"] + assert demote_list == ["old-explicit"] + # to_install = desired - currently_installed_native + # = {keep-explicit, new-pkg} - {keep-explicit, old-explicit} + # = {new-pkg} + assert install_list == ["new-pkg"] + + # Upgrade summary printed even in dry-run + assert any("Upgrading packages." in s for s in printed_summaries) + + # No mutating calls in dry-run + assert fake_pm.remove_called_with is None + assert fake_pm.set_as_deps_called_with is None + assert fake_pm.upgrade_called is False + assert fake_pm.install_called_with is None + + +def test_apply_returns_false_on_command_failure(monkeypatch: pytest.MonkeyPatch) -> None: + pacman = pacman_plugin.Pacman() + store = FakeStore() + pacman.packages = set() + + class FailingPM: + def __init__(self, *args, **kwargs) -> None: # noqa: D401 + pass + + def get_native_explicit(self) -> set[str]: + raise pacman_plugin.errors.CommandFailedError(["get_native_explicit"], "boom") + + monkeypatch.setattr(pacman_plugin, "PacmanInterface", FailingPM) + + errors_logged: list[str] = [] + continuations: list[str] = [] + traceback_called = [] + + def fake_print_error(msg: str) -> None: + errors_logged.append(msg) + + def fake_print_continuation(msg: str) -> None: + continuations.append(msg) + + def fake_print_traceback() -> None: + traceback_called.append(True) + + monkeypatch.setattr(pacman_plugin.output, "print_error", fake_print_error) + monkeypatch.setattr(pacman_plugin.output, "print_continuation", fake_print_continuation) + monkeypatch.setattr(pacman_plugin.output, "print_traceback", fake_print_traceback) + + ok = pacman.apply(store, dry_run=False) + + assert ok is False + assert any("pacman command failed" in msg for msg in errors_logged) + assert any("boom" in msg for msg in continuations) + assert traceback_called # at least once + + +def test_ignored_packages_are_not_removed_or_installed(monkeypatch: pytest.MonkeyPatch) -> None: + pacman = pacman_plugin.Pacman() + store = FakeStore() + + # Desired state: "already" and "new" should be managed normally. + # "ignored-installed" is currently installed but not desired -> would normally be removed. + # "ignored-uninstalled" is desired but not installed -> would normally be installed. + pacman.packages = {"already", "new", "ignored-uninstalled"} + pacman.ignored_packages = {"ignored-installed", "ignored-uninstalled"} + + class FakePM: + def __init__(self, commands, print_highlights, keywords) -> None: + self.commands = commands + self.print_highlights = print_highlights + self.keywords = keywords + + self.remove_called_with: set[str] | None = None + self.install_called_with: set[str] | None = None + self.set_as_deps_called_with: set[str] | None = None + self.upgrade_called = False + + def get_native_explicit(self) -> set[str]: + # currently installed explicit packages + return {"ignored-installed", "already"} + + def get_foreign_explicit(self) -> set[str]: + return set() + + def get_native_orphans(self) -> set[str]: + return set() + + def get_dependants(self, pkg: str) -> set[str]: + return set() + + def remove(self, pkgs: set[str]) -> None: + self.remove_called_with = pkgs + + def set_as_dependencies(self, pkgs: set[str]) -> None: + self.set_as_deps_called_with = pkgs + + def upgrade(self) -> None: + self.upgrade_called = True + + def install(self, pkgs: set[str]) -> None: + self.install_called_with = pkgs + + fake_pm = FakePM(None, None, None) + + def fake_pm_ctor(commands, print_highlights, keywords) -> FakePM: + fake_pm.commands = commands + fake_pm.print_highlights = print_highlights + fake_pm.keywords = keywords + return fake_pm + + monkeypatch.setattr(pacman_plugin, "PacmanInterface", fake_pm_ctor) + + printed_lists: list[tuple[str, list[str]]] = [] + + def fake_print_list(title: str, items: list[str]) -> None: + printed_lists.append((title, items)) + + # don't care about summaries here + monkeypatch.setattr(pacman_plugin.output, "print_list", fake_print_list) + monkeypatch.setattr(pacman_plugin.output, "print_summary", lambda *_args, **_kw: None) + + ok = pacman.apply(store, dry_run=False) + + assert ok is True + + # Ignored packages must never be passed to remove() or install() + assert ( + fake_pm.remove_called_with is None or "ignored-installed" not in fake_pm.remove_called_with + ) + assert fake_pm.install_called_with is not None + assert "ignored-uninstalled" not in fake_pm.install_called_with + + # Also ensure the printed install list doesn't contain ignored packages + install_items = next( + items for title, items in printed_lists if "Installing pacman packages:" in title + ) + assert "ignored-uninstalled" not in install_items + # "new" is the only package that should be installed in this scenario + assert install_items == ["new"]