Support fetching results from ReportPortal

tmt-recipe-tool/README.md

@@ -2,7 +2,9 @@
 
 A command-line tool for filtering and rerunning [tmt](https://tmt.readthedocs.io/) tests based on result outcomes.
 
-Given a [tmt recipe](https://tmt.readthedocs.io/en/stable/spec/recipe.html) and its associated [test results](https://tmt.readthedocs.io/en/stable/spec/results.html), `tmt-recipe-tool` can produce a new recipe containing only the tests that matched specific outcomes (e.g. failed or errored tests), and optionally rerun them immediately.
+Given a [tmt recipe](https://tmt.readthedocs.io/en/stable/spec/recipe.html) and its associated test results, `tmt-recipe-tool` can produce a new recipe containing only the tests that matched specific outcomes (e.g. failed or errored tests), and optionally rerun them immediately.
+
+Results can be sourced either from a local tmt [results file](https://tmt.readthedocs.io/en/stable/spec/results.html) or from a [ReportPortal](https://reportportal.io/) instance when the recipe's report phase is configured with `how: reportportal`.
 
 ## Requirements
 
@@ -44,12 +46,29 @@ tmt-recipe-tool [OPTIONS] COMMAND [ARGS]...
 Filter recipe tests by their result outcome, keeping only those that match.
 
 ```
-tmt-recipe-tool -i RECIPE filter-tests [--result RESULT]...
+tmt-recipe-tool -i RECIPE filter-tests [--use-reportportal] [--result RESULT]...
 ```
 
 | Option | Default | Description |
 |--------|---------|-------------|
 | `--result RESULT` | `fail`, `error`, `warn` | Keep tests with this outcome (repeatable) |
+| `--use-reportportal` | `false` | Fetch test results from ReportPortal instead of a local results file |
+
+When `--use-reportportal` is used, results are fetched from any report phase in the recipe that has `how: reportportal`. The phase must include `launch-uuid` and `test-uuids` (populated automatically by the tmt [ReportPortal](https://tmt.readthedocs.io/en/stable/plugins/report/reportportal.html) plugin after a run finishes). The `launch-uuid` and `test-uuids` fields are stripped from the output recipe so that a subsequent run creates a new ReportPortal launch. Plans that do not have a `reportportal` report phase always fall back to their local `results.yaml` file, even when `--use-reportportal` is passed.
+
+Because ReportPortal only has three result statuses (`PASSED`, `FAILED`, `SKIPPED`), tmt outcomes are mapped before filtering:
+
+| tmt outcome | ReportPortal status |
+|-------------|---------------------|
+| `pass` | `PASSED` |
+| `fail` | `FAILED` |
+| `warn` | `FAILED` |
+| `error` | `FAILED` |
+| `info` | `SKIPPED` |
+| `skip` | `SKIPPED` |
+| `pending` | `SKIPPED` |
+
+As a result, `fail`, `warn`, and `error` are indistinguishable when filtering via ReportPortal, and all three will select tests with status `FAILED`. Similarly, `info`, `skip`, and `pending` will all select tests with status `SKIPPED`.
 
 ### Examples
 
@@ -82,3 +101,15 @@ Combine multiple result filters:
 ```bash
 tmt-recipe-tool -i recipe.yaml -o subset.yaml filter-tests --result fail --result error
 ```
+
+Filter using ReportPortal results (requires the recipe to have a `reportportal` report phase with `launch-uuid` and `test-uuids`):
+
+```bash
+tmt-recipe-tool -i recipe.yaml -o filtered.yaml filter-tests --use-reportportal
+```
+
+Fetch failures from ReportPortal and rerun them immediately:
+
+```bash
+tmt-recipe-tool -i recipe.yaml --run filter-tests --use-reportportal --result fail
+```

tmt-recipe-tool/pyproject.toml

@@ -12,8 +12,9 @@ license = "MIT"
 dependencies = [
     "click>=8.0.3",
     "pydantic>=2.0",
+    "requests>=2.25.1",
     "ruamel.yaml>=0.16.6",
-    "tmt>=1.72",
+    "tmt>=1.73",
 ]
 
 [project.scripts]
@@ -58,6 +59,7 @@ ignore = [
     "S101",     # use of `assert` detected
     "PLR2004",  # magic value used in comparison
     "PLR0913",  # Too many arguments in function definition
+    "UP007",    # X | Y for type annotations
 ]
 
 [dependency-groups]

tmt-recipe-tool/tmt_recipe_tool/cli.py

@@ -81,15 +81,24 @@ def main(
         "Can be specified multiple times."
     ),
 )
+@click.option(
+    "--use-reportportal",
+    is_flag=True,
+    default=False,
+    help="Fetch test results from ReportPortal instead of a local results file.",
+)
 @click.pass_context
-def filter_tests(ctx: click.Context, results: list[str], **kwargs: Any) -> None:
+def filter_tests(
+    ctx: click.Context, results: list[str], use_reportportal: bool, **kwargs: Any
+) -> None:
     """Filter recipe tests by their result outcome."""
     assert ctx.parent is not None
 
     ctx.obj["recipe"] = filter_recipe(
         ctx.parent.params["input"],
         results,
         run_workdir=ctx.parent.params.get("run_workdir"),
+        use_reportportal=use_reportportal,
     )
 
 

tmt-recipe-tool/tmt_recipe_tool/models.py

@@ -1,9 +1,29 @@
 from __future__ import annotations
 
+from typing import Literal, Optional
+
 from pydantic import BaseModel, Field
 
 
 class Result(BaseModel, extra="allow"):
     name: str
     serial_number: int = Field(alias="serial-number")
     result: str
+
+
+class ReportPortalPhase(BaseModel, extra="allow"):
+    name: str
+    how: Literal["reportportal"]
+    url: str
+    project: str
+    token: str
+    launch_uuid: Optional[str] = Field(None, alias="launch-uuid")
+    launch_url: Optional[str] = Field(None, alias="launch-url")
+    ssl_verify: bool = Field(True, alias="ssl-verify")
+    test_uuids: dict[int, dict[str, str]] = Field(default_factory=dict, alias="test-uuids")
+    api_version: str = Field(alias="api-version")
+
+
+class ReportPortalResult(BaseModel, extra="allow"):
+    uuid: str
+    status: str

tmt-recipe-tool/tmt_recipe_tool/recipe.py

@@ -3,9 +3,12 @@
 from typing import Optional, cast
 
 import tmt
+import tmt.recipe
+import tmt.utils
 from pydantic import ValidationError
 
 from tmt_recipe_tool.models import Result
+from tmt_recipe_tool.reportportal import edit_rp_phases, filter_tests_from_rp, get_rp_phases
 from tmt_recipe_tool.utils import create_tmt_logger, load_yaml
 
 
@@ -85,15 +88,25 @@ def filter_recipe(
     input_path: Path,
     filter_results: list[str],
     run_workdir: Optional[Path] = None,
+    use_reportportal: bool = False,
 ) -> tmt.recipe.Recipe:
     """Load a recipe and keep only tests matching the specified result outcomes."""
     recipe = _load_recipe(input_path)
 
     filtered_plans = []
     for plan in recipe.plans:
-        if plan.discover.tests:
-            results_path = _resolve_results_path(plan, input_path, run_workdir)
-            results = _load_results(results_path, plan.name)
+        if not plan.discover.tests:
+            continue
+        rp_phases = get_rp_phases(plan)
+        if rp_phases and use_reportportal:
+            plan.discover.tests = list(
+                filter_tests_from_rp(plan.discover.tests, rp_phases, filter_results)
+            )
+            plan.report.phases = edit_rp_phases(plan.report.phases)
+        else:
+            results = _load_results(
+                _resolve_results_path(plan, input_path, run_workdir), plan.name
+            )
             plan.discover.tests = list(_filter_tests(plan.discover.tests, results, filter_results))
         filtered_plans.append(plan)
 

tmt-recipe-tool/tmt_recipe_tool/reportportal.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import requests  # type: ignore[import-untyped]
+from tmt.recipe import _RecipePlan, _RecipeTest
+from tmt.result import ResultOutcome
+from tmt.steps import _RawStepData
+from tmt.utils import retry_session
+
+from tmt_recipe_tool.models import ReportPortalPhase, ReportPortalResult
+from tmt_recipe_tool.utils import create_tmt_logger
+
+STATUS_FORCELIST = (
+    429,  # Too Many Requests
+    500,  # Internal Server Error
+    502,  # Bad Gateway
+    503,  # Service Unavailable
+    504,  # Gateway Timeout
+)
+
+TMT_TO_RP_RESULT_STATUS = {
+    ResultOutcome.PASS.value: "PASSED",
+    ResultOutcome.FAIL.value: "FAILED",
+    ResultOutcome.INFO.value: "SKIPPED",
+    ResultOutcome.WARN.value: "FAILED",
+    ResultOutcome.ERROR.value: "FAILED",
+    ResultOutcome.SKIP.value: "SKIPPED",
+    ResultOutcome.PENDING.value: "SKIPPED",
+}
+
+
+class ReportPortalError(Exception):
+    """Raised when fetching results from ReportPortal fails."""
+
+
+def _handle_response(response: requests.Response) -> None:
+    """
+    Check the endpoint response and raise an exception if needed
+    """
+    if not response.ok:
+        raise ReportPortalError(
+            f"Received non-ok status code {response.status_code} "
+            f"from ReportPortal: {response.text}"
+        )
+
+
+def get_rp_phases(plan: _RecipePlan) -> list[ReportPortalPhase]:
+    """Get the ReportPortal phases from the plan"""
+    return [
+        ReportPortalPhase.model_validate(phase)
+        for phase in plan.report.phases
+        if phase.get("how", None) == "reportportal"
+    ]
+
+
+def edit_rp_phases(phases: Iterable[_RawStepData]) -> list[_RawStepData]:
+    """Edit the ReportPortal phases so they don't overwrite previous launches"""
+    filtered_phases = []
+    for phase in phases:
+        if phase.get("how", None) == "reportportal":
+            phase.pop("launch-url", None)  # type: ignore[typeddict-item]
+            phase.pop("launch-uuid", None)  # type: ignore[typeddict-item]
+            phase.pop("suite-uuid", None)  # type: ignore[typeddict-item]
+            phase.pop("test-uuids", None)  # type: ignore[typeddict-item]
+        filtered_phases.append(phase)
+    return filtered_phases
+
+
+def _get_launch_id(
+    session: requests.Session,
+    headers: dict[str, str],
+    api_version: str,
+    rp_phase: ReportPortalPhase,
+) -> int:
+    """Resolve the numeric launch ID from launch-uuid"""
+    if not rp_phase.launch_uuid:
+        raise ReportPortalError(f"ReportPortal phase '{rp_phase.name}' has missing 'launch-uuid'.")
+    base_url = rp_phase.url.rstrip("/")
+    url = f"{base_url}/api/{api_version}/{rp_phase.project}/launch/uuid/{rp_phase.launch_uuid}"
+
+    response = session.get(url, headers=headers)
+    _handle_response(response)
+    return int(response.json()["id"])
+
+
+def _fetch_rp_results(
+    session: requests.Session,
+    headers: dict[str, str],
+    api_version: str,
+    rp_phase: ReportPortalPhase,
+    launch_id: int,
+) -> dict[str, ReportPortalResult]:
+    """Fetch the ReportPortal results for a given launch ID"""
+    base_url = rp_phase.url.rstrip("/")
+    results: list[ReportPortalResult] = []
+    page = 1
+    while True:
+        response = session.get(
+            f"{base_url}/api/{api_version}/{rp_phase.project}/item",
+            headers=headers,
+            params={
+                "filter.eq.launchId": launch_id,
+                "filter.eq.type": "STEP",
+                "page.size": 100,
+                "page.page": page,
+            },
+        )
+        _handle_response(response)
+        data = response.json()
+        results.extend(data.get("content", []))
+        page_info = data.get("page", {})
+        if page >= page_info.get("totalPages", 1):
+            break
+        page += 1
+    return {result.uuid: ReportPortalResult.model_validate(result) for result in results}
+
+
+def filter_tests_from_rp(
+    tests: list[_RecipeTest],
+    rp_phases: list[ReportPortalPhase],
+    filter_results: list[str],
+) -> Iterable[_RecipeTest]:
+    """Filter the tests from the ReportPortal results"""
+    filter_results = list(
+        {TMT_TO_RP_RESULT_STATUS.get(result, result) for result in filter_results}
+    )
+
+    filtered_tests: dict[int, _RecipeTest] = {}
+
+    for phase in rp_phases:
+        with retry_session(
+            status_forcelist=STATUS_FORCELIST,
+            logger=create_tmt_logger(),
+        ) as session:
+            session.verify = phase.ssl_verify
+            headers = {
+                "Authorization": f"Bearer {phase.token}",
+                "Accept": "*/*",
+                "Content-Type": "application/json",
+            }
+
+            rp_results = _fetch_rp_results(
+                session=session,
+                headers=headers,
+                api_version=phase.api_version,
+                rp_phase=phase,
+                launch_id=_get_launch_id(session, headers, phase.api_version, phase),
+            )
+
+        serial_number_to_uuids: dict[int, list[str]] = {
+            serial: list(value.values()) for serial, value in phase.test_uuids.items()
+        }
+
+        for test in tests:
+            # Test can have multiple results with the same serial-number
+            uuids = serial_number_to_uuids.get(test.serial_number, [])
+            for uuid in uuids:
+                result = rp_results.get(uuid, None)
+                if not result:
+                    continue
+                if result.status in filter_results:
+                    filtered_tests[test.serial_number] = test
+                    break
+
+    return filtered_tests.values()