feat: add fwts test to imgtests lib

src/imgtests/exec/loaders/__init__.py

@@ -3,6 +3,7 @@
 from .fio import Fio as Fio
 from .fio import FioPlot as FioPlot
 from .fwts import Fwts as Fwts
+from .fwts import FwtsResult as FwtsResult
 from .iperf3 import Iperf3 as Iperf3
 from .iperf3 import Iperf3Bundle as Iperf3Bundle
 from .kirk import Kirk as Kirk

src/imgtests/exec/loaders/fwts.py

@@ -1,14 +1,126 @@
 import logging
-from typing import TYPE_CHECKING
+import re
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 from imgtests.exec.base_util import GenericUtil
+from imgtests.results_adapter import AdapterResult
+from imgtests.types import MetricSample
 
 if TYPE_CHECKING:
-    from imgtests.exec.exec import SSHClient
+    from imgtests.exec.exec import ExecResult, SSHClient
 
 logger = logging.getLogger(__name__)
 
 
+class FwtsResult(NamedTuple):
+    tests: list[dict[str, Any]]
+    summary: dict[str, int]
+
+
 class Fwts(GenericUtil):
     def __init__(self, ssh_client: SSHClient | None = None) -> None:
         super().__init__("fwts", ssh_client)
+
+    def run(self) -> tuple[ExecResult, FwtsResult]:
+        result = self()
+        parsed = (
+            self.parse_metrics(result.stdout)
+            if result.stdout
+            else FwtsResult(
+                tests=[],
+                summary={"passed": 0, "failed": 0, "skipped": 0, "aborted": 0},
+            )
+        )
+        return result, parsed
+
+    @staticmethod
+    def parse_metrics(raw_output: str) -> FwtsResult:  # noqa: PLR0912, C901
+        statuses: dict[str, int] = {"passed": 0, "failed": 0, "skipped": 0, "aborted": 0}
+        tests: list[dict[str, Any]] = []
+
+        sections = re.split(r"^Test: ", raw_output, flags=re.MULTILINE)
+
+        for section in sections[1:]:
+            lines = section.splitlines()
+            test_name = lines[0].strip().rstrip(".")
+            subtotal: dict[str, int] = {"passed": 0, "failed": 0, "skipped": 0, "aborted": 0}
+
+            for line in lines[1:]:
+                stripped = line.strip().lower()
+
+                if stripped in ("test aborted", "test aborted."):
+                    subtotal["aborted"] += 1
+                elif stripped in ("test skipped", "test skipped."):
+                    subtotal["skipped"] += 1
+                else:
+                    for match in re.finditer(
+                        r"(\d+)?\s+(passed|failed|skipped|aborted|info only)",
+                        stripped.lower(),
+                    ):
+                        if match.group(1) is not None:
+                            count = int(match.group(1))
+                            status = match.group(2)
+                            if status == "info only":
+                                """
+                                Info only tests are tests that just check some system info, e.g.
+                                Gather kernel system information.                       1 info only
+                                count as passed
+                                """
+                                subtotal["passed"] += count
+                            elif status in subtotal:
+                                subtotal[status] += count
+                        else:
+                            status = match.group(2)
+                            if status == "info only":
+                                subtotal["passed"] += 1
+                            elif status in subtotal:
+                                subtotal[status] += 1
+
+            if all(v == 0 for v in subtotal.values()):
+                subtotal["skipped"] += 1
+
+            for key, value in subtotal.items():
+                statuses[key] += value
+            tests.append({"name": test_name, "subtests": subtotal})
+
+        return FwtsResult(tests=tests, summary=statuses)
+
+    @staticmethod
+    def metrics_to_json(metrics: FwtsResult) -> AdapterResult:
+        raw: dict[str, Any] = {"fwts_tests": metrics.tests, "fwts_summary": metrics.summary}
+        return Fwts.split_result(raw_metrics=raw)
+
+    @staticmethod
+    def split_result(raw_metrics: dict[str, Any]) -> AdapterResult:
+        if not raw_metrics:
+            return AdapterResult(tool="fwts", test_type={}, time={}, metrics={})
+
+        tests = raw_metrics.get("fwts_tests", [])
+        summary = raw_metrics.get("fwts_summary", {})
+
+        test_type = {"type": "firmware"}
+        time = {}
+        metrics: dict[str, Any] = {str(i): t for i, t in enumerate(tests)}
+        metrics["summary"] = summary
+
+        return AdapterResult(tool="fwts", test_type=test_type, time=time, metrics=metrics)
+
+
+def fwts_metrics_to_samples(
+    stage_name: str,
+    subsystem: str,
+    metrics: dict[str, Any],
+) -> list[MetricSample]:
+    samples: list[MetricSample] = []
+    summary = metrics.get("summary", {})
+    for key, label in (
+        ("passed", "Passed"),
+        ("failed", "Failed"),
+        ("skipped", "Skipped"),
+        ("aborted", "Aborted"),
+    ):
+        value = summary.get(key, 0)
+        samples.append(
+            MetricSample(stage_name, subsystem, f"fwts.{key}", float(value), label=label),
+        )
+    return samples

src/imgtests/suites/firmware.py

@@ -0,0 +1,68 @@
+from typing import TYPE_CHECKING
+
+from imgtests.exec.loaders import Fwts
+from imgtests.planning import AbstractRunnableManyTimesTest
+from imgtests.types import Subsystem, TestResult, TestStatus
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+    from concurrent.futures import ThreadPoolExecutor
+
+    from imgtests.exec.exec import SSHClient
+
+
+class FwtsTest(AbstractRunnableManyTimesTest):
+    def __init__(
+        self,
+        iterations: int = 1,
+    ) -> None:
+        super().__init__("FWTS firmware tests.", frozenset({Subsystem.SYSTEM}), iterations)
+
+    def _run(
+        self,
+        executor: ThreadPoolExecutor,
+        client: SSHClient | None,
+        iterations: int,
+    ) -> Iterable[TestResult]:
+        fwts = Fwts(client)
+
+        for _ in range(iterations):
+            future = executor.submit(fwts.run)
+            result, parsed = future.result()
+
+            if result.returncode and not parsed.tests:
+                self.logger.error("FWTS test BROKEN (returncode=%d)", result.returncode)
+                yield TestResult(status=TestStatus.BROKEN)
+                continue
+
+            tests_passed = parsed.summary.get("passed", 0)
+            tests_failed = parsed.summary.get("failed", 0)
+            tests_skipped = parsed.summary.get("skipped", 0)
+            tests_aborted = parsed.summary.get("aborted", 0)
+
+            if tests_failed > 0:
+                self.logger.error(
+                    "FWTS test FAILED (%d passed, %d failed, %d skipped, %d aborted)",
+                    tests_passed,
+                    tests_failed,
+                    tests_skipped,
+                    tests_aborted,
+                )
+                yield TestResult(
+                    status=TestStatus.FAILED,
+                    metrics=fwts.metrics_to_json(parsed),
+                    command="fwts",
+                )
+            else:
+                self.logger.info(
+                    "FWTS test PASSED (%d passed, %d failed, %d skipped, %d aborted)",
+                    tests_passed,
+                    tests_failed,
+                    tests_skipped,
+                    tests_aborted,
+                )
+                yield TestResult(
+                    status=TestStatus.PASSED,
+                    metrics=fwts.metrics_to_json(parsed),
+                    command="fwts",
+                )