Expose actual guest HW description in guest topology

Related to #2402. The patch
bootstraps `hardware` field in the guest topology content, and
populates it it very basic content. Extending it with more info,
e.g. `disk` or `system`, will be task for future patches.

spec/plans/guest-topology.fmf

@@ -20,6 +20,14 @@ description: |
         The shell-friendly file contains arrays, therefore it's compatible
         with Bash 4.x and newer.
 
+    .. note::
+
+        The ``hardware`` key, describing the actual HW configuration of the
+        guest, as understood by tmt, is using
+        :ref:`hardware specification </spec/hardware>`
+        and as such is still a work in progress. Not all hardware properties
+        might be available yet.
+
     .. note::
 
         The shell-friendly file is easy to ingest for a shell-based tests,
@@ -34,6 +42,7 @@ description: |
             name: ...
             role: ...
             hostname: ...
+            hardware: ...
 
         # List of names of all provisioned guests.
         guest-names:
@@ -48,10 +57,12 @@ description: |
                 name: guest1
                 role: ...
                 hostname: ...
+                hardware: ...
             guest2:
                 name: guest2
                 role: ...
                 hostname: ...
+                hardware: ...
             ...
 
         # List of all known roles.
@@ -102,6 +113,9 @@ description: |
         TMT_ROLES[role2]="guestN ..."
         ...
 
+    .. versionadded:: 1.31
+        Guest HW exposed via ``hardware`` key.
+
 example:
   - |
     # A trivial pseudo-test script

tmt/hardware.py

@@ -1324,6 +1324,21 @@ def parse_hw_requirements(spec: Spec) -> BaseConstraint:
     return _parse_block(spec)
 
 
+def simplify_actual_hardware(hardware: BaseConstraint) -> Spec:
+    assert isinstance(hardware, And)
+
+    as_spec: Spec = {}
+
+    for constraint in hardware.constraints:
+        constraint_spec = constraint.to_spec()
+
+        assert isinstance(constraint_spec, dict)
+
+        as_spec.update(constraint_spec)
+
+    return as_spec
+
+
 @dataclasses.dataclass
 class Hardware(SpecBasedContainer[Spec, Spec]):
     constraint: Optional[BaseConstraint]

tmt/steps/__init__.py

@@ -29,6 +29,7 @@
 from click.core import ParameterSource
 
 import tmt.export
+import tmt.hardware
 import tmt.log
 import tmt.options
 import tmt.queue
@@ -1908,11 +1909,13 @@ class GuestTopology(SerializableContainer):
     name: str
     role: Optional[str]
     hostname: Optional[str]
+    hardware: Optional[tmt.hardware.Spec]
 
     def __init__(self, guest: 'Guest') -> None:
         self.name = guest.name
         self.role = guest.role
         self.hostname = guest.topology_address
+        self.hardware = tmt.hardware.simplify_actual_hardware(guest.actual_hardware)
 
 
 @dataclasses.dataclass(init=False)
