Add primitives for loading and saving step results

tmt/steps/__init__.py

@@ -8,7 +8,7 @@
 import re
 import shutil
 import textwrap
-from collections.abc import Iterable, Iterator
+from collections.abc import Iterable, Iterator, Sequence
 from re import Pattern
 from typing import (
     TYPE_CHECKING,
@@ -34,6 +34,7 @@
 import tmt.queue
 import tmt.utils
 from tmt.options import option, show_step_method_hints
+from tmt.result import BaseResult
 from tmt.utils import (
     DEFAULT_NAME,
     Environment,
@@ -231,6 +232,9 @@ class _RawStepData(TypedDict, total=False):
 #: A type variable representing a return value of plugin's ``go()`` method.
 PluginReturnValueT = TypeVar('PluginReturnValueT')
 
+#: A type variable representing a result type.
+ResultT = TypeVar('ResultT', bound=BaseResult)
+
 
 @dataclasses.dataclass
 class StepData(
@@ -664,6 +668,40 @@ def save(self) -> None:
             }
         self.write(Path('step.yaml'), tmt.utils.dict_to_yaml(content))
 
+    def _load_results(
+            self,
+            result_class: type[ResultT],
+            allow_missing: bool = False) -> list[ResultT]:
+        """ Load results of this step from the workdir """
+
+        try:
+            raw_results: list[Any] = tmt.utils.yaml_to_list(self.read(Path('results.yaml')))
+
+            return [
+                result_class.from_serialized(raw_result) for raw_result in raw_results
+                ]
+
+        except tmt.utils.FileError as exc:
+            if allow_missing:
+                self.debug(f'{self.__class__.__name__} results not found.', level=2)
+                return []
+
+            raise GeneralError('Cannot load step results.') from exc
+
+        except Exception as exc:
+            raise GeneralError('Cannot load step results.') from exc
+
+    def _save_results(self, results: Sequence[BaseResult]) -> None:
+        """ Save results of this step to the workdir """
+
+        try:
+            raw_results = [result.to_serialized() for result in results]
+
+            self.write(Path('results.yaml'), tmt.utils.dict_to_yaml(raw_results))
+
+        except Exception as exc:
+            raise GeneralError('Cannot save step results.') from exc
+
     def wake(self) -> None:
         """ Wake up the step (process workdir and command line) """
         # Cleanup possible old workdir if called with --force, but not

tmt/steps/execute/__init__.py

@@ -821,17 +821,14 @@ def __init__(
     def load(self) -> None:
         """ Load test results """
         super().load()
-        try:
-            results = tmt.utils.yaml_to_list(self.read(Path('results.yaml')))
-            self._results = [Result.from_serialized(data) for data in results]
-        except tmt.utils.FileError:
-            self.debug('Test results not found.', level=2)
+
+        self._results = self._load_results(Result, allow_missing=True)
 
     def save(self) -> None:
         """ Save test results to the workdir """
         super().save()
-        results = [result.to_serialized() for result in self.results()]
-        self.write(Path('results.yaml'), tmt.utils.dict_to_yaml(results))
+
+        self._save_results(self.results())
 
     def wake(self) -> None:
         """ Wake up the step (process workdir and command line) """