Extract "package manager" functionality into plugins

More and more we see the need for primitives like "install a package".
Plugins start to require packages to be present on guests, we already do
have `prepare/install` plugin, and the upcoming all mighty install
plugin. To support all these functions, `tmt.package_managers` would
provide necessary primites, allowing implementations to be shipped as
plugins.

The patch starts building the primitives, converting
`tmt.steps.prepare.install` to use them in the process.

tmt/package_managers/__init__.py

@@ -0,0 +1,179 @@
+from typing import Optional, Callable, TYPE_CHECKING
+
+import fmf
+import fmf.utils
+
+import tmt
+import tmt.log
+import tmt.utils
+from tmt.utils import Command, CommandOutput, Path, ShellScript, field
+import enum
+import tmt.plugins
+
+if TYPE_CHECKING:
+ from tmt.steps.provision import Guest
+
+# COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
+
+
+class GuestPackageManager(enum.Enum):
+ DNF = 'dnf'
+ DNF5 = 'dnf5'
+ YUM = 'yum'
+ RPM_OSTREE = 'rpm-ostree'
+
+
+PackageManagerClass = type['PackageManager']
+
+
+_PACKAGE_MANAGER_PLUGIN_REGISTRY: tmt.plugins.PluginRegistry[PackageManagerClass] = \
+ tmt.plugins.PluginRegistry()
+
+
+def provides_package_manager(package_manager: str) -> Callable[[PackageManagerClass], PackageManagerClass]:
+ """
+ A decorator for registering package managers.
+
+ Decorate a package manager plugin class to register a package manager.
+ """
+
+ def _provides_package_manager(package_manager_cls: PackageManagerClass) -> PackageManagerClass:
+ _PACKAGE_MANAGER_PLUGIN_REGISTRY.register_plugin(
+ plugin_id=package_manager,
+ plugin=package_manager_cls,
+ logger=tmt.log.Logger.get_bootstrap_logger())
+
+ return package_manager_cls
+
+ return _provides_package_manager
+
+
+def find_package_manager(name: GuestPackageManager) -> 'PackageManagerClass':
+ """
+ Find a package manager by its name.
+
+ :raises GeneralError: when the plugin does not exist.
+ """
+
+ plugin = _PACKAGE_MANAGER_PLUGIN_REGISTRY.get_plugin(name.value)
+
+ if plugin is None:
+ raise tmt.utils.GeneralError(
+ f"Package manager '{name}' was not found in package manager registry.")
+
+ return plugin
+
+
+class PackageManager(tmt.utils.Common):
+ base_command: Command
+
+ command: Command
+ options: Command
+
+ # Each installer knows its package manager and copr plugin
+ # package_manager: str
+ # copr_plugin: str
+
+ # skip_missing: bool = False
+
+ # packages: list[str]
+ # directories: list[Path]
+ # exclude: list[str]
+
+ # local_packages: list[Path]
+ # remote_packages: list[str]
+ # debuginfo_packages: list[str]
+ # repository_packages: list[str]
+
+ # rpms_directory: Path
+
+ def __init__(self, *, guest: 'Guest', logger: tmt.log.Logger) -> None:
+ super().__init__(logger=logger)
+
+ self.guest = guest
+ self.command, self.options = self.prepare_command()
+
+ # self.debug(f"Using '{self.command}' for all package operations.")
+ # self.debug(f"Options for package operations are '{self.options}'.")
+
+ def prepare_command(self) -> tuple[Command, Command]:
+ """ Prepare installation command and subcommand options """
+ raise NotImplementedError
+
+ def operation_script(self, subcommand: Command, args: Command) -> ShellScript:
+ """
+ Render a shell script to perform the requested package operation.
+
+ .. warning::
+
+ Each and every argument from ``args`` **will be** sanitized by
+ escaping. This is not compatible with operations that wish to use
+ shell wildcards. Such operations need to be constructed manually.
+
+ :param subcommand: package manager subcommand, e.g. ``install`` or ``erase``.
+ :param args: arguments for the subcommand, e.g. package names.
+ """
+
+ return ShellScript(
+ f"{self.command.to_script()} {subcommand.to_script()} "
+ f"{self.options.to_script()} {args.to_script()}")
+
+ def perform_operation(self, subcommand: Command, args: Command) -> tmt.utils.CommandOutput:
+ """
+ Perform the requested package operation.
+
+ .. warning::
+
+ Each and every argument from ``args`` **will be** sanitized by
+ escaping. This is not compatible with operations that wish to use
+ shell wildcards. Such operations need to be constructed manually.
+
+ :param subcommand: package manager subcommand, e.g. ``install`` or ``erase``.
+ :param args: arguments for the subcommand, e.g. package names.
+ :returns: command output.
+ """
+
+ return self.guest.execute(self.operation_script(subcommand, args))
+
+ def query(self, packages: list[Command]) -> CommandOutput:
+ raise NotImplementedError
+
+ def install(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False,
+ check_first: bool = True) -> CommandOutput:
+ raise NotImplementedError
+
+ def install_local(
+ self,
+ packages: list[str],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ raise NotImplementedError
+
+ def reinstall(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ raise NotImplementedError
+
+ def reinstall_local(
+ self,
+ packages: list[str],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ raise NotImplementedError
+
+ def install_debuginfo(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ raise NotImplementedError
+
+# self.execute(ShellScript(f"{package_manager} install -y rsync" + readonly))
+
+

tmt/package_managers/dnf.py

@@ -0,0 +1,159 @@
+from typing import Optional
+import tmt.package_managers
+from tmt.utils import Command, CommandOutput, ShellScript
+from tmt.package_managers import provides_package_manager
+
+
+@provides_package_manager('dnf')
+class Dnf(tmt.package_managers.PackageManager):
+ base_command = Command('dnf')
+
+ skip_missing_option = '--skip-broken'
+
+ def prepare_command(self) -> tuple[Command, Command]:
+ options = Command('-y')
+ command = Command()
+
+ if self.guest.facts.is_superuser is False:
+ command += Command('sudo')
+
+ command += self.base_command
+
+ return (command, options)
+
+ def _extra_options(
+ self,
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> Command:
+ extra_options = Command()
+
+ for package in (excluded_packages or []):
+ extra_options += Command('--exclude', package)
+
+ if skip_missing:
+ extra_options += Command(self.skip_missing_option)
+
+ return extra_options
+
+ def query(self, packages: list[Command]) -> CommandOutput:
+ return self.guest.execute(ShellScript(
+ f"""
+ rpm -q {" ".join(str(package.to_script()) for package in packages)}
+ """
+ ))
+
+ def install(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False,
+ check_first: bool = True) -> CommandOutput:
+ extra_options = self._extra_options(
+ excluded_packages=excluded_packages,
+ skip_missing=skip_missing)
+
+ packages_as_arguments = " ".join(str(package.to_script()) for package in packages)
+
+ if check_first:
+ return self.guest.execute(ShellScript(
+ f'rpm -q --whatprovides {packages_as_arguments}'
+ ' || '
+ f'{self.command.to_script()} install {self.options.to_script()} {extra_options} {packages_as_arguments}'
+ ))
+
+ return self.guest.execute(ShellScript(
+ f"""
+ {self.command.to_script()} install \
+ {self.options.to_script()} \
+ {extra_options} \
+ {packages_as_arguments}
+ """
+ ))
+
+ def install_local(
+ self,
+ packages: list[str],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ extra_options = self._extra_options(
+ excluded_packages=excluded_packages,
+ skip_missing=skip_missing)
+
+ return self.guest.execute(ShellScript(
+ f"""
+ {self.command.to_script()} install \
+ {self.options.to_script()} \
+ {extra_options} \
+ {" ".join(*packages)}
+ """
+ ))
+
+ def reinstall(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ extra_options = self._extra_options(
+ excluded_packages=excluded_packages,
+ skip_missing=skip_missing)
+
+ return self.guest.execute(ShellScript(
+ f"""
+ {self.command.to_script()} reinstall \
+ {self.options.to_script()} \
+ {extra_options} \
+ {" ".join(str(package.to_script()) for package in packages)}
+ """
+ ))
+
+ def reinstall_local(
+ self,
+ packages: list[str],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ extra_options = self._extra_options(
+ excluded_packages=excluded_packages,
+ skip_missing=skip_missing)
+
+ return self.guest.execute(ShellScript(
+ f"""
+ {self.command.to_script()} reinstall \
+ {self.options.to_script()} \
+ {extra_options} \
+ {" ".join(*packages)}
+ """
+ ))
+
+ def install_debuginfo(
+ self,
+ packages: list[Command],
+ excluded_packages: Optional[list[str]] = None,
+ skip_missing: bool = False) -> CommandOutput:
+ # Make sure debuginfo-install is present on the target system
+ self.perform_operation(
+ Command('install'),
+ Command('/usr/bin/debuginfo-install')
+ )
+
+ extra_options = self._extra_options(
+ excluded_packages=excluded_packages,
+ skip_missing=skip_missing)
+
+ return self.guest.execute(ShellScript(
+ f"""
+ debuginfo-install -y \
+ {self.options.to_script()} \
+ {extra_options} \
+ {" ".join(str(package.to_script()) for package in packages)}
+ """
+ ))
+
+
+@provides_package_manager('dnf5')
+class Dnf5(Dnf):
+ base_command = Command('dnf5')
+
+
+@provides_package_manager('yum')
+class Yum(Dnf):
+ base_command = Command('yum')