diff --git a/src/imgtests/exec/loaders/__init__.py b/src/imgtests/exec/loaders/__init__.py index 1a17da9e..dc8ed359 100644 --- a/src/imgtests/exec/loaders/__init__.py +++ b/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 diff --git a/src/imgtests/exec/loaders/fwts.py b/src/imgtests/exec/loaders/fwts.py index e8cc9804..b805f23b 100644 --- a/src/imgtests/exec/loaders/fwts.py +++ b/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 diff --git a/src/imgtests/suites/firmware.py b/src/imgtests/suites/firmware.py new file mode 100644 index 00000000..282810d5 --- /dev/null +++ b/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", + ) diff --git a/tests/image/endurance/__init__.py b/tests/image/endurance/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/tests/image/endurance/firmware.py b/tests/image/endurance/firmware.py deleted file mode 100644 index 25d2e799..00000000 --- a/tests/image/endurance/firmware.py +++ /dev/null @@ -1,27 +0,0 @@ -import logging -from typing import TYPE_CHECKING - -from imgtests.exec.loaders import Fwts - -if TYPE_CHECKING: - from concurrent.futures import ThreadPoolExecutor - - from imgtests.exec.exec import SSHClient - - -logger = logging.getLogger(__name__) - - -def test_fwts(executor: ThreadPoolExecutor, client: SSHClient | None) -> None: - fwts = Fwts(client) - future = executor.submit(fwts) - r = future.result() - if r.returncode: - logger.error("FWTS endurance test FAILED") - else: - logger.info("FWTS endurance test PASSED") - - if r.stdout: - logger.info("fwts stdout:\n%s", r.stdout) - if r.stderr: - logger.info("fwts stderr:\n%s", r.stderr) diff --git a/tests/unit/imgtests/exec/loaders/test_fwts.py b/tests/unit/imgtests/exec/loaders/test_fwts.py new file mode 100644 index 00000000..9b355c91 --- /dev/null +++ b/tests/unit/imgtests/exec/loaders/test_fwts.py @@ -0,0 +1,237 @@ +from textwrap import dedent + +import pytest + +from imgtests.exec.loaders.fwts import Fwts, FwtsResult + + +@pytest.mark.parametrize( + ("raw_output", "expected"), + [ + ( + "", + FwtsResult( + tests=[], + summary={"passed": 0, "failed": 0, "skipped": 0, "aborted": 0}, + ), + ), + ( + "Test: ACPI DSDT\n1 passed\n", + FwtsResult( + tests=[ + { + "name": "ACPI DSDT", + "subtests": {"passed": 1, "failed": 0, "skipped": 0, "aborted": 0}, + }, + ], + summary={"passed": 1, "failed": 0, "skipped": 0, "aborted": 0}, + ), + ), + ( + dedent("""\ + Test: APIC + 2 passed + 1 failed + """), + FwtsResult( + tests=[ + { + "name": "APIC", + "subtests": {"passed": 2, "failed": 1, "skipped": 0, "aborted": 0}, + }, + ], + summary={"passed": 2, "failed": 1, "skipped": 0, "aborted": 0}, + ), + ), + ( + "Test: MADT\nTest aborted\n", + FwtsResult( + tests=[ + { + "name": "MADT", + "subtests": {"passed": 0, "failed": 0, "skipped": 0, "aborted": 1}, + }, + ], + summary={"passed": 0, "failed": 0, "skipped": 0, "aborted": 1}, + ), + ), + ( + "Test: MADT\nTest aborted.\n", + FwtsResult( + tests=[ + { + "name": "MADT", + "subtests": {"passed": 0, "failed": 0, "skipped": 0, "aborted": 1}, + }, + ], + summary={"passed": 0, "failed": 0, "skipped": 0, "aborted": 1}, + ), + ), + ( + "Test: UEFI\nTest skipped\n", + FwtsResult( + tests=[ + { + "name": "UEFI", + "subtests": {"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + }, + ], + summary={"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + ), + ), + ( + "Test: UEFI\nTest skipped.\n", + FwtsResult( + tests=[ + { + "name": "UEFI", + "subtests": {"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + }, + ], + summary={"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + ), + ), + ( + dedent("""\ + Test: SBDR + 3 passed + 1 failed + Test: KBD + 2 skipped + Test aborted + """), + FwtsResult( + tests=[ + { + "name": "SBDR", + "subtests": {"passed": 3, "failed": 1, "skipped": 0, "aborted": 0}, + }, + { + "name": "KBD", + "subtests": {"passed": 0, "failed": 0, "skipped": 2, "aborted": 1}, + }, + ], + summary={"passed": 3, "failed": 1, "skipped": 2, "aborted": 1}, + ), + ), + ( + dedent("""\ + Test: CPU + 3 info only + """), + FwtsResult( + tests=[ + { + "name": "CPU", + "subtests": {"passed": 3, "failed": 0, "skipped": 0, "aborted": 0}, + }, + ], + summary={"passed": 3, "failed": 0, "skipped": 0, "aborted": 0}, + ), + ), + ( + dedent("""\ + Test: MCHV + 2 passed + 1 info only + """), + FwtsResult( + tests=[ + { + "name": "MCHV", + "subtests": {"passed": 3, "failed": 0, "skipped": 0, "aborted": 0}, + }, + ], + summary={"passed": 3, "failed": 0, "skipped": 0, "aborted": 0}, + ), + ), + ( + "Test: HPET\n", + FwtsResult( + tests=[ + { + "name": "HPET", + "subtests": {"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + }, + ], + summary={"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + ), + ), + ( + dedent("""\ + Test: FAN + 5 passed + 0 failed + 3 skipped + 1 aborted + """), + FwtsResult( + tests=[ + { + "name": "FAN", + "subtests": {"passed": 5, "failed": 0, "skipped": 3, "aborted": 1}, + }, + ], + summary={"passed": 5, "failed": 0, "skipped": 3, "aborted": 1}, + ), + ), + ( + dedent("""\ + Test: ACPI DSDT + 1 passed + Test: APIC + 2 passed + 1 failed + Test: MADT + Test aborted + Test: UEFI + Test skipped + Test: CPU + 3 info only + """), + FwtsResult( + tests=[ + { + "name": "ACPI DSDT", + "subtests": {"passed": 1, "failed": 0, "skipped": 0, "aborted": 0}, + }, + { + "name": "APIC", + "subtests": {"passed": 2, "failed": 1, "skipped": 0, "aborted": 0}, + }, + { + "name": "MADT", + "subtests": {"passed": 0, "failed": 0, "skipped": 0, "aborted": 1}, + }, + { + "name": "UEFI", + "subtests": {"passed": 0, "failed": 0, "skipped": 1, "aborted": 0}, + }, + { + "name": "CPU", + "subtests": {"passed": 3, "failed": 0, "skipped": 0, "aborted": 0}, + }, + ], + summary={"passed": 6, "failed": 1, "skipped": 1, "aborted": 1}, + ), + ), + ], + ids=[ + "empty_output", + "one_test_all_passed", + "one_test_mix_pass_fail", + "test_aborted_without_dot", + "test_aborted_with_dot", + "test_skipped_without_dot", + "test_skipped_with_dot", + "multiple_tests_mixed_status", + "all_info_only", + "passed_and_info_only", + "no_subresults_skipped", + "all_counters_nonzero", + "complex", + ], +) +def test_parse_metrics(raw_output: str, expected: FwtsResult) -> None: + result = Fwts.parse_metrics(raw_output) + assert result == expected