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

@@ -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())

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,