Rename packages to plugins

This commit is contained in:
Kivi Kaitaniemi
2025-12-27 04:04:04 +02:00
parent 9b2fc79829
commit 041c976552
20 changed files with 12 additions and 12 deletions
+20
View File
@@ -0,0 +1,20 @@
[project]
name = "decman-flatpak"
version = "1.0.0"
requires-python = ">=3.13"
dependencies = ["decman==1.0.0"]
[project.entry-points."decman.plugins"]
flatpak = "decman.plugins.flatpak:Flatpak"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
include = ["decman.plugins*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
@@ -0,0 +1,273 @@
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 flatpak package names that should be installed.
Return type of ``fn``: ``set[str]``
"""
fn.__flatpak__packages__ = True
return fn
def user_packages(fn):
"""
Annotate that this function returns a dict of users and flatpak packages that should be
installed.
Return type of ``fn``: ``dict[str, set[str]]``
"""
fn.__flatpak__user__packages__ = True
return fn
class Flatpak(plugins.Plugin):
"""
Plugin that manages flatpak packages added directly to ``packages`` or declared by modules via
``@flatpak.packages``. User packages are managed as well.
"""
NAME = "flatpak"
def __init__(self) -> None:
self.packages: set[str] = set()
self.user_packages: dict[str, set[str]] = {}
self.ignored_packages: set[str] = set()
self.commands = FlatpakCommands()
def available(self) -> bool:
return shutil.which("flatpak") is not None
def process_modules(self, store: _store.Store, modules: set[module.Module]):
# These store keys are used to track changes in modules.
# This way when these change, module can be marked as changed
store.ensure("flatpaks_for_module", {})
store.ensure("user_flatpaks_for_module", {})
for mod in modules:
store["flatpaks_for_module"].setdefault(mod.name, set())
store["user_flatpaks_for_module"].setdefault(mod.name, {})
packages = plugins.run_method_with_attribute(mod, "__flatpak__packages__") or set()
user_packages = (
plugins.run_method_with_attribute(mod, "__flatpak__user__packages__") or {}
)
if store["flatpaks_for_module"][mod.name] != packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified system flatpaks."
)
if store["user_flatpaks_for_module"][mod.name] != user_packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified user flatpaks."
)
self.packages |= packages
for user, flatpaks in user_packages.items():
self.user_packages.setdefault(user, set()).update(flatpaks)
store["flatpaks_for_module"][mod.name] = packages
store["user_flatpaks_for_module"][mod.name] = user_packages
def apply(
self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None
) -> bool:
pm = FlatpakInterface(self.commands)
try:
self.apply_packages(pm, None, self.packages, self.ignored_packages, dry_run)
for user, packages in self.user_packages.items():
self.apply_packages(pm, user, packages, self.ignored_packages, dry_run)
except errors.CommandFailedError as error:
output.print_error("Running a flatpak command failed.")
output.print_error(
"Flatpak command exited with an unexpected return code. You may have cancelled a "
"flatpak operation."
)
output.print_error(str(error))
if error.output:
output.print_command_output(error.output)
output.print_traceback()
return False
return True
def apply_packages(
self,
flatpak: "FlatpakInterface",
user: str | None,
packages: set[str],
ignored_packages: set[str],
dry_run: bool,
):
currently_installed = flatpak.get_apps(user)
to_remove = currently_installed - packages - ignored_packages
to_install = packages - currently_installed - ignored_packages
for_user_msg = f" for {user}" if user else ""
if to_remove:
output.print_list(f"Removing flatpak packages{for_user_msg}:", sorted(to_remove))
if not dry_run:
flatpak.remove(to_remove, user)
output.print_summary(f"Upgrading packages{for_user_msg}.")
if not dry_run:
flatpak.upgrade(user)
if to_install:
output.print_list(f"Installing flatpak packages{for_user_msg}:", sorted(to_install))
if not dry_run:
flatpak.install(to_install, user)
class FlatpakCommands:
def list_apps(self, as_user: bool) -> list[str]:
"""
Running this command outputs a newline separated list of installed flatpak application IDs.
If ``as_user`` is ``True``, run the command as the user whose packages should be listed.
NOTE: The first line says 'Application ID' and should be ignored.
"""
return [
"flatpak",
"list",
"--app",
"--user" if as_user else "--system",
"--columns",
"application",
]
def install(self, pkgs: set[str], as_user: bool) -> list[str]:
"""
Running this command installs all listed packages, and their dependencies/runtimes
automatically.
If ``as_user`` is ``True``, run the command as the user for whom packages are installed.
"""
return [
"flatpak",
"install",
"--user" if as_user else "--system",
] + sorted(pkgs)
def upgrade(self, as_user: bool) -> list[str]:
"""
Updates all installed flatpaks including runtimes and dependencies.
If ``as_user`` is ``True``, run the command as the user whose flatpaks are updated.
"""
return [
"flatpak",
"update",
"--user" if as_user else "--system",
]
def remove(self, pkgs: set[str], as_user: bool) -> list[str]:
"""
Running this command will remove the listed packages.
If ``as_user`` is ``True``, run the command as the user for whom packages are removed.
"""
return [
"flatpak",
"remove",
"--user" if as_user else "--system",
] + sorted(pkgs)
def remove_unused(self, as_user: bool) -> list[str]:
"""
This will remove all unused flatpak dependencies and runtimes.
If ``as_user`` is ``True``, run the command as the user for whom packages are removed.
"""
return [
"flatpak",
"remove",
"--unused",
"--user" if as_user else "--system",
]
class FlatpakInterface:
"""
High level interface for running pacman commands.
On failure methods raise a ``CommandFailedError``.
"""
def __init__(self, commands: FlatpakCommands) -> None:
self._commands = commands
def get_apps(self, user: str | None = None) -> set[str]:
"""
Returns a set of installed flatpak apps.
If ``user`` is set, returns flatpak apps for that user.
"""
as_user = user is not None
cmd = self._commands.list_apps(as_user=as_user)
_, packages_text = command.check_run_result(
cmd, command.run(cmd, user=user, mimic_login=as_user)
)
packages = packages_text.strip().split("\n")
# In case no apps are installed, the list contains this
if "" in packages:
packages.remove("")
return set(packages)
def install(self, packages: set[str], user: str | None = None):
"""
Installs the given packages.
If ``user`` is set, installs packages for that user.
"""
if not packages:
return
as_user = user is not None
cmd = self._commands.install(packages, as_user)
command.prg(cmd, user=user, mimic_login=as_user)
def upgrade(self, user: str | None = None):
"""
Upgrades all packages.
If ``user`` is set, upgrades packages for that user.
"""
as_user = user is not None
cmd = self._commands.upgrade(as_user)
command.prg(cmd, user=user, mimic_login=as_user)
def remove(self, packages: set[str], user: str | None = None):
"""
Removes the given packages as well as unused dependencies.
If ``user`` is set, removes packages for that user.
"""
if not packages:
return
as_user = user is not None
cmd = self._commands.remove(packages, as_user)
command.prg(cmd, user=user, mimic_login=as_user)
cmd = self._commands.remove_unused(as_user)
command.prg(cmd, user=user, mimic_login=as_user)
+25
View File
@@ -0,0 +1,25 @@
[project]
name = "decman-pacman"
version = "1.0.0"
requires-python = ">=3.13"
dependencies = [
"decman==1.0.0",
"pyalpm",
"requests",
]
[project.entry-points."decman.plugins"]
pacman = "decman.plugins.pacman:Pacman"
aur = "decman.plugins.aur:AUR"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
include = ["decman.plugins*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
@@ -0,0 +1,258 @@
import os
import shutil
import pyalpm
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
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
# Re-exports
__all__ = [
"AUR",
"AurCommands",
"CustomPackage",
"packages",
"custom_packages",
]
def 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
``@aur.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.database_signature_level = pyalpm.SIG_DATABASE_OPTIONAL
self.database_path = "/var/lib/pacman/"
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()
)
custom_package_strs = set(map(str, custom_packages))
if store["aur_packages_for_module"][mod.name] != aur_packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified aur packages."
)
if store["custom_packages_for_module"][mod.name] != custom_package_strs:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified custom packages."
)
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_package_strs
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: "
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,
self.database_signature_level,
self.database_path,
)
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_error(str(error))
output.print_traceback()
return False
except DependencyCycleError as error:
output.print_error("Foreign package dependency cycle detected.")
output.print_error(str(error))
output.print_traceback()
return False
except PKGBUILDParseError as error:
output.print_error("Failed to parse a CustomPackage PKGBUILD.")
output.print_error(str(error))
output.print_traceback()
return False
except ForeignPackageManagerError as error:
output.print_error("Foreign package manager failed.")
output.print_error(str(error))
output.print_traceback()
return False
except pyalpm.error as error:
output.print_error("Failed to query pacman databases with pyalpm.")
output.print_error(str(error))
output.print_traceback()
return False
except errors.CommandFailedError as error:
output.print_error(
"AUR command exited with an unexpected return code. You may have cancelled a "
"pacman operation."
)
output.print_error(str(error))
if error.output:
output.print_command_output(error.output)
output.print_traceback()
return False
return True
@@ -0,0 +1,199 @@
import decman.plugins.pacman as pacman
import pyalpm
import decman.config as config
import decman.core.command as command
class AurCommands(pacman.PacmanCommands):
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 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
def print_srcinfo(self) -> list[str]:
"""
Running this command prints SRCINFO generated from the package in the current
working directory.
"""
return ["makepkg", "--printsrcinfo"]
class AurPacmanInterface(pacman.PacmanInterface):
"""
High level interface for running pacman commands.
On failure methods raise a ``CommandFailedError``.
"""
def __init__(
self,
commands: AurCommands,
print_highlights: bool,
keywords: set[str],
dbsiglevel: int,
dbpath: str,
) -> None:
super().__init__(commands, print_highlights, keywords, dbsiglevel, dbpath)
self._installable: dict[str, bool] = {}
self._aur_commands = commands
def get_foreign_orphans(self) -> set[str]:
"""
Returns a set of orphaned foreign packages.
"""
out: set[str] = set()
for pkg in self._handle.get_localdb().pkgcache:
if pkg.reason != pyalpm.PKG_REASON_DEPEND:
continue
if pkg.compute_requiredby():
continue
if not self._is_native(pkg.name):
out.add(pkg.name)
return out
def is_installable(self, pkg: str) -> bool:
"""
Returns True if a package can be installed using pacman.
"""
return pkg in self._name_index or pacman.strip_dependency(pkg) in self._provides_index
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.
"""
out: list[tuple[str, str]] = []
for pkg in self._handle.get_localdb().pkgcache:
if not self._is_native(pkg.name):
out.append((pkg.name, pkg.version))
return out
def install_dependencies(self, deps: set[str]):
"""
Installs the given dependencies.
"""
if not deps:
return
cmd = self._aur_commands.install_as_dependencies(deps)
pacman_output = command.prg(cmd)
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._aur_commands.install_files_as_dependencies(files)
pacman_output = command.prg(cmd)
self.print_highlighted_pacman_messages(pacman_output)
if not as_explicit:
return
cmd = self._commands.set_as_explicit(as_explicit)
command.prg(cmd, pty=config.debug_output)
@@ -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}' "
f"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}")
@@ -0,0 +1,814 @@
import os
import shutil
import time
import typing
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
import decman.config as config
import decman.core.command as command
import decman.core.error as errors
import decman.core.output as output
import decman.core.store as _store
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: "
f"{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(e.strerror or str(e))
output.print_error("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: AurPacmanInterface,
search: PackageSearch,
commands: AurCommands,
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_info("Determining foreign packages to upgrade.")
all_foreign_pkgs = self._pacman.get_versioned_foreign_packages()
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)))
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_foreign_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:",
sorted(resolved_dependencies.foreign_pkgs),
)
output.print_list(
"The following foreign packages will be installed as dependencies:",
sorted(resolved_dependencies.foreign_dep_pkgs),
)
output.print_list(
"The following foreign packages will be built in order to install other packages. "
"They will not be installed:",
sorted(resolved_dependencies.foreign_build_dep_pkgs),
)
if not output.prompt_confirm("Proceed?", default=True):
raise ForeignPackageManagerError("Installing aborted by the user.")
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)
vercmp_output = command.prg(cmd, pty=False)
should_upgrade = int(vercmp_output) < 0
output.print_debug(
f"Installed version is: {installed_version}. "
f"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: AurPacmanInterface,
resolved_deps: ResolvedDependencies,
commands: AurCommands,
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.prg(
cmd, env_overrides=mkarchroot_env_vars, pass_environment=False, pty=config.debug_output
)
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.prg(cmd, pty=config.debug_output)
output.print_info("Making package.")
cmd = self._commands.make_chroot_pkg(
self.chroot_wd_dir, self.makepkg_user, chroot_pkg_files
)
command.prg(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.prg(cmd, pty=config.debug_output)
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. "
f"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, ".")
command.prg(cmd, pty=config.debug_output)
if pkgbuild_directory:
try:
shutil.copytree(pkgbuild_directory, ".", dirs_exist_ok=True)
# Chmod to 755 to allow reading files
mode = 0o755
for root, dirs, files in os.walk("."):
for name in dirs + files:
os.chmod(os.path.join(root, name), mode)
os.chmod(".", mode)
except OSError as error:
raise ForeignPackageManagerError(f"Failed to copy {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()
git_output = command.prg(cmd, pty=False)
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)
command.prg(cmd)
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)
command.prg(cmd)
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:
commit_id = git_output.strip()
self._store["pkgbuild_latest_reviewed_commits"][pkgbase] = commit_id
else:
output.print_debug(
f"{pkgbase} is not in a git repository. Commit ID cannot be saved."
)
else:
raise ForeignPackageManagerError("Building aborted.")
@@ -0,0 +1,714 @@
import dataclasses
import os
import pathlib
import shutil
import tempfile
import decman.plugins.pacman as pacman_module
import requests # type: ignore
from decman.plugins.aur.commands import AurCommands, AurPacmanInterface
from decman.plugins.aur.error import AurRPCError, PKGBUILDParseError
import decman.config as config
import decman.core.command as command
import decman.core.error as errors
import decman.core.output as output
@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)
# Caches (excluded from eq/hash)
_native_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
_foreign_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
_native_make_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
_foreign_make_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
_native_check_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
_foreign_check_dependencies: tuple[str, ...] | None = dataclasses.field(
default=None, init=False, repr=False, compare=False
)
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}"
# --- public API ---------------------------------------------------------
def foreign_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of foreign dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_dependencies_cached(pacman)
assert self._foreign_dependencies is not None
return list(self._foreign_dependencies)
def foreign_make_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of foreign make dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_make_dependencies_cached(pacman)
assert self._foreign_make_dependencies is not None
return list(self._foreign_make_dependencies)
def foreign_check_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of foreign check dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_check_dependencies_cached(pacman)
assert self._foreign_check_dependencies is not None
return list(self._foreign_check_dependencies)
def native_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of native dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_dependencies_cached(pacman)
assert self._native_dependencies is not None
return list(self._native_dependencies)
def native_make_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of native make dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_make_dependencies_cached(pacman)
assert self._native_make_dependencies is not None
return list(self._native_make_dependencies)
def native_check_dependencies(self, pacman: AurPacmanInterface) -> list[str]:
"""
Returns a list of native check dependencies of this package.
The dependencies are stripped of their version constraints if there are any.
"""
self._ensure_check_dependencies_cached(pacman)
assert self._native_check_dependencies is not None
return list(self._native_check_dependencies)
# --- internal helpers ---------------------------------------------------
@staticmethod
def _classify_dependencies(
deps: tuple[str, ...], pacman: AurPacmanInterface
) -> tuple[tuple[str, ...], tuple[str, ...]]:
native: list[str] = []
foreign: list[str] = []
for dependency in deps:
stripped = pacman_module.strip_dependency(dependency)
if pacman.is_installable(dependency):
native.append(stripped)
else:
foreign.append(stripped)
return tuple(native), tuple(foreign)
def _ensure_dependencies_cached(self, pacman: AurPacmanInterface) -> None:
if self._native_dependencies is not None:
return
native, foreign = self._classify_dependencies(self.dependencies, pacman)
object.__setattr__(self, "_native_dependencies", native)
object.__setattr__(self, "_foreign_dependencies", foreign)
def _ensure_make_dependencies_cached(self, pacman: AurPacmanInterface) -> None:
if self._native_make_dependencies is not None:
return
native, foreign = self._classify_dependencies(self.make_dependencies, pacman)
object.__setattr__(self, "_native_make_dependencies", native)
object.__setattr__(self, "_foreign_make_dependencies", foreign)
def _ensure_check_dependencies_cached(self, pacman: AurPacmanInterface) -> None:
if self._native_check_dependencies is not None:
return
native, foreign = self._classify_dependencies(self.check_dependencies, pacman)
object.__setattr__(self, "_native_check_dependencies", native)
object.__setattr__(self, "_foreign_check_dependencies", foreign)
class CustomPackage:
"""
Custom package installed from some other location than the official repos or the AUR.
``pkgname`` is required because the PKGBUILD might be for split packages.
Exactly one of ``git_url`` or ``pkgbuild_directory`` must be provided.
Parameters:
``pkgname``:
Name of the package.
``git_url``:
URL to a git repository containing the PKGBUILD.
``pkgbuild_directory``:
Path to the directory containing the PKGBUILD.
"""
def __init__(
self, pkgname: str, git_url: str | None = None, pkgbuild_directory: str | None = 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.pkgname = pkgname
self.git_url = git_url
self.pkgbuild_directory = pkgbuild_directory
def parse(self, commands: AurCommands) -> PackageInfo:
"""
Parses this package's PKGBUILD to ``PackageInfo``.
If this fails, raises a ``PKGBUILDParseError``.
"""
if self.pkgbuild_directory is not None:
srcinfo = self._srcinfo_from_pkgbuild_directory(commands)
else:
srcinfo = self._srcinfo_from_git(commands)
return self._parse_srcinfo(srcinfo)
def __eq__(self, other: object) -> bool:
if not isinstance(other, CustomPackage):
return False
return (
self.git_url == other.git_url
and self.pkgbuild_directory == other.pkgbuild_directory
and self.pkgname == other.pkgname
)
def __hash__(self) -> int:
return hash((self.pkgname, self.git_url, self.pkgbuild_directory))
def __str__(self) -> str:
if self.git_url is not None:
return f"CustomPackage(pkgname={self.pkgname}, git_url={self.git_url})"
return (
f"CustomPackage(pkgname={self.pkgname}, pkgbuild_directory={self.pkgbuild_directory})"
)
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."
)
path = pathlib.Path(self.pkgbuild_directory)
if not path.is_dir():
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
f"pkgbuild_directory '{path}' does not exist or is not a directory.",
)
if not (path / "PKGBUILD").exists():
raise PKGBUILDParseError(
self.git_url, self.pkgbuild_directory, f"No PKGBUILD found in '{path}'."
)
try:
with tempfile.TemporaryDirectory(prefix="decman-pkgbuild-") as tmpdir:
tmp_path = pathlib.Path(tmpdir)
# Allow the user 'nobody' to use this directory
os.chmod(tmpdir, 0o777)
shutil.copy(path / "PKGBUILD", tmp_path / "PKGBUILD")
os.chmod(tmp_path / "PKGBUILD", 0o644)
return self._run_makepkg_printsrcinfo(tmp_path, commands)
except OSError as error:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
"Failed to create temporary directory for the PKGBUILD.",
) from error
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."
try:
with tempfile.TemporaryDirectory(prefix="decman-pkgbuild-") as tmpdir:
tmp_path = pathlib.Path(tmpdir)
# Allow the user 'nobody' to use this directory
os.chmod(tmpdir, 0o777)
try:
cmd = commands.git_clone(self.git_url, tmpdir)
# Use the user nobody, since that will be used later to generate SRCINFO
command.prg(cmd, user="nobody", pty=config.debug_output)
except errors.CommandFailedError as error:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
"Failed to clone PKGBUILD repository.",
) from error
if not (tmp_path / "PKGBUILD").exists():
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
f"Cloned repository '{self.git_url}' does not contain a PKGBUILD.",
)
return self._run_makepkg_printsrcinfo(tmp_path, commands)
except OSError as error:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
"Failed to create temporary directory for the PKGBUILD.",
) from error
def _run_makepkg_printsrcinfo(self, path: pathlib.Path, commands: AurCommands) -> str:
orig_wd = os.getcwd()
try:
os.chdir(path)
cmd = commands.print_srcinfo()
# No need to use the makepkg_user config option here.
# For just printing the SRCINFO, hardcoded 'nobody' works
srcinfo = command.prg(cmd, user="nobody", pty=False)
except errors.CommandFailedError as error:
raise PKGBUILDParseError(
self.git_url, self.pkgbuild_directory, "Failed to generate SRCINFO using makepkg."
) from error
finally:
os.chdir(orig_wd)
return srcinfo
def _parse_srcinfo(self, srcinfo: str) -> PackageInfo:
pkgbase: str | None = None
pkgver: str | None = None
pkgrel: str | None = None
epoch: str | None = None
provides: list[str] = []
# I'm not sure if split packages can have dependencies listed in the base.
# Easy to handle regardless
base_depends: list[str] = []
base_makedepends: list[str] = []
base_checkdepends: list[str] = []
pkg_depends: list[str] = []
pkg_makedepends: list[str] = []
pkg_checkdepends: list[str] = []
current_pkg: str | None = None
found_pkgnames = set()
for raw in srcinfo.splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = (part.strip() for part in line.split("=", 1))
is_base = current_pkg is None
is_target_pkg = current_pkg == self.pkgname
match key:
case "pkgbase":
pkgbase = value
current_pkg = None
case "pkgname":
current_pkg = value
found_pkgnames.add(value)
case "pkgver":
if pkgver is None or current_pkg == self.pkgname:
pkgver = value
case "pkgrel":
if pkgrel is None or current_pkg == self.pkgname:
pkgrel = value
case "epoch":
if epoch is None or current_pkg == self.pkgname:
epoch = value
case "provides":
if is_target_pkg:
provides.append(value)
case "depends":
if is_base:
base_depends.append(value)
elif is_target_pkg:
pkg_depends.append(value)
case "makedepends":
if is_base:
base_makedepends.append(value)
elif is_target_pkg:
pkg_makedepends.append(value)
case "checkdepends":
if is_base:
base_checkdepends.append(value)
elif is_target_pkg:
pkg_checkdepends.append(value)
case _ if key.startswith("depends") and key.removeprefix("depends_") == config.arch:
if is_base:
base_depends.append(value)
elif is_target_pkg:
pkg_depends.append(value)
case _ if (
key.startswith("makedepends")
and key.removeprefix("makedepends_") == config.arch
):
if is_base:
base_makedepends.append(value)
elif is_target_pkg:
pkg_makedepends.append(value)
case _ if (
key.startswith("checkdepends")
and key.removeprefix("checkdepends_") == config.arch
):
if is_base:
base_checkdepends.append(value)
elif is_target_pkg:
pkg_checkdepends.append(value)
if pkgbase is None or pkgver is None:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
"Missing required fields (pkgbase/pkgver) in SRCINFO.",
)
if self.pkgname not in found_pkgnames:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
f"Package {self.pkgname} not found in SRCINFO. "
f"Packages present: {' '.join(found_pkgnames)}.",
)
version_core = pkgver
if pkgrel is not None:
version_core = f"{version_core}-{pkgrel}"
if epoch is not None:
version = f"{epoch}:{version_core}"
else:
version = version_core
return PackageInfo(
pkgname=self.pkgname,
pkgbase=pkgbase,
version=version,
git_url=self.git_url,
pkgbuild_directory=self.pkgbuild_directory,
provides=tuple(provides),
dependencies=tuple(base_depends + pkg_depends),
make_dependencies=tuple(base_makedepends + pkg_makedepends),
check_dependencies=tuple(base_checkdepends + pkg_checkdepends),
)
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
@@ -0,0 +1,119 @@
import typing
from decman.plugins.aur.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
@@ -0,0 +1,379 @@
import re
import shutil
import pyalpm
import decman.config as config
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
def strip_dependency(dep: str) -> str:
"""
Removes version spefications from a dependency name.
"""
rx = re.compile("(=.*|>.*|<.*)")
return rx.sub("", dep)
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",
}
self.database_signature_level = pyalpm.SIG_DATABASE_OPTIONAL
self.database_path = "/var/lib/pacman/"
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
output.print_debug(
f"Module '{mod.name}' set to changed due to modified pacman packages."
)
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:
try:
pm = PacmanInterface(
self.commands,
self.print_highlights,
self.keywords,
self.database_signature_level,
self.database_path,
)
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 pyalpm.error as error:
output.print_error("Failed to query pacman databases with pyalpm.")
output.print_error(str(error))
output.print_traceback()
return False
except errors.CommandFailedError as error:
output.print_error(
"Pacman command exited with an unexpected return code. You may have cancelled a "
"pacman operation."
)
output.print_error(str(error))
if error.output:
output.print_command_output(error.output)
output.print_traceback()
return False
return True
class PacmanCommands:
def list_pacman_repos(self) -> list[str]:
"""
Running this command prints a newline seperated list of pacman repositories.
"""
return ["pacman-conf", "--repo-list"]
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`` or ``pyalpm.error``.
"""
def __init__(
self,
commands: PacmanCommands,
print_highlights: bool,
keywords: set[str],
dbsiglevel: int,
dbpath: str,
) -> None:
self._commands = commands
self._print_highlights = print_highlights
self._keywords = keywords
self._dbsiglevel = dbsiglevel
self._dbpath = dbpath
self._handle = self._create_pyalpm_handle()
self._name_index = self._create_name_index()
self._provides_index = self._create_provides_index()
def _create_pyalpm_handle(self):
root = "/"
h = pyalpm.Handle(root, self._dbpath)
cmd = self._commands.list_pacman_repos()
repos = command.prg(cmd, pty=False).strip().split("\n")
# Empty string means no DBs
if "" in repos and len(repos) == 1:
return
for repo in repos:
h.register_syncdb(repo, self._dbsiglevel)
return h
def _create_name_index(self) -> set[str]:
return {pkg.name for db in self._handle.get_syncdbs() for pkg in db.pkgcache}
def _create_provides_index(self) -> dict[str, set[str]]:
out: dict[str, set[str]] = {}
for db in self._handle.get_syncdbs():
for pkg in db.pkgcache:
for p in pkg.provides:
out.setdefault(strip_dependency(p), set()).add(pkg.name)
return out
def _is_native(self, package: str) -> bool:
return package in self._name_index
def get_native_explicit(self) -> set[str]:
"""
Returns a set of explicitly installed native packages.
"""
out: set[str] = set()
for pkg in self._handle.get_localdb().pkgcache:
if pkg.reason == pyalpm.PKG_REASON_EXPLICIT and self._is_native(pkg.name):
out.add(pkg.name)
return out
return packages
def get_native_orphans(self) -> set[str]:
"""
Returns a set of orphaned native packages.
"""
out: set[str] = set()
for pkg in self._handle.get_localdb().pkgcache:
if pkg.reason != pyalpm.PKG_REASON_DEPEND:
continue
if pkg.compute_requiredby():
continue
if self._is_native(pkg.name):
out.add(pkg.name)
return out
def get_foreign_explicit(self) -> set[str]:
"""
Returns a set of explicitly installed foreign packages.
"""
out: set[str] = set()
for pkg in self._handle.get_localdb().pkgcache:
if pkg.reason == pyalpm.PKG_REASON_EXPLICIT and not self._is_native(pkg.name):
out.add(pkg.name)
return out
def get_dependants(self, package: str) -> set[str]:
"""
Returns a set of installed packages that depend on the given package.
Includes the package itself.
"""
local = self._handle.get_localdb()
seen: set[str] = set()
stack = [package]
while stack:
name = stack.pop()
if name in seen:
continue
seen.add(name)
pkg = local.get_pkg(name)
if pkg is None:
continue
for dep in pkg.compute_requiredby():
if dep not in seen:
stack.append(dep)
return seen
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.prg(cmd, pty=config.debug_output)
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.prg(cmd)
self.print_highlighted_pacman_messages(pacman_output)
cmd = self._commands.set_as_explicit(packages)
command.prg(cmd, pty=config.debug_output)
def upgrade(self):
"""
Upgrades all packages.
"""
cmd = self._commands.upgrade()
pacman_output = command.prg(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.prg(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
lines = pacman_output.split("\n")
highlight_lines = []
for index, line in enumerate(lines):
for keyword in self._keywords:
if keyword.lower() in line.lower():
highlight_lines.append(f"lines: {index}-{index + 2}")
if index >= 1:
highlight_lines.append(lines[index - 1])
highlight_lines.append(line)
if index + 1 < len(lines):
highlight_lines.append(lines[index + 1])
highlight_lines.append("")
# Break, as to not print the same line again if it contains multiple keywords
break
if highlight_lines:
output.print_summary("Pacman output highlights:")
for line in highlight_lines:
if line.startswith("lines:"):
output.print_summary(line)
else:
output.print_continuation(line)
@@ -0,0 +1,310 @@
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"] == {str(cp1)}
assert store["custom_packages_for_module"]["mod2"] == {str(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, dbsiglevel, dbpath) -> 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, None, None)
def fake_pm_ctor(
commands,
print_highlights,
keywords,
dbsiglevel,
dbpath,
) -> 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_traceback() -> None:
traceback_called.append(True)
monkeypatch.setattr(aur_plugin.output, "print_error", fake_print_error)
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 errors_logged)
assert traceback_called
@@ -0,0 +1,743 @@
import pathlib
import pytest
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,
)
@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
)
# --- 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("pkg", git_url=None, pkgbuild_directory=None)
with pytest.raises(ValueError, match="cannot be set"):
CustomPackage("pkg", git_url="git://example", pkgbuild_directory="/tmp")
class DummyCommands:
"""Minimal stub; only here so type checks pass where needed."""
pass
@pytest.mark.parametrize(
"srcinfo, expected_version",
[
(
"""
pkgbase = foo
pkgver = 1.2.3
pkgrel = 4
pkgname = foo
""",
"1.2.3-4",
),
(
"""
pkgbase = foo
pkgver = 1.2.3
pkgrel = 4
epoch = 2
pkgname = foo
""",
"2:1.2.3-4",
),
(
"""
pkgbase = foo
pkgver = 1.2.3
pkgname = foo
""",
"1.2.3",
),
],
)
def test_parse_srcinfo_version_handling(srcinfo: str, expected_version: str) -> None:
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/dummy")
info = pkg._parse_srcinfo(srcinfo)
assert info.pkgname == "foo"
assert info.pkgbase == "foo"
assert info.version == expected_version
def test_parse_srcinfo_single_package_dependencies() -> None:
srcinfo = """
pkgbase = foo
pkgver = 1.2.3
pkgrel = 1
depends = bar>=1.0
makedepends = baz
checkdepends = qux
pkgname = foo
"""
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/dummy")
info = pkg._parse_srcinfo(srcinfo)
assert info.dependencies == ("bar>=1.0",)
assert info.make_dependencies == ("baz",)
assert info.check_dependencies == ("qux",)
def test_parse_srcinfo_split_package_uses_only_target_pkg_dependencies(monkeypatch) -> None:
# Ensure arch-specific keys match
monkeypatch.setattr(pkg_mod.config, "arch", "x86_64", raising=False)
srcinfo = """
pkgbase = clion
pkgver = 2025.3
pkgrel = 1
makedepends = rsync
depends = base-dep
depends_x86_64 = base-arch-dep
pkgname = clion
depends = libdbusmenu-glib
depends_x86_64 = clion-arch-dep
checkdepends = clion-check
pkgname = clion-jre
depends = jre-dep
makedepends = jre-make
pkgname = clion-cmake
depends = cmake-dep
"""
pkg = CustomPackage(pkgname="clion", git_url=None, pkgbuild_directory="/dummy")
info = pkg._parse_srcinfo(srcinfo)
# version
assert info.pkgbase == "clion"
assert info.version == "2025.3-1"
# base deps + target pkg deps (including arch-specific)
assert info.dependencies == (
"base-dep",
"base-arch-dep",
"libdbusmenu-glib",
"clion-arch-dep",
)
# only base and target pkg makedepends
assert info.make_dependencies == ("rsync",)
# base + target pkg checkdepends
assert info.check_dependencies == ("clion-check",)
def test_parse_srcinfo_arch_specific_ignored_for_other_arch(monkeypatch) -> None:
# Different arch → *_x86_64 keys should be ignored
monkeypatch.setattr(pkg_mod.config, "arch", "aarch64", raising=False)
srcinfo = """
pkgbase = foo
pkgver = 1.0
pkgrel = 1
depends_x86_64 = base-arch-dep
pkgname = foo
depends = common-dep
depends_x86_64 = pkg-arch-dep
"""
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/dummy")
info = pkg._parse_srcinfo(srcinfo)
# Only common deps, no *_x86_64 because arch != x86_64
assert info.dependencies == ("common-dep",)
def test_parse_srcinfo_missing_required_fields_raises() -> None:
# Missing pkgbase
srcinfo_no_pkgbase = """
pkgver = 1.0
pkgrel = 1
pkgname = foo
"""
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/dummy")
with pytest.raises(PKGBUILDParseError) as excinfo:
pkg._parse_srcinfo(srcinfo_no_pkgbase)
assert "pkgbase/pkgver" in str(excinfo.value)
# Missing pkgver
srcinfo_no_pkgver = """
pkgbase = foo
pkgname = foo
"""
with pytest.raises(PKGBUILDParseError) as excinfo2:
pkg._parse_srcinfo(srcinfo_no_pkgver)
assert "pkgbase/pkgver" in str(excinfo2.value)
def test_parse_srcinfo_missing_target_pkg_raises() -> None:
srcinfo = """
pkgbase = foo
pkgver = 1.0
pkgrel = 1
pkgname = other
"""
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/dummy")
with pytest.raises(PKGBUILDParseError) as excinfo:
pkg._parse_srcinfo(srcinfo)
msg = str(excinfo.value)
assert "Package foo not found in SRCINFO" in msg
assert "other" in msg # listed in present packages
def test_srcinfo_from_pkgbuild_directory_missing_dir_raises(tmp_path: pathlib.Path) -> None:
missing = tmp_path / "does-not-exist"
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory=str(missing))
with pytest.raises(PKGBUILDParseError) as excinfo:
pkg._srcinfo_from_pkgbuild_directory(DummyCommands())
msg = str(excinfo.value)
assert "does not exist or is not a directory" in msg
def test_srcinfo_from_pkgbuild_directory_missing_pkgbuild_raises(tmp_path: pathlib.Path) -> None:
path = tmp_path / "pkgdir"
path.mkdir()
pkg = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory=str(path))
with pytest.raises(PKGBUILDParseError) as excinfo:
pkg._srcinfo_from_pkgbuild_directory(DummyCommands())
msg = str(excinfo.value)
assert "No PKGBUILD found" in msg
def test_custom_package_equality_and_hash() -> None:
a1 = CustomPackage(
pkgname="foo", git_url="https://example.com/repo.git", pkgbuild_directory=None
)
a2 = CustomPackage(
pkgname="foo", git_url="https://example.com/repo.git", pkgbuild_directory=None
)
b = CustomPackage(pkgname="foo", git_url=None, pkgbuild_directory="/some/path")
assert a1 == a2
assert hash(a1) == hash(a2)
assert a1 != b
assert hash(a1) != hash(b)
def test_custom_package_str_git_and_directory() -> None:
git_pkg = CustomPackage(
pkgname="foo",
git_url="https://example.com/repo.git",
pkgbuild_directory=None,
)
dir_pkg = CustomPackage(
pkgname="foo",
git_url=None,
pkgbuild_directory="/some/path",
)
assert "pkgname=foo" in str(git_pkg)
assert "git_url=https://example.com/repo.git" in str(git_pkg)
assert "pkgname=foo" in str(dir_pkg)
assert "pkgbuild_directory=/some/path" in str(dir_pkg)
# --- 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
@@ -0,0 +1,87 @@
import pytest
from decman.plugins.aur.error import DependencyCycleError
from decman.plugins.aur.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())
@@ -0,0 +1,313 @@
from typing import Any
import pytest
from decman.plugins import pacman as pacman_plugin
@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 pacman_plugin.strip_dependency(dep) == expected
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, database_signature_level, database_path
) -> 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, None, None)
def fake_pm_ctor(
commands, print_highlights, keywords, database_signature_level, database_path
) -> 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_traceback() -> None:
traceback_called.append(True)
def fake_print_continuation(msg: str) -> None:
continuations.append(msg)
monkeypatch.setattr(pacman_plugin.output, "print_error", fake_print_error)
monkeypatch.setattr(pacman_plugin.output, "print_traceback", fake_print_traceback)
monkeypatch.setattr(pacman_plugin.output, "print_continuation", fake_print_continuation)
ok = pacman.apply(store, dry_run=False)
assert ok is False
assert any("Pacman command exited with an unexpected" 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, database_signature_level, database_path
) -> None: # noqa: D401
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, None, None)
def fake_pm_ctor(
commands, print_highlights, keywords, database_signature_level, database_path
) -> 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"]
+20
View File
@@ -0,0 +1,20 @@
[project]
name = "decman-systemd"
version = "1.0.0"
requires-python = ">=3.13"
dependencies = ["decman==1.0.0"]
[project.entry-points."decman.plugins"]
systemd = "decman.plugins.systemd:Systemd"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
include = ["decman.plugins*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
@@ -0,0 +1,253 @@
import shutil
import decman.config as config
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 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
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
class SystemdCommands:
"""
Default commands for the Systemd plugin.
"""
def enable_units(self, units: set[str]) -> list[str]:
"""
Running this command enables the given systemd units.
"""
return ["systemctl", "enable"] + list(units)
def disable_units(self, units: set[str]) -> list[str]:
"""
Running this command disables the given systemd units.
"""
return ["systemctl", "disable"] + list(units)
def enable_user_units(self, units: set[str], user: str) -> list[str]:
"""
Running this command enables the given systemd units for the user.
"""
return ["systemctl", "--user", "-M", f"{user}@", "enable"] + list(units)
def disable_user_units(self, units: set[str], user: str) -> list[str]:
"""
Running this command disables the given systemd units for the user.
"""
return ["systemctl", "--user", "-M", f"{user}@", "disable"] + list(units)
def daemon_reload(self) -> list[str]:
"""
Running this command reloads the systemd daemon.
"""
return ["systemctl", "daemon-reload"]
def user_daemon_reload(self, user: str) -> list[str]:
"""
Running this command reloads the systemd daemon for the given user.
"""
return ["systemctl", "--user", "-M", f"{user}@", "daemon-reload"]
class Systemd(plugins.Plugin):
NAME = "systemd"
def __init__(self) -> None:
self.enabled_units: set[str] = set()
self.enabled_user_units: dict[str, set[str]] = {}
self.commands = SystemdCommands()
def available(self) -> bool:
return shutil.which("systemctl") is not None
def process_modules(self, store: _store.Store, modules: set[module.Module]):
# These store keys are used to track changes in modules.
# This way when these change, module can be marked as changed
store.ensure("systemd_units_for_module", {})
store.ensure("systemd_user_units_for_module", {})
for mod in modules:
store["systemd_units_for_module"].setdefault(mod.name, set())
store["systemd_user_units_for_module"].setdefault(mod.name, {})
units = plugins.run_method_with_attribute(mod, "__systemd__units__") or set()
user_units = plugins.run_method_with_attribute(mod, "__systemd__user__units__") or {}
if store["systemd_units_for_module"][mod.name] != units:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified systemd units."
)
if store["systemd_user_units_for_module"][mod.name] != user_units:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified systemd user units."
)
self.enabled_units |= units
for user, u_units in user_units.items():
self.enabled_user_units.setdefault(user, set()).update(u_units)
store["systemd_units_for_module"][mod.name] = units
store["systemd_user_units_for_module"][mod.name] = user_units
def apply(
self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None
) -> bool:
store.ensure("systemd_units", set())
store.ensure("systemd_user_units", {})
units_to_enable = set()
units_to_disable = set()
user_units_to_enable: dict[str, set[str]] = {}
user_units_to_disable: dict[str, set[str]] = {}
for unit in self.enabled_units:
if unit not in store["systemd_units"]:
units_to_enable.add(unit)
for unit in store["systemd_units"]:
if unit not in self.enabled_units:
units_to_disable.add(unit)
for user, units in self.enabled_user_units.items():
store["systemd_user_units"].setdefault(user, set())
user_units_to_enable.setdefault(user, set())
for unit in units:
if unit not in store["systemd_user_units"][user]:
user_units_to_enable[user].add(unit)
for user, units in store["systemd_user_units"].items():
self.enabled_user_units.setdefault(user, set())
user_units_to_disable.setdefault(user, set())
for unit in units:
if unit not in self.enabled_user_units[user]:
user_units_to_disable[user].add(unit)
try:
output.print_info("Reloading systemd daemon.")
if not dry_run:
self.reload_daemon()
output.print_info("Reloading systemd daemon for users.")
if not dry_run:
for user in user_units_to_enable.keys() | user_units_to_disable.keys():
self.reload_user_daemon(user)
output.print_list("Enabling systemd units:", list(units_to_enable))
if not dry_run:
self.enable_units(store, units_to_enable)
output.print_list("Disabling systemd units:", list(units_to_disable))
if not dry_run:
self.disable_units(store, units_to_disable)
for user, units in user_units_to_enable.items():
output.print_list(f"Enabling systemd units for {user}:", list(units))
if not dry_run:
self.enable_user_units(store, units, user)
for user, units in user_units_to_disable.items():
output.print_list(f"Disabling systemd units for {user}:", list(units))
if not dry_run:
self.disable_user_units(store, units, user)
except errors.CommandFailedError as error:
output.print_error("Running a systemd command failed.")
output.print_error(str(error))
if error.output:
output.print_command_output(error.output)
output.print_traceback()
return False
return True
def enable_units(self, store: _store.Store, units: set[str]):
"""
Enables the given units.
"""
if not units:
return
cmd = self.commands.enable_units(units)
command.prg(cmd, pty=config.debug_output)
store["systemd_units"] |= units
def disable_units(self, store: _store.Store, units: set[str]):
"""
Disables the given units.
"""
if not units:
return
cmd = self.commands.disable_units(units)
command.prg(cmd, pty=config.debug_output)
store["systemd_units"] -= units
def enable_user_units(self, store: _store.Store, units: set[str], user: str):
"""
Enables the given units for the given user.
"""
if not units:
return
cmd = self.commands.enable_user_units(units, user)
command.prg(cmd, pty=config.debug_output)
store["systemd_user_units"].setdefault(user, set())
store["systemd_user_units"][user] |= units
def disable_user_units(self, store: _store.Store, units: set[str], user: str):
"""
Disables the given units for the given user.
"""
if not units:
return
cmd = self.commands.disable_user_units(units, user)
command.prg(cmd, pty=config.debug_output)
store["systemd_user_units"].setdefault(user, set())
store["systemd_user_units"][user] -= units
def reload_user_daemon(self, user: str):
"""
Reloads the user's systemd daemon.
"""
cmd = self.commands.user_daemon_reload(user)
command.prg(cmd, pty=config.debug_output)
def reload_daemon(self):
"""
Reloads the systemd daemon.
"""
cmd = self.commands.daemon_reload()
command.prg(cmd, pty=config.debug_output)
@@ -0,0 +1,354 @@
import pytest
from decman.plugins import systemd as systemd_mod
class DummyStore(dict):
def ensure(self, key, default):
if key not in self:
self[key] = default
class DummyModule:
def __init__(self, name: str):
self.name = name
self._changed = False
@pytest.fixture
def store():
return DummyStore()
@pytest.fixture
def systemd():
return systemd_mod.Systemd()
def test_units_decorator_sets_attribute():
@systemd_mod.units
def fn():
pass
assert getattr(fn, "__systemd__units__", False) is True
def test_user_units_decorator_sets_attribute():
@systemd_mod.user_units
def fn():
pass
assert getattr(fn, "__systemd__user__units__", False) is True
def test_available_true_if_systemctl_found(monkeypatch, systemd):
called = {}
def fake_which(name):
called["name"] = name
return "/bin/systemctl"
monkeypatch.setattr(systemd_mod.shutil, "which", fake_which)
assert systemd.available() is True
assert called["name"] == "systemctl"
def test_available_false_if_systemctl_missing(monkeypatch, systemd):
monkeypatch.setattr(systemd_mod.shutil, "which", lambda name: None)
assert systemd.available() is False
def test_process_modules_marks_changed_and_updates_store(monkeypatch, store, systemd):
# initial store empty; ensure keys will be created
m1 = DummyModule("mod1")
m2 = DummyModule("mod2")
def fake_run_method(mod, attr):
if mod is m1 and attr == "__systemd__units__":
return {"a.service"}
if mod is m1 and attr == "__systemd__user__units__":
return {"alice": {"u1.service"}}
# m2 has no units
return None
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
systemd.process_modules(store, {m1, m2})
# m1 changed from default -> marked _changed
assert m1._changed is True
# m2 had no units
assert m2._changed is False
# enabled units aggregated
assert systemd.enabled_units == {"a.service"}
assert systemd.enabled_user_units == {"alice": {"u1.service"}}
# store updated per module
assert store["systemd_units_for_module"]["mod1"] == {"a.service"}
assert store["systemd_user_units_for_module"]["mod1"] == {"alice": {"u1.service"}}
assert store["systemd_units_for_module"]["mod2"] == set()
assert store["systemd_user_units_for_module"]["mod2"] == {}
def test_process_modules_no_change_second_run(monkeypatch, store, systemd):
m1 = DummyModule("mod1")
def fake_run_method(mod, attr):
if attr == "__systemd__units__":
return {"a.service"}
if attr == "__systemd__user__units__":
return {"alice": {"u1.service"}}
return None
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
# first run populates store
systemd.process_modules(store, {m1})
m1._changed = False
# new instance (fresh per-process in real usage)
systemd2 = systemd_mod.Systemd()
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
systemd2.process_modules(store, {m1})
# values in store are same -> _changed stays False
assert m1._changed is False
def test_apply_enables_and_disables_units_and_user_units(store):
s = systemd_mod.Systemd()
# Current enabled according to modules
s.enabled_units = {"new.service"}
s.enabled_user_units = {"alice": {"newuser.service"}}
# Store says we had an old unit enabled before
store["systemd_units"] = {"old.service"}
store["systemd_user_units"] = {"alice": {"olduser.service"}}
calls = []
def fake_reload_daemon():
calls.append(("reload_daemon",))
def fake_reload_user_daemon(user):
calls.append(("reload_user_daemon", user))
def fake_enable_units(store_arg, units_arg):
calls.append(("enable_units", frozenset(units_arg)))
store_arg["systemd_units"] |= units_arg
def fake_disable_units(store_arg, units_arg):
calls.append(("disable_units", frozenset(units_arg)))
store_arg["systemd_units"] -= units_arg
def fake_enable_user_units(store_arg, units_arg, user):
calls.append(("enable_user_units", user, frozenset(units_arg)))
store_arg["systemd_user_units"].setdefault(user, set()).update(units_arg)
def fake_disable_user_units(store_arg, units_arg, user):
calls.append(("disable_user_units", user, frozenset(units_arg)))
store_arg["systemd_user_units"].setdefault(user, set()).difference_update(units_arg)
# patch instance methods (no self parameter expected)
s.reload_daemon = fake_reload_daemon
s.reload_user_daemon = fake_reload_user_daemon
s.enable_units = fake_enable_units
s.disable_units = fake_disable_units
s.enable_user_units = fake_enable_user_units
s.disable_user_units = fake_disable_user_units
result = s.apply(store, dry_run=False, params=None)
# reloads called once
assert ("reload_daemon",) in calls
assert ("reload_user_daemon", "alice") in calls
# enable/disable correct units
assert ("enable_units", frozenset({"new.service"})) in calls
assert ("disable_units", frozenset({"old.service"})) in calls
assert ("enable_user_units", "alice", frozenset({"newuser.service"})) in calls
assert ("disable_user_units", "alice", frozenset({"olduser.service"})) in calls
# store reconciled
assert store["systemd_units"] == {"new.service"}
assert store["systemd_user_units"]["alice"] == {"newuser.service"}
def test_apply_dry_run_does_not_mutate_store_or_call_commands(store):
s = systemd_mod.Systemd()
s.enabled_units = {"new.service"}
s.enabled_user_units = {"alice": {"newuser.service"}}
store["systemd_units"] = {"old.service"}
store["systemd_user_units"] = {"alice": {"olduser.service"}}
called = {"reload": False, "enable": False, "disable": False}
s.reload_daemon = lambda: called.__setitem__("reload", True) or True
s.reload_user_daemon = lambda user: called.__setitem__("reload", True) or True
s.enable_units = lambda st, u: called.__setitem__("enable", True) or True
s.disable_units = lambda st, u: called.__setitem__("disable", True) or True
s.enable_user_units = lambda st, u, user: called.__setitem__("enable", True) or True
s.disable_user_units = lambda st, u, user: called.__setitem__("disable", True) or True
result = s.apply(store, dry_run=True, params=None)
assert result is True
# no commands should be called
assert called == {"reload": False, "enable": False, "disable": False}
# store unchanged
assert store["systemd_units"] == {"old.service"}
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
def test_enable_units_success(monkeypatch, store, systemd):
store["systemd_units"] = {"old.service"}
def fake_run(cmd, **kwargs):
assert cmd[0] == "systemctl"
assert cmd[1] == "enable"
assert "new.service" in cmd[2:]
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.enable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service", "new.service"}
def test_enable_units_failure_does_not_update_store(monkeypatch, store, systemd):
store["systemd_units"] = {"old.service"}
def fake_run(cmd, **kwargs):
return 1, "error"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.enable_units(store, {"new.service"})
# unchanged
assert store["systemd_units"] == {"old.service"}
def test_disable_units_success(monkeypatch, store, systemd):
store["systemd_units"] = {"old.service", "new.service"}
def fake_run(cmd, **kwargs):
assert cmd[0] == "systemctl"
assert cmd[1] == "disable"
assert "new.service" in cmd[2:]
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.disable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service"}
def test_disable_units_failure_does_not_update_store(monkeypatch, store, systemd):
store["systemd_units"] = {"old.service", "new.service"}
def fake_run(cmd, **kwargs):
return 1, "error"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.disable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service", "new.service"}
def test_enable_user_units_success(monkeypatch, store, systemd):
store["systemd_user_units"] = {"alice": {"olduser.service"}}
def fake_run(cmd, **kwargs):
assert cmd[0] == "systemctl"
assert "--user" in cmd
assert "enable" in cmd
assert "newuser.service" in cmd
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {
"olduser.service",
"newuser.service",
}
def test_enable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
store["systemd_user_units"] = {"alice": {"olduser.service"}}
def fake_run(cmd, **kwargs):
return 1, "error"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
def test_disable_user_units_success(monkeypatch, store, systemd):
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
def fake_run(cmd, **kwargs):
assert cmd[0] == "systemctl"
assert "--user" in cmd
assert "disable" in cmd
assert "newuser.service" in cmd
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
def test_disable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
def fake_run(cmd, **kwargs):
return 1, "error"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {
"olduser.service",
"newuser.service",
}
def test_reload_daemon_uses_command_run(monkeypatch, systemd):
called = {}
def fake_run(cmd, **kwargs):
called["cmd"] = cmd
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.reload_daemon()
assert called["cmd"][:2] == ["systemctl", "daemon-reload"]
def test_reload_user_daemon_uses_command_run(monkeypatch, systemd):
called = {}
def fake_run(cmd, **kwargs):
called["cmd"] = cmd
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
systemd.reload_user_daemon("alice")
cmd = called["cmd"]
assert cmd[0] == "systemctl"
assert "--user" in cmd
assert "daemon-reload" in cmd