@@ -2103,6 +2106,49 @@ def push(
 
         return environment
 
+    @classmethod
+    def inject(
+            cls,
+            *,
+            environment: Environment,
+            guests: list['Guest'],
+            guest: 'Guest',
+            dirpath: Path,
+            filename_base: Optional[Path] = None,
+            logger: tmt.log.Logger) -> 'Topology':
+        """
+        Create, save and push topology to a given guest.
+
+        .. note::
+
+            A helper for simplified use from plugins. It delivers exactly what
+            :py:meth:`save` and :py:meth:`push` would do, but is easier to use
+            for a common plugin developer.
+
+        :param environment: environment to update with topology variables.
+        :param guests: list of all provisioned guests.
+        :param guest: a guest on which the plugin would run its actions.
+        :param dirpath: a directory to save the topology into.
+        :param filename_base: if set, it would be used as a base for filenames,
+            correct suffixes would be added.
+        :param logger: logger to use for logging.
+        :returns: instantiated topology container.
+        """
+
+        topology = cls(guests)
+        topology.guest = GuestTopology(guest)
+
+        environment.update(
+            topology.push(
+                dirpath=dirpath,
+                guest=guest,
+                filename_base=filename_base,
+                logger=logger
+                )
+            )
+
+        return topology
+
 
 @dataclasses.dataclass
 class ActionTask(tmt.queue.GuestlessTask[None]):

tmt/steps/execute/__init__.py

@@ -25,7 +25,7 @@
 from tmt.steps import Action, ActionTask, PhaseQueue, PluginTask, Step
 from tmt.steps.discover import Discover, DiscoverPlugin, DiscoverStepData
 from tmt.steps.provision import Guest
-from tmt.utils import Path, ShellScript, Stopwatch, cached_property, field
+from tmt.utils import Environment, Path, ShellScript, Stopwatch, cached_property, field
 
 if TYPE_CHECKING:
     import tmt.cli
@@ -240,6 +240,22 @@ def reboot_requested(self) -> bool:
         """ Whether a guest reboot has been requested while the test was running """
         return self.soft_reboot_requested or self.hard_reboot_requested
 
+    def inject_topology(self, environment: Environment) -> tmt.steps.Topology:
+        """
+        Collect and install the guest topology and make it visible to test.
+
+        :param environment: environment to update with topology variables.
+        :returns: instantiated topology container.
+        """
+
+        return tmt.steps.Topology.inject(
+            environment=environment,
+            guests=self.phase.step.plan.provision.guests(),
+            guest=self.guest,
+            dirpath=self.path,
+            logger=self.logger
+            )
+
     def handle_reboot(self) -> bool:
         """
         Reboot the guest if the test requested it.

tmt/steps/execute/internal.py

@@ -321,13 +321,7 @@ def execute(
             options=["-s", "-p", "--chmod=755"])
 
         # Create topology files
-        topology = tmt.steps.Topology(self.step.plan.provision.guests())
-        topology.guest = tmt.steps.GuestTopology(guest)
-
-        environment.update(topology.push(
-            dirpath=invocation.path,
-            guest=guest,
-            logger=logger))
+        invocation.inject_topology(environment)
 
         command: str
         if guest.become and not guest.facts.is_superuser:

tmt/steps/finish/shell.py

@@ -67,13 +67,25 @@ def go(
         """ Perform finishing tasks on given guest """
         super().go(guest=guest, environment=environment, logger=logger)
 
+        environment = environment or tmt.utils.Environment()
+
         # Give a short summary
         overview = fmf.utils.listed(self.data.script, 'script')
         self.info('overview', f'{overview} found', 'green')
 
         workdir = self.step.plan.worktree
         assert workdir is not None  # narrow type
 
+        if not self.is_dry_run:
+            tmt.steps.Topology.inject(
+                environment=environment,
+                guests=self.step.plan.provision.guests(),
+                guest=guest,
+                dirpath=workdir,
+                filename_base=safe_filename(tmt.steps.TEST_TOPOLOGY_FILENAME_BASE, self, guest),
+                logger=logger
+                )
+
         finish_wrapper_filename = safe_filename(FINISH_WRAPPER_FILENAME, self, guest)
         finish_wrapper_path = workdir / finish_wrapper_filename
 

tmt/steps/prepare/shell.py

@@ -75,16 +75,14 @@ def go(
         assert workdir is not None  # narrow type
 
         if not self.is_dry_run:
-            topology = tmt.steps.Topology(self.step.plan.provision.guests())
-            topology.guest = tmt.steps.GuestTopology(guest)
-
-            environment.update(
-                topology.push(
-                    dirpath=workdir,
-                    guest=guest,
-                    logger=logger,
-                    filename_base=safe_filename(tmt.steps.TEST_TOPOLOGY_FILENAME_BASE, self, guest)
-                    ))
+            tmt.steps.Topology.inject(
+                environment=environment,
+                guests=self.step.plan.provision.guests(),
+                guest=guest,
+                dirpath=workdir,
+                filename_base=safe_filename(tmt.steps.TEST_TOPOLOGY_FILENAME_BASE, self, guest),
+                logger=logger
+                )
 
         prepare_wrapper_filename = safe_filename(PREPARE_WRAPPER_FILENAME, self, guest)
         prepare_wrapper_path = workdir / prepare_wrapper_filename

tmt/steps/provision/__init__.py

@@ -687,6 +687,12 @@ def package_manager(self) -> 'tmt.package_managers.PackageManager':
         return tmt.package_managers.find_package_manager(
             self.facts.package_manager)(guest=self, logger=self._logger)
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        """ An actual HW configuration expressed with tmt hardware specification """
+
+        raise NotImplementedError
+
     @classmethod
     def options(cls, how: Optional[str] = None) -> list[tmt.options.ClickOptionDecoratorType]:
         """ Prepare command line options related to guests """
@@ -801,6 +807,19 @@ def show(self, show_multihost_name: bool = True) -> None:
             elif key in GUEST_FACTS_VERBOSE_FIELDS:
                 self.verbose(key_formatted, value_formatted, color='green')
 
+        if self.hardware and self.hardware.constraint:
+            self.info(
+                'hardware',
+                tmt.utils.dict_to_yaml(self.hardware.to_spec()).strip(),
+                color='green')
+
+        self.info(
+            'actual hardware',
+            tmt.utils.dict_to_yaml(
+                tmt.hardware.simplify_actual_hardware(
+                    self.actual_hardware)).strip(),
+            color='green')
+
     def _ansible_verbosity(self) -> list[str]:
         """ Prepare verbose level based on the --debug option count """
         if self.debug_level < 3:

tmt/steps/provision/artemis.py

@@ -498,6 +498,16 @@ def is_ready(self) -> bool:
         #        return True if self.guest is not None
         return self.primary_address is not None
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        return tmt.hardware.parse_hw_requirements(
+            tmt.utils.yaml_to_dict(
+                f"""
+                arch: {self.arch}
+                """
+                )
+            )
+
     def _create(self) -> None:
         environment: dict[str, Any] = {
             'hw': {

tmt/steps/provision/connect.py

@@ -2,6 +2,7 @@
 from typing import Any, Optional, Union
 
 import tmt
+import tmt.hardware
 import tmt.steps
 import tmt.steps.provision
 import tmt.utils
@@ -85,6 +86,16 @@ class GuestConnect(tmt.steps.provision.GuestSsh):
     soft_reboot: Optional[ShellScript]
     hard_reboot: Optional[ShellScript]
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        return tmt.hardware.parse_hw_requirements(
+            tmt.utils.yaml_to_dict(
+                f"""
+                arch: {self.facts.arch}
+                """
+                )
+            )
+
     def reboot(
             self,
             hard: bool = False,

tmt/steps/provision/local.py

@@ -1,8 +1,10 @@
 import dataclasses
+import os
 from typing import Any, Optional, Union
 
 import tmt
 import tmt.base
+import tmt.hardware
 import tmt.log
 import tmt.steps
 import tmt.steps.provision
@@ -26,6 +28,19 @@ def is_ready(self) -> bool:
         """ Local is always ready """
         return True
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        memory = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')
+
+        return tmt.hardware.parse_hw_requirements(
+            tmt.utils.yaml_to_dict(
+                f"""
+                arch: {self.facts.arch}
+                memory: {memory} bytes
+                """
+                )
+            )
+
     def _run_ansible(
             self,
             playbook: Path,

tmt/steps/provision/mrack.py

@@ -714,6 +714,16 @@ def is_ready(self) -> bool:
         except mrack.errors.MrackError:
             return False
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        return tmt.hardware.parse_hw_requirements(
+            tmt.utils.yaml_to_dict(
+                f"""
+                arch: {self.arch}
+                """
+                )
+            )
+
     def _create(self, tmt_name: str) -> None:
         """ Create beaker job xml request and submit it to Beaker hub """
 

tmt/steps/provision/podman.py

@@ -5,6 +5,7 @@
 
 import tmt
 import tmt.base
+import tmt.hardware
 import tmt.log
 import tmt.steps
 import tmt.steps.provision
@@ -114,6 +115,16 @@ def is_ready(self) -> bool:
             ))
         return str(cmd_output.stdout).strip() == 'true'
 
+    @property
+    def actual_hardware(self) -> tmt.hardware.BaseConstraint:
+        return tmt.hardware.parse_hw_requirements(
+            tmt.utils.yaml_to_dict(
+                f"""
+                arch: {self.facts.arch}
+                """
+                )
+            )
+
     def wake(self) -> None:
         """ Wake up the guest """
         self.debug(