From a02535719ee9c2ca9e253e70e2c6dc0454d99eed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Prchl=C3=ADk?= Date: Sat, 8 Mar 2025 20:40:52 +0100 Subject: [PATCH] Split package manager code into scripts and execution Current package manager classes were given a guest, and then were responsible for running commands on the guest to install stuff. To support delayed running of these commands - I'm looking at you, image mode! - we would like to use the existing implementations, use the existing code preparing commands like `dnf install foo`, we just don't want to run them on the guest. Package manager classes are now split into two parts: "engine" that is pretty much what package managers are now, i.e. prepares commands that would reaize the requested action, and the "package manager" whose job is to urn these commands. Each package manager has its own engine. The effect is, we can now can access an engine we pick, and use it to construct the commands without actually running them. We can put them into Containerfiles, or run them in a special SSH session, and so on. And it will use the same code as the usual "install me a package *now*" workflow of regular testing, preventing duplication and exceptions. --- tmt/package_managers/__init__.py | 103 +++++++++++++----- tmt/package_managers/apk.py | 93 +++++++++------- tmt/package_managers/apt.py | 113 ++++++++++---------- tmt/package_managers/dnf.py | 164 ++++++++++++++++------------- tmt/package_managers/rpm_ostree.py | 148 ++++++++++++++++---------- tmt/steps/prepare/distgit.py | 2 +- tmt/steps/prepare/install.py | 8 +- tmt/steps/provision/__init__.py | 14 ++- 8 files changed, 388 insertions(+), 257 deletions(-) diff --git a/tmt/package_managers/__init__.py b/tmt/package_managers/__init__.py index 829e33aa67..2aff48001f 100644 --- a/tmt/package_managers/__init__.py +++ b/tmt/package_managers/__init__.py @@ -1,13 +1,13 @@ import shlex from collections.abc import Iterator -from typing import TYPE_CHECKING, Any, Callable, Optional, Union +from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union import tmt import tmt.log import tmt.plugins import tmt.utils from tmt.container import container, simple_field -from tmt.utils import Command, CommandOutput, Path +from tmt.utils import Command, CommandOutput, Path, ShellScript if TYPE_CHECKING: from tmt._compat.typing import TypeAlias @@ -72,27 +72,32 @@ def __lt__(self, other: Any) -> bool: Installable = Union[Package, FileSystemPath, PackagePath, PackageUrl] -PackageManagerClass = type['PackageManager'] +PackageManagerEngineT = TypeVar('PackageManagerEngineT', bound='PackageManagerEngine') +PackageManagerClass = type['PackageManager[PackageManagerEngineT]'] -_PACKAGE_MANAGER_PLUGIN_REGISTRY: tmt.plugins.PluginRegistry[PackageManagerClass] = ( - tmt.plugins.PluginRegistry() -) +_PACKAGE_MANAGER_PLUGIN_REGISTRY: tmt.plugins.PluginRegistry[ + 'PackageManagerClass[PackageManagerEngine]' +] = tmt.plugins.PluginRegistry() def provides_package_manager( package_manager: str, -) -> Callable[[PackageManagerClass], PackageManagerClass]: +) -> Callable[ + ['PackageManagerClass[PackageManagerEngineT]'], 'PackageManagerClass[PackageManagerEngineT]' +]: """ 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: + def _provides_package_manager( + package_manager_cls: 'PackageManagerClass[PackageManagerEngineT]', + ) -> 'PackageManagerClass[PackageManagerEngineT]': _PACKAGE_MANAGER_PLUGIN_REGISTRY.register_plugin( plugin_id=package_manager, - plugin=package_manager_cls, + plugin=package_manager_cls, # type: ignore[arg-type] logger=tmt.log.Logger.get_bootstrap_logger(), ) @@ -101,7 +106,9 @@ def _provides_package_manager(package_manager_cls: PackageManagerClass) -> Packa return _provides_package_manager -def find_package_manager(name: 'GuestPackageManager') -> 'PackageManagerClass': +def find_package_manager( + name: 'GuestPackageManager', +) -> 'PackageManagerClass[PackageManagerEngine]': """ Find a package manager by its name. @@ -148,13 +155,66 @@ class Options: allow_untrusted: bool = False -class PackageManager(tmt.utils.Common): +class PackageManagerEngine(tmt.utils.Common): + command: Command + options: Command + + def __init__(self, *, guest: 'Guest', logger: tmt.log.Logger) -> None: + super().__init__(logger=logger) + + self.guest = guest + + self.command, self.options = self.prepare_command() + + def prepare_command(self) -> tuple[Command, Command]: + """ + Prepare installation command and subcommand options + """ + + raise NotImplementedError + + def check_presence(self, *installables: Installable) -> ShellScript: + """ + Return a presence status for each given installable + """ + + raise NotImplementedError + + def install( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise NotImplementedError + + def reinstall( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise NotImplementedError + + def install_debuginfo( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise NotImplementedError + + def refresh_metadata(self) -> ShellScript: + raise NotImplementedError + + +class PackageManager(tmt.utils.Common, Generic[PackageManagerEngineT]): """ A base class for package manager plugins """ NAME: str + _engine_class: type[PackageManagerEngineT] + engine: PackageManagerEngineT + #: A command to run to check whether the package manager is available on #: a guest. probe_command: Command @@ -166,21 +226,12 @@ class PackageManager(tmt.utils.Common): #: may be installed togethers, and therefore a priority is needed. probe_priority: int = 0 - command: Command - options: Command - 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.engine = self._engine_class(guest=guest, logger=logger) - def prepare_command(self) -> tuple[Command, Command]: - """ - Prepare installation command and subcommand options - """ - - raise NotImplementedError + self.guest = guest def check_presence(self, *installables: Installable) -> dict[Installable, bool]: """ @@ -194,21 +245,21 @@ def install( *installables: Installable, options: Optional[Options] = None, ) -> CommandOutput: - raise NotImplementedError + return self.guest.execute(self.engine.install(*installables, options=options)) def reinstall( self, *installables: Installable, options: Optional[Options] = None, ) -> CommandOutput: - raise NotImplementedError + return self.guest.execute(self.engine.reinstall(*installables, options=options)) def install_debuginfo( self, *installables: Installable, options: Optional[Options] = None, ) -> CommandOutput: - raise NotImplementedError + return self.guest.execute(self.engine.install_debuginfo(*installables, options=options)) def refresh_metadata(self) -> CommandOutput: - raise NotImplementedError + return self.guest.execute(self.engine.refresh_metadata()) diff --git a/tmt/package_managers/apk.py b/tmt/package_managers/apk.py index fbcc12c402..2515802e15 100644 --- a/tmt/package_managers/apk.py +++ b/tmt/package_managers/apk.py @@ -1,13 +1,14 @@ import re from typing import Optional, Union -import tmt.package_managers import tmt.utils from tmt.package_managers import ( FileSystemPath, Installable, Options, Package, + PackageManager, + PackageManagerEngine, PackagePath, escape_installables, provides_package_manager, @@ -32,12 +33,7 @@ } -@provides_package_manager('apk') -class Apk(tmt.package_managers.PackageManager): - NAME = 'apk' - - probe_command = Command('apk', '--version') - +class ApkEngine(PackageManagerEngine): install_command = Command('add') _sudo_prefix: Command @@ -100,42 +96,17 @@ def _construct_presence_script( return reduced_packages, shell_script - def check_presence(self, *installables: Installable) -> dict[Installable, bool]: - reduced_packages, presence_script = self._construct_presence_script(*installables) - - try: - output = self.guest.execute(presence_script) - stdout, stderr = output.stdout, output.stderr - - except RunError as exc: - stdout, stderr = exc.stdout, exc.stderr + def check_presence(self, *installables: Installable) -> ShellScript: + return self._construct_presence_script(*installables)[1] - if stdout is None or stderr is None: - raise GeneralError("apk presence check output provided no output") - - results: dict[Installable, bool] = {} - - for installable, package in zip(installables, reduced_packages): - match = re.search(rf'^{re.escape(str(package))}\s', stdout) - - if match is not None: - results[installable] = True - continue - - results[installable] = False - - return results - - def refresh_metadata(self) -> CommandOutput: - script = ShellScript(f'{self.command.to_script()} update') - - return self.guest.execute(script) + def refresh_metadata(self) -> ShellScript: + return ShellScript(f'{self.command.to_script()} update') def install( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() packages = self._reduce_to_packages(*installables) @@ -152,13 +123,13 @@ def install( if options.skip_missing: script = script | ShellScript('/bin/true') - return self.guest.execute(script) + return script def reinstall( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() packages = self._reduce_to_packages(*installables) @@ -173,7 +144,49 @@ def reinstall( if options.skip_missing: script = script | ShellScript('/bin/true') - return self.guest.execute(script) + return script + + def install_debuginfo( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise tmt.utils.GeneralError("There is no support for debuginfo packages in apk.") + + +@provides_package_manager('apk') +class Apk(PackageManager[ApkEngine]): + NAME = 'apk' + + _engine_class = ApkEngine + + probe_command = Command('apk', '--version') + + def check_presence(self, *installables: Installable) -> dict[Installable, bool]: + reduced_packages, presence_script = self.engine._construct_presence_script(*installables) + + try: + output = self.guest.execute(presence_script) + stdout, stderr = output.stdout, output.stderr + + except RunError as exc: + stdout, stderr = exc.stdout, exc.stderr + + if stdout is None or stderr is None: + raise GeneralError("apk presence check output provided no output") + + results: dict[Installable, bool] = {} + + for installable, package in zip(installables, reduced_packages): + match = re.search(rf'^{re.escape(str(package))}\s', stdout) + + if match is not None: + results[installable] = True + continue + + results[installable] = False + + return results def install_debuginfo( self, diff --git a/tmt/package_managers/apt.py b/tmt/package_managers/apt.py index 50c988255e..ca0df1fbf5 100644 --- a/tmt/package_managers/apt.py +++ b/tmt/package_managers/apt.py @@ -1,13 +1,14 @@ import re from typing import Optional, Union -import tmt.package_managers import tmt.utils from tmt.package_managers import ( FileSystemPath, Installable, Options, Package, + PackageManager, + PackageManagerEngine, PackagePath, escape_installables, provides_package_manager, @@ -15,8 +16,6 @@ from tmt.utils import ( Command, CommandOutput, - Environment, - EnvVarValue, GeneralError, RunError, ShellScript, @@ -25,12 +24,7 @@ ReducedPackages = list[Union[Package, PackagePath]] -@provides_package_manager('apt') -class Apt(tmt.package_managers.PackageManager): - NAME = 'apt' - - probe_command = Command('apt', '--version') - +class AptEngine(PackageManagerEngine): install_command = Command('install') _sudo_prefix: Command @@ -104,37 +98,8 @@ def _construct_presence_script( f'dpkg-query --show {" ".join(escape_installables(*reduced_packages))}' ) - def check_presence(self, *installables: Installable) -> dict[Installable, bool]: - reduced_packages, presence_script = self._construct_presence_script(*installables) - - try: - output = self.guest.execute(presence_script) - stdout, stderr = output.stdout, output.stderr - - except RunError as exc: - stdout, stderr = exc.stdout, exc.stderr - - if stdout is None or stderr is None: - raise GeneralError("apt presence check provided no output") - - results: dict[Installable, bool] = {} - - for installable, package in zip(installables, reduced_packages): - match = re.search( - rf'dpkg-query: no packages found matching {re.escape(str(package))}', stderr - ) - - if match is not None: - results[installable] = False - continue - - match = re.search(rf'^{re.escape(str(package))}\s', stdout) - - if match is not None: - results[installable] = True - continue - - return results + def check_presence(self, *installables: Installable) -> ShellScript: + return self._construct_presence_script(*installables)[1] def _extra_options(self, options: Options) -> Command: extra_options = Command() @@ -144,18 +109,16 @@ def _extra_options(self, options: Options) -> Command: return extra_options - def refresh_metadata(self) -> CommandOutput: - script = ShellScript(f'{self.command.to_script()} update') - - return self.guest.execute( - script, env=Environment({'DEBIAN_FRONTEND': EnvVarValue('noninteractive')}) + def refresh_metadata(self) -> ShellScript: + return ShellScript( + f'export DEBIAN_FRONTEND=noninteractive; {self.command.to_script()} update' ) def install( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() extra_options = self._extra_options(options) @@ -176,15 +139,13 @@ def install( if options.skip_missing: script = script | ShellScript('/bin/true') - return self.guest.execute( - script, env=Environment({'DEBIAN_FRONTEND': EnvVarValue('noninteractive')}) - ) + return ShellScript(f'export DEBIAN_FRONTEND=noninteractive; {script}') def reinstall( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() extra_options = self._extra_options(options) @@ -202,9 +163,55 @@ def reinstall( if options.skip_missing: script = script | ShellScript('/bin/true') - return self.guest.execute( - script, env=Environment({'DEBIAN_FRONTEND': EnvVarValue('noninteractive')}) - ) + return ShellScript(f'export DEBIAN_FRONTEND=noninteractive; {script}') + + def install_debuginfo( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise tmt.utils.GeneralError("There is no support for debuginfo packages in apt.") + + +@provides_package_manager('apt') +class Apt(PackageManager[AptEngine]): + NAME = 'apt' + + _engine_class = AptEngine + + probe_command = Command('apt', '--version') + + def check_presence(self, *installables: Installable) -> dict[Installable, bool]: + reduced_packages, presence_script = self.engine._construct_presence_script(*installables) + + try: + output = self.guest.execute(presence_script) + stdout, stderr = output.stdout, output.stderr + + except RunError as exc: + stdout, stderr = exc.stdout, exc.stderr + + if stdout is None or stderr is None: + raise GeneralError("apt presence check provided no output") + + results: dict[Installable, bool] = {} + + for installable, package in zip(installables, reduced_packages): + match = re.search( + rf'dpkg-query: no packages found matching {re.escape(str(package))}', stderr + ) + + if match is not None: + results[installable] = False + continue + + match = re.search(rf'^{re.escape(str(package))}\s', stdout) + + if match is not None: + results[installable] = True + continue + + return results def install_debuginfo( self, diff --git a/tmt/package_managers/dnf.py b/tmt/package_managers/dnf.py index cb7612f2e8..d58304017e 100644 --- a/tmt/package_managers/dnf.py +++ b/tmt/package_managers/dnf.py @@ -1,32 +1,20 @@ import re from typing import Optional, cast -import tmt.package_managers from tmt.package_managers import ( FileSystemPath, Installable, Options, Package, + PackageManager, + PackageManagerEngine, escape_installables, provides_package_manager, ) -from tmt.utils import Command, CommandOutput, GeneralError, RunError, ShellScript +from tmt.utils import Command, GeneralError, RunError, ShellScript -@provides_package_manager('dnf') -class Dnf(tmt.package_managers.PackageManager): - NAME = 'dnf' - - probe_command = ShellScript( - """ - type dnf && ((dnf --version | grep -E 'dnf5 version') && exit 1 || exit 0) - """ - ).to_shell_command() - # The priority of preference: `rpm-ostree` > `dnf5` > `dnf` > `yum`. - # `rpm-ostree` has its own implementation and its own priority, and - # the `dnf` family just stays below it. - probe_priority = 50 - +class DnfEngine(PackageManagerEngine): _base_command = Command('dnf') _base_debuginfo_command = Command('debuginfo-install') @@ -78,33 +66,8 @@ def _construct_presence_script( return ShellScript(f'rpm -q {" ".join(escape_installables(*installables))}') - def check_presence(self, *installables: Installable) -> dict[Installable, bool]: - try: - output = self.guest.execute(self._construct_presence_script(*installables)) - stdout = output.stdout - - except RunError as exc: - stdout = exc.stdout - - if stdout is None: - raise GeneralError("rpm presence check provided no output") - - results: dict[Installable, bool] = {} - - for line, installable in zip(stdout.strip().splitlines(), installables): - match = re.match(rf'package {re.escape(str(installable))} is not installed', line) - if match is not None: - results[installable] = False - continue - - match = re.match(rf'no package provides {re.escape(str(installable))}', line) - if match is not None: - results[installable] = False - continue - - results[installable] = True - - return results + def check_presence(self, *installables: Installable) -> ShellScript: + return self._construct_presence_script(*installables) def _construct_install_script( self, *installables: Installable, options: Optional[Options] = None @@ -155,38 +118,36 @@ def _construct_install_debuginfo_script( f'{" ".join(escape_installables(*installables))}' ) - def refresh_metadata(self) -> CommandOutput: - script = ShellScript( + def refresh_metadata(self) -> ShellScript: + return ShellScript( f'{self.command.to_script()} makecache {self.options.to_script()} --refresh' ) - return self.guest.execute(script) - def install( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: - return self.guest.execute(self._construct_install_script(*installables, options=options)) + ) -> ShellScript: + return self._construct_install_script(*installables, options=options) def reinstall( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: - return self.guest.execute(self._construct_reinstall_script(*installables, options=options)) + ) -> ShellScript: + return self._construct_reinstall_script(*installables, options=options) def install_debuginfo( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: # Make sure debuginfo-install is present on the target system - self.install(FileSystemPath('/usr/bin/debuginfo-install')) + script = self.install(FileSystemPath('/usr/bin/debuginfo-install')) options = options or Options() - script = cast( # type: ignore[redundant-cast] + script &= cast( # type: ignore[redundant-cast] ShellScript, self._construct_install_debuginfo_script( # type: ignore[reportGeneralIssues,unused-ignore] *installables, options=options @@ -203,31 +164,70 @@ def install_debuginfo( ), ) - return self.guest.execute(script) + return script -@provides_package_manager('dnf5') -class Dnf5(Dnf): - NAME = 'dnf5' +@provides_package_manager('dnf') +class Dnf(PackageManager[DnfEngine]): + NAME = 'dnf' - probe_command = probe_command = Command('dnf5', '--version') - probe_priority = 60 + _engine_class = DnfEngine + + probe_command = ShellScript( + """ + type dnf && ((dnf --version | grep -E 'dnf5 version') && exit 1 || exit 0) + """ + ).to_shell_command() + # The priority of preference: `rpm-ostree` > `dnf5` > `dnf` > `yum`. + # `rpm-ostree` has its own implementation and its own priority, and + # the `dnf` family just stays below it. + probe_priority = 50 + + def check_presence(self, *installables: Installable) -> dict[Installable, bool]: + try: + output = self.guest.execute(self.engine.check_presence(*installables)) + stdout = output.stdout + + except RunError as exc: + stdout = exc.stdout + + if stdout is None: + raise GeneralError("rpm presence check provided no output") + results: dict[Installable, bool] = {} + + for line, installable in zip(stdout.strip().splitlines(), installables): + match = re.match(rf'package {re.escape(str(installable))} is not installed', line) + if match is not None: + results[installable] = False + continue + + match = re.match(rf'no package provides {re.escape(str(installable))}', line) + if match is not None: + results[installable] = False + continue + + results[installable] = True + + return results + + +class Dnf5Engine(DnfEngine): _base_command = Command('dnf5') skip_missing_packages_option = '--skip-unavailable' -@provides_package_manager('yum') -class Yum(Dnf): - NAME = 'yum' +@provides_package_manager('dnf5') +class Dnf5(Dnf): + NAME = 'dnf5' - probe_command = ShellScript( - """ - type yum && ((yum --version | grep -E 'dnf5 version') && exit 1 || exit 0) - """ - ).to_shell_command() - probe_priority = 40 + _engine_class = Dnf5Engine + + probe_command = probe_command = Command('dnf5', '--version') + probe_priority = 60 + +class YumEngine(DnfEngine): _base_command = Command('yum') # TODO: get rid of those `type: ignore` below. I think it's caused by the @@ -236,7 +236,7 @@ class Yum(Dnf): # for now. def install( self, *installables: Installable, options: Optional[Options] = None - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() script = cast( # type: ignore[redundant-cast] @@ -258,11 +258,11 @@ def install( ), ) - return self.guest.execute(script) + return script def reinstall( self, *installables: Installable, options: Optional[Options] = None - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() script = cast( # type: ignore[redundant-cast] @@ -284,9 +284,21 @@ def reinstall( ), ) - return self.guest.execute(script) + return script + + def refresh_metadata(self) -> ShellScript: + return ShellScript(f'{self.command.to_script()} makecache') - def refresh_metadata(self) -> CommandOutput: - script = ShellScript(f'{self.command.to_script()} makecache') - return self.guest.execute(script) +@provides_package_manager('yum') +class Yum(Dnf): + NAME = 'yum' + + _engine_class = YumEngine + + probe_command = ShellScript( + """ + type yum && ((yum --version | grep -E 'dnf5 version') && exit 1 || exit 0) + """ + ).to_shell_command() + probe_priority = 40 diff --git a/tmt/package_managers/rpm_ostree.py b/tmt/package_managers/rpm_ostree.py index cc964f8855..a4fb78d93b 100644 --- a/tmt/package_managers/rpm_ostree.py +++ b/tmt/package_managers/rpm_ostree.py @@ -1,26 +1,19 @@ import re from typing import Optional -import tmt.package_managers -import tmt.utils from tmt.package_managers import ( FileSystemPath, Installable, Options, + PackageManager, + PackageManagerEngine, escape_installables, provides_package_manager, ) from tmt.utils import Command, CommandOutput, GeneralError, RunError, ShellScript -@provides_package_manager('rpm-ostree') -class RpmOstree(tmt.package_managers.PackageManager): - NAME = 'rpm-ostree' - - probe_command = Command('stat', '/run/ostree-booted') - # Needs to be bigger than priorities of `yum`, `dnf` and `dnf5`. - probe_priority = 100 - +class RpmOstreeEngine(PackageManagerEngine): def prepare_command(self) -> tuple[Command, Command]: """ Prepare installation command for rpm-ostree @@ -43,49 +36,11 @@ def _construct_presence_script(self, *installables: Installable) -> ShellScript: return ShellScript(f'rpm -q --whatprovides {" ".join(escape_installables(*installables))}') - def check_presence(self, *installables: Installable) -> dict[Installable, bool]: + def check_presence(self, *installables: Installable) -> ShellScript: if len(installables) == 1 and isinstance(installables[0], FileSystemPath): - try: - self.guest.execute(ShellScript(f'rpm -qf {installables[0]}')) - - except RunError as exc: - if exc.returncode == 1: - return {installables[0]: False} - - raise exc - - return {installables[0]: True} - - try: - output = self.guest.execute( - ShellScript( - f'rpm -q --whatprovides {" ".join(escape_installables(*installables))}' - ) - ) - stdout = output.stdout - - except RunError as exc: - stdout = exc.stdout - - if stdout is None: - raise GeneralError("rpm presence check provided no output") - - results: dict[Installable, bool] = {} - - for line, installable in zip(stdout.strip().splitlines(), installables): - match = re.match(rf'package {re.escape(str(installable))} is not installed', line) - if match is not None: - results[installable] = False - continue - - match = re.match(rf'no package provides {re.escape(str(installable))}', line) - if match is not None: - results[installable] = False - continue - - results[installable] = True + return ShellScript(f'rpm -qf {installables[0]}') - return results + return ShellScript(f'rpm -q --whatprovides {" ".join(escape_installables(*installables))}') def _extra_options(self, options: Options) -> Command: extra_options = Command() @@ -104,10 +59,10 @@ def _extra_options(self, options: Options) -> Command: return extra_options - def refresh_metadata(self) -> CommandOutput: + def refresh_metadata(self) -> ShellScript: self.guest.warn("Metadata refresh is not supported with rpm-ostree.") - return CommandOutput(stdout=None, stderr=None) + return ShellScript('/bin/true') # The following should work, but it hits some ostree issue: # @@ -124,7 +79,7 @@ def install( self, *installables: Installable, options: Optional[Options] = None, - ) -> CommandOutput: + ) -> ShellScript: options = options or Options() extra_options = self._extra_options(options) @@ -141,7 +96,90 @@ def install( if options.skip_missing: script = script | ShellScript('/bin/true') - return self.guest.execute(script) + return script + + def reinstall( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise GeneralError("rpm-ostree does not support reinstall operation.") + + def install_debuginfo( + self, + *installables: Installable, + options: Optional[Options] = None, + ) -> ShellScript: + raise GeneralError("rpm-ostree does not support debuginfo packages.") + + +@provides_package_manager('rpm-ostree') +class RpmOstree(PackageManager[RpmOstreeEngine]): + NAME = 'rpm-ostree' + + _engine_class = RpmOstreeEngine + + probe_command = Command('stat', '/run/ostree-booted') + # Needs to be bigger than priorities of `yum`, `dnf` and `dnf5`. + probe_priority = 100 + + def check_presence(self, *installables: Installable) -> dict[Installable, bool]: + script = self.engine.check_presence(*installables) + + if len(installables) == 1 and isinstance(installables[0], FileSystemPath): + try: + self.guest.execute(script) + + except RunError as exc: + if exc.returncode == 1: + return {installables[0]: False} + + raise exc + + return {installables[0]: True} + + try: + output = self.guest.execute(script) + stdout = output.stdout + + except RunError as exc: + stdout = exc.stdout + + if stdout is None: + raise GeneralError("rpm presence check provided no output") + + results: dict[Installable, bool] = {} + + for line, installable in zip(stdout.strip().splitlines(), installables): + match = re.match(rf'package {re.escape(str(installable))} is not installed', line) + if match is not None: + results[installable] = False + continue + + match = re.match(rf'no package provides {re.escape(str(installable))}', line) + if match is not None: + results[installable] = False + continue + + results[installable] = True + + return results + + def refresh_metadata(self) -> CommandOutput: + self.guest.warn("Metadata refresh is not supported with rpm-ostree.") + + return CommandOutput(stdout=None, stderr=None) + + # The following should work, but it hits some ostree issue: + # + # System has not been booted with systemd as init system (PID 1). Can't operate. + # Failed to connect to bus: Host is down + # System has not been booted with systemd as init system (PID 1). Can't operate. + # Failed to connect to bus: Host is down + # error: Loading sysroot: exit status: 1 + # + # script = ShellScript(f'{self.command.to_script()} refresh-md --force') + # return self.guest.execute(script) def reinstall( self, diff --git a/tmt/steps/prepare/distgit.py b/tmt/steps/prepare/distgit.py index bc320b5f9b..8be87422fd 100644 --- a/tmt/steps/prepare/distgit.py +++ b/tmt/steps/prepare/distgit.py @@ -253,7 +253,7 @@ def go( raise tmt.utils.PrepareError('No src.rpm file created by the `rpmbuild -br` call.') # Install build requires # Create the package manager command - cmd, _ = guest.package_manager.prepare_command() + cmd, _ = guest.package_manager.engine.prepare_command() # Can't set 'cwd' as the check for its existence fails for local workdir cmd += Command("builddep", "-y", f"SRPMS/{src_rpm_name}") guest.execute(command=cmd, cwd=Path(source_dir)) diff --git a/tmt/steps/prepare/install.py b/tmt/steps/prepare/install.py index 3ca647ad37..15120393c2 100644 --- a/tmt/steps/prepare/install.py +++ b/tmt/steps/prepare/install.py @@ -281,10 +281,12 @@ def enable_copr(self, repositories: list[str]) -> None: if not repositories: return + package_manager = self.guest.package_manager + # Try to install copr plugin self.debug('Make sure the copr plugin is available.') try: - self.guest.package_manager.install(Package(self.copr_plugin)) + package_manager.install(Package(self.copr_plugin)) # Enable repositories manually for epel6 except tmt.utils.RunError: @@ -299,8 +301,8 @@ def enable_copr(self, repositories: list[str]) -> None: self.guest.execute( ShellScript( - f"{self.guest.package_manager.command.to_script()} copr " - f"{self.guest.package_manager.options.to_script()} enable -y {repository}" + f"{package_manager.engine.command.to_script()} copr " + f"{package_manager.engine.options.to_script()} enable -y {repository}" ) ) diff --git a/tmt/steps/provision/__init__.py b/tmt/steps/provision/__init__.py index 6bc6e60aa7..94d79b0e92 100644 --- a/tmt/steps/provision/__init__.py +++ b/tmt/steps/provision/__init__.py @@ -45,7 +45,11 @@ from tmt.container import SerializableContainer, container, field, key_to_option from tmt.log import Logger from tmt.options import option -from tmt.package_managers import FileSystemPath, Package, PackageManagerClass +from tmt.package_managers import ( + FileSystemPath, + Package, + PackageManagerClass, +) from tmt.plugins import PluginRegistry from tmt.steps import Action, ActionTask, PhaseQueue from tmt.utils import ( @@ -550,7 +554,9 @@ def _query_package_manager( # one available. Collect them, and sort them by their priorities # to find the most suitable one. - discovered_package_managers: list[PackageManagerClass] = [] + discovered_package_managers: list[ + PackageManagerClass[tmt.package_managers.PackageManagerEngine] + ] = [] for ( _, @@ -1088,7 +1094,9 @@ def is_ready(self) -> bool: raise NotImplementedError @functools.cached_property - def package_manager(self) -> 'tmt.package_managers.PackageManager': + def package_manager( + self, + ) -> 'tmt.package_managers.PackageManager[tmt.package_managers.PackageManagerEngine]': if not self.facts.package_manager: raise tmt.utils.GeneralError( f"Package manager was not detected on guest '{self.name}'."