Add a "collection of results" class for collection-wide actions

tmt/base/run.py

@@ -19,7 +19,6 @@
 import tmt.config
 import tmt.log
 import tmt.policy
-import tmt.result
 import tmt.steps
 import tmt.steps.cleanup
 import tmt.steps.execute
@@ -36,7 +35,7 @@
     field,
 )
 from tmt.recipe import RecipeManager
-from tmt.result import Result
+from tmt.result import Results
 from tmt.utils import (
     Command,
     Environment,
@@ -551,10 +550,10 @@ def finish(self) -> None:
         interesting_results = execute.enabled or report.enabled
 
         # Gather all results and give an overall summary
-        results = [result for plan in self.plans for result in plan.execute.results()]
+        results = Results(result for plan in self.plans for result in plan.execute.results())
         if interesting_results:
             self.info('')
-            self.info('total', Result.summary(results), color='cyan')
+            self.info('total', results.summary(), color='cyan')
 
         # Remove the workdir if enabled
         if self.remove and self.plans[0].cleanup.enabled:
@@ -570,7 +569,7 @@ def finish(self) -> None:
             raise SystemExit(0)
 
         # Return appropriate exit code based on the total stats
-        raise SystemExit(tmt.result.results_to_exit_code(results, bool(execute.enabled)))
+        raise SystemExit(results.to_exit_code(execute_enabled=bool(execute.enabled)))
 
     def follow(self) -> None:
         """

tmt/result.py

@@ -1,5 +1,5 @@
 import enum
-from typing import TYPE_CHECKING, Any, Callable, Optional, cast
+from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar, cast
 
 import fmf.utils
 
@@ -13,10 +13,14 @@
 
 if TYPE_CHECKING:
     import tmt.base.core
+    import tmt.cli
     import tmt.guest
     import tmt.steps.execute
 
 
+ResultT = TypeVar('ResultT', bound='BaseResult')
+
+
 class ResultOutcome(enum.Enum):
     PASS = 'pass'
     FAIL = 'fail'
@@ -512,25 +516,43 @@ def to_subresult(self) -> 'SubResult':
             {option: value for option, value in self.to_serialized().items() if option in options}
         )
 
-    @staticmethod
-    def total(results: list['Result']) -> dict[ResultOutcome, int]:
+    @property
+    def failure_logs(self) -> list[Path]:
         """
-        Return dictionary with total stats for given results
+        Return paths to all failure logs from the result
+        """
+
+        failure_logs = super().failure_logs
+        for check in self.check:
+            failure_logs += check.failure_logs
+        return list(set(failure_logs))
+
+
+class Results(list[ResultT]):
+    """
+    A collection of results.
+
+    Effectively a fancy list of results, with a few helper methods for
+    the collection as a whole.
+    """
+
+    def total(self) -> dict[ResultOutcome, int]:
+        """
+        Return dictionary with total stats.
         """
 
         stats = dict.fromkeys(RESULT_OUTCOME_COLORS, 0)
 
-        for result in results:
+        for result in self:
             stats[result.result] += 1
         return stats
 
-    @staticmethod
-    def summary(results: list['Result']) -> str:
+    def summary(self) -> str:
         """
-        Prepare a nice human summary of provided results
+        Prepare a nice human summary of results.
         """
 
-        stats = Result.total(results)
+        stats = self.total()
         comments = []
         if stats.get(ResultOutcome.PASS):
             passed = ' ' + style('passed', fg='green')
@@ -556,60 +578,53 @@ def summary(results: list['Result']) -> str:
         # FIXME: cast() - https://github.com/teemtee/fmf/issues/185
         return cast(str, fmf.utils.listed(comments or ['no results found']))
 
-    @property
-    def failure_logs(self) -> list[Path]:
-        """
-        Return paths to all failure logs from the result
+    def to_exit_code(self, execute_enabled: bool = True) -> 'tmt.cli.TmtExitCode':
         """
+        Map results to a tmt exit code.
 
-        failure_logs = super().failure_logs
-        for check in self.check:
-            failure_logs += check.failure_logs
-        return list(set(failure_logs))
-
-
-def results_to_exit_code(results: list[Result], execute_enabled: bool = True) -> int:
-    """
-    Map results to a tmt exit code
-    """
+        :param execute_enabled: if set, the :py:mod:`tmt.steps.execute`
+            step was enabled.
+        :raises tmt.utils.GeneralError: when it was not possible to map
+            results to any defined exit code.
+        """
 
-    from tmt.cli import TmtExitCode
+        from tmt.cli import TmtExitCode
 
-    stats = Result.total(results)
+        stats = self.total()
 
-    # Quoting the specification:
+        # Quoting the specification:
 
-    # "No test results found."
-    if sum(stats.values()) == 0:
-        return TmtExitCode.NO_RESULTS_FOUND
+        # "No test results found."
+        if sum(stats.values()) == 0:
+            return TmtExitCode.NO_RESULTS_FOUND
 
-    # "Errors occurred during test execution."
-    if stats[ResultOutcome.ERROR]:
-        return TmtExitCode.ERROR
+        # "Errors occurred during test execution."
+        if stats[ResultOutcome.ERROR]:
+            return TmtExitCode.ERROR
 
-    # "There was a fail or warn identified, but no error."
-    if stats[ResultOutcome.FAIL] + stats[ResultOutcome.WARN]:
-        return TmtExitCode.FAIL
+        # "There was a fail or warn identified, but no error."
+        if stats[ResultOutcome.FAIL] + stats[ResultOutcome.WARN]:
+            return TmtExitCode.FAIL
 
-    # "Tests were executed, and all reported the ``skip`` result."
-    if sum(stats.values()) == stats[ResultOutcome.SKIP]:
-        return TmtExitCode.ALL_TESTS_SKIPPED
+        # "Tests were executed, and all reported the ``skip`` result."
+        if sum(stats.values()) == stats[ResultOutcome.SKIP]:
+            return TmtExitCode.ALL_TESTS_SKIPPED
 
-    # "No errors or fails, but there are pending tests."
-    if execute_enabled and stats[ResultOutcome.PENDING]:
-        return TmtExitCode.ERROR
+        # "No errors or fails, but there are pending tests."
+        if execute_enabled and stats[ResultOutcome.PENDING]:
+            return TmtExitCode.ERROR
 
-    # "At least one test passed, there was no fail, warn or error."
-    if (
-        sum(stats.values())
-        == stats[ResultOutcome.PASS]
-        + stats[ResultOutcome.INFO]
-        + stats[ResultOutcome.SKIP]
-        + stats[ResultOutcome.PENDING]
-    ):
-        return TmtExitCode.SUCCESS
+        # "At least one test passed, there was no fail, warn or error."
+        if (
+            sum(stats.values())
+            == stats[ResultOutcome.PASS]
+            + stats[ResultOutcome.INFO]
+            + stats[ResultOutcome.SKIP]
+            + stats[ResultOutcome.PENDING]
+        ):
+            return TmtExitCode.SUCCESS
 
-    raise GeneralError("Unhandled combination of test result.")
+        raise GeneralError("Unhandled combination of test result.")
 
 
 def save_failures(

tmt/steps/__init__.py

@@ -57,7 +57,7 @@
     simple_field,
 )
 from tmt.options import ClickOptionDecoratorType, option
-from tmt.result import ResultOutcome
+from tmt.result import BaseResult, ResultOutcome, Results
 from tmt.utils import (
     DEFAULT_NAME,
     Command,
@@ -1030,7 +1030,7 @@ def _load_results(
         self,
         result_class: type[ResultT],
         allow_missing: bool = False,
-    ) -> list[ResultT]:
+    ) -> tmt.result.Results[ResultT]:
         """
         Load results of this step from the workdir
         """
@@ -1040,19 +1040,19 @@ def _load_results(
         try:
             raw_results: list[Any] = self.plan.my_run.read_state(self.step_workdir / 'results')
 
-            return [result_class.from_serialized(raw_result) for raw_result in raw_results]
+            return Results(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 []
+                return Results()
 
             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:
+    def _save_results(self, results: Sequence[BaseResult]) -> None:
         """
         Save results of this step to the workdir
         """

tmt/steps/execute/__init__.py

@@ -30,6 +30,7 @@
     ResultGuestData,
     ResultInterpret,
     ResultOutcome,
+    Results,
 )
 from tmt.steps import Action, ActionTask, PluginTask, Step
 from tmt.steps.context.abort import AbortContext, AbortStep
@@ -750,19 +751,19 @@ def prepare_tests(self, guest: Guest, logger: tmt.log.Logger) -> list[TestInvoca
             if self.should_run_again:
                 assert self.parent is not None  # narrow type
                 assert isinstance(self.parent, Execute)  # narrow type
-                self.parent._results = [
+                self.parent._results = Results(
                     result
                     for result in self.parent._results
                     if not (
                         test.name == result.name and test.serial_number == result.serial_number
                     )
-                ]
+                )
 
         # Keep old results in another variable to have numbers only for actually executed tests
         if self.should_run_again:
             assert self.parent is not None  # narrow type
             assert isinstance(self.parent, Execute)  # narrow type
-            self.parent._old_results = self.parent._results[:]
+            self.parent._old_results = Results(self.parent._results[:])
             self.parent._results.clear()
 
         return invocations
@@ -1079,8 +1080,8 @@ def __init__(
 
         super().__init__(plan=plan, raw_data=raw_data, logger=logger)
         # List of Result() objects representing test results
-        self._results: list[tmt.Result] = []
-        self._old_results: list[tmt.Result] = []
+        self._results: Results[Result] = Results()
+        self._old_results: Results[Result] = Results()
 
     @property
     def _preserved_workdir_members(self) -> set[str]:
@@ -1197,17 +1198,17 @@ def update_results(self, results: list['Result']) -> None:
             # Replace existing pending result with the new one.
             results_to_save[(result.serial_number, result.name, result.guest.name)] = result
 
-        self._results = list(results_to_save.values())
+        self._results = Results(results_to_save.values())
 
-    def create_results(self, tests: list['tmt.steps.discover.TestOrigin']) -> list['Result']:
+    def create_results(self, tests: list['tmt.steps.discover.TestOrigin']) -> Results[Result]:
         """
         Get all available results from tests. For tests not yet executed, create a pending
         result.
         """
 
         guests = self.plan.provision.get_guests_info()
 
-        results = []
+        results: Results[Result] = Results()
         for result, test_origin in self.results_for_tests(tests):
             if result:
                 results.append(result)
@@ -1327,7 +1328,7 @@ def go(self, force: bool = False) -> None:
 
         self._assert_required_tests_executed()
 
-    def results(self) -> list["tmt.result.Result"]:
+    def results(self) -> 'tmt.result.Results[tmt.result.Result]':
         """
         Results from executed tests
 

tmt/steps/execute/upgrade.py

@@ -10,6 +10,7 @@
 import tmt.steps
 import tmt.utils
 from tmt.container import container, field, key_to_option
+from tmt.result import Results
 from tmt.steps.discover import Discover, DiscoverPlugin, DiscoverStepData, normalize_ref
 from tmt.steps.discover.fmf import (
     DiscoverFmf,
@@ -567,7 +568,7 @@ def _remove_old_results(self, prefix: str) -> None:
             if result.name.startswith(f'/{prefix}/')
         ]
 
-        self.step.plan.execute._results = [
+        self.step.plan.execute._results = Results(
             result for result in results if result.name not in old_result_names
-        ]
+        )
         self.step.plan.execute.save()

tmt/steps/prepare/__init__.py

@@ -8,7 +8,7 @@
 from tmt.container import container, simple_field
 from tmt.guest import Guest
 from tmt.plugins import PluginRegistry
-from tmt.result import PhaseResult, ResultGuestData, ResultOutcome
+from tmt.result import PhaseResult, ResultGuestData, ResultOutcome, Results
 from tmt.steps import (
     Action,
     PluginOutcome,
@@ -115,7 +115,7 @@ class Prepare(tmt.steps.StepWithQueue[PrepareStepData, PluginOutcome]):
 
     _plugin_base_class = PreparePlugin
 
-    results: list[PhaseResult]
+    results: Results[PhaseResult]
 
     @property
     def _preserved_workdir_members(self) -> set[str]:
@@ -145,7 +145,7 @@ def __init__(
 
         super().__init__(plan=plan, raw_data=raw_data, logger=logger)
 
-        self.results = []
+        self.results = Results()
         self.preparations_applied = 0
 
     def load(self) -> None:
@@ -403,7 +403,7 @@ def _emit_phase(
                     ],
                 )
 
-        self.results: list[PhaseResult] = []
+        self.results: Results[PhaseResult] = Results()
         exceptions: list[Exception] = []
 
         def _record_exception(

tmt/steps/report/__init__.py

@@ -1,7 +1,6 @@
 from typing import Optional, TypeVar, Union, cast
 
 import tmt.log
-import tmt.result
 import tmt.steps
 from tmt.container import container
 from tmt.plugins import PluginRegistry
@@ -79,7 +78,7 @@ def summary(self) -> None:
         Give a concise report summary
         """
 
-        summary = tmt.result.Result.summary(self.plan.execute.results())
+        summary = self.plan.execute.results().summary()
         self.info('summary', summary, 'green', shift=1)
 
     def go(self, force: bool = False) -> None: