path: root/dts/framework/test_result.py

# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2023 PANTHEON.tech s.r.o.
# Copyright(c) 2023 University of New Hampshire
# Copyright(c) 2024 Arm Limited

r"""Record and process DTS results.

The results are recorded in a hierarchical manner:

    * :class:`DTSResult` contains
    * :class:`TestRunResult` contains
    * :class:`TestSuiteResult` contains
    * :class:`TestCaseResult`

Each result may contain multiple lower level results, e.g. there are multiple
:class:`TestSuiteResult`\s in a :class:`TestRunResult`.
The results have common parts, such as setup and teardown results, captured in :class:`BaseResult`,
which also defines some common behaviors in its methods.

Each result class has its own idiosyncrasies which they implement in overridden methods.

The :option:`--output` command line argument and the :envvar:`DTS_OUTPUT_DIR` environment
variable modify the directory where the files with results will be stored.
"""

import json
from collections.abc import MutableSequence
from dataclasses import asdict, dataclass, field
from enum import Enum, auto
from pathlib import Path
from typing import Any, Callable, TypedDict

from framework.testbed_model.capability import Capability

from .config import TestRunConfiguration, TestSuiteConfig
from .exception import DTSError, ErrorSeverity
from .logger import DTSLogger
from .test_suite import TestCase, TestSuite
from .testbed_model.os_session import OSSessionInfo
from .testbed_model.port import Port
from .testbed_model.sut_node import DPDKBuildInfo


@dataclass(slots=True, frozen=True)
class TestSuiteWithCases:
    """A test suite class with test case methods.

    An auxiliary class holding a test case class with test case methods. The intended use of this
    class is to hold a subset of test cases (which could be all test cases) because we don't have
    all the data to instantiate the class at the point of inspection. The knowledge of this subset
    is needed in case an error occurs before the class is instantiated and we need to record
    which test cases were blocked by the error.

    Attributes:
        test_suite_class: The test suite class.
        test_cases: The test case methods.
        required_capabilities: The combined required capabilities of both the test suite
            and the subset of test cases.
    """

    test_suite_class: type[TestSuite]
    test_cases: list[type[TestCase]]
    required_capabilities: set[Capability] = field(default_factory=set, init=False)

    def __post_init__(self):
        """Gather the required capabilities of the test suite and all test cases."""
        for test_object in [self.test_suite_class] + self.test_cases:
            self.required_capabilities.update(test_object.required_capabilities)

    def create_config(self) -> TestSuiteConfig:
        """Generate a :class:`TestSuiteConfig` from the stored test suite with test cases.

        Returns:
            The :class:`TestSuiteConfig` representation.
        """
        return TestSuiteConfig(
            test_suite=self.test_suite_class.__name__,
            test_cases=[test_case.__name__ for test_case in self.test_cases],
        )

    def mark_skip_unsupported(self, supported_capabilities: set[Capability]) -> None:
        """Mark the test suite and test cases to be skipped.

        The mark is applied if object to be skipped requires any capabilities and at least one of
        them is not among `supported_capabilities`.

        Args:
            supported_capabilities: The supported capabilities.
        """
        for test_object in [self.test_suite_class, *self.test_cases]:
            capabilities_not_supported = test_object.required_capabilities - supported_capabilities
            if capabilities_not_supported:
                test_object.skip = True
                capability_str = (
                    "capability" if len(capabilities_not_supported) == 1 else "capabilities"
                )
                test_object.skip_reason = (
                    f"Required {capability_str} '{capabilities_not_supported}' not found."
                )
        if not self.test_suite_class.skip:
            if all(test_case.skip for test_case in self.test_cases):
                self.test_suite_class.skip = True

                self.test_suite_class.skip_reason = (
                    "All test cases are marked to be skipped with reasons: "
                    f"{' '.join(test_case.skip_reason for test_case in self.test_cases)}"
                )

    @property
    def skip(self) -> bool:
        """Skip the test suite if all test cases or the suite itself are to be skipped.

        Returns:
            :data:`True` if the test suite should be skipped, :data:`False` otherwise.
        """
        return all(test_case.skip for test_case in self.test_cases) or self.test_suite_class.skip


class Result(Enum):
    """The possible states that a setup, a teardown or a test case may end up in."""

    #:
    PASS = auto()
    #:
    FAIL = auto()
    #:
    ERROR = auto()
    #:
    BLOCK = auto()
    #:
    SKIP = auto()

    def __bool__(self) -> bool:
        """Only :attr:`PASS` is True."""
        return self is self.PASS


class TestCaseResultDict(TypedDict):
    """Represents the `TestCaseResult` results.

    Attributes:
        test_case_name: The name of the test case.
        result: The result name of the test case.
    """

    test_case_name: str
    result: str


class TestSuiteResultDict(TypedDict):
    """Represents the `TestSuiteResult` results.

    Attributes:
        test_suite_name: The name of the test suite.
        test_cases: A list of test case results contained in this test suite.
    """

    test_suite_name: str
    test_cases: list[TestCaseResultDict]


class TestRunResultDict(TypedDict, total=False):
    """Represents the `TestRunResult` results.

    Attributes:
        compiler_version: The version of the compiler used for the DPDK build.
        dpdk_version: The version of DPDK being tested.
        ports: A list of ports associated with the test run.
        test_suites: A list of test suite results included in this test run.
        summary: A dictionary containing overall results, such as pass/fail counts.
    """

    compiler_version: str | None
    dpdk_version: str | None
    ports: list[dict[str, Any]]
    test_suites: list[TestSuiteResultDict]
    summary: dict[str, int | float]


class DtsRunResultDict(TypedDict):
    """Represents the `DtsRunResult` results.

    Attributes:
        test_runs: A list of test run results.
        summary: A summary dictionary containing overall statistics for the test runs.
    """

    test_runs: list[TestRunResultDict]
    summary: dict[str, int | float]


class FixtureResult:
    """A record that stores the result of a setup or a teardown.

    :attr:`~Result.FAIL` is a sensible default since it prevents false positives (which could happen
    if the default was :attr:`~Result.PASS`).

    Preventing false positives or other false results is preferable since a failure
    is mostly likely to be investigated (the other false results may not be investigated at all).

    Attributes:
        result: The associated result.
        error: The error in case of a failure.
    """

    result: Result
    error: Exception | None = None

    def __init__(
        self,
        result: Result = Result.FAIL,
        error: Exception | None = None,
    ):
        """Initialize the constructor with the fixture result and store a possible error.

        Args:
            result: The result to store.
            error: The error which happened when a failure occurred.
        """
        self.result = result
        self.error = error

    def __bool__(self) -> bool:
        """A wrapper around the stored :class:`Result`."""
        return bool(self.result)


class BaseResult:
    """Common data and behavior of DTS results.

    Stores the results of the setup and teardown portions of the corresponding stage.
    The hierarchical nature of DTS results is captured recursively in an internal list.
    A stage is each level in this particular hierarchy (pre-run or the top-most level,
    test run, test suite and test case).

    Attributes:
        setup_result: The result of the setup of the particular stage.
        teardown_result: The results of the teardown of the particular stage.
        child_results: The results of the descendants in the results hierarchy.
    """

    setup_result: FixtureResult
    teardown_result: FixtureResult
    child_results: MutableSequence["BaseResult"]

    def __init__(self):
        """Initialize the constructor."""
        self.setup_result = FixtureResult()
        self.teardown_result = FixtureResult()
        self.child_results = []

    def update_setup(self, result: Result, error: Exception | None = None) -> None:
        """Store the setup result.

        If the result is :attr:`~Result.BLOCK`, :attr:`~Result.ERROR` or :attr:`~Result.FAIL`,
        then the corresponding child results in result hierarchy
        are also marked with :attr:`~Result.BLOCK`.

        Args:
            result: The result of the setup.
            error: The error that occurred in case of a failure.
        """
        self.setup_result.result = result
        self.setup_result.error = error

        if result != Result.PASS:
            result_to_mark = Result.BLOCK if result != Result.SKIP else Result.SKIP
            self.update_teardown(result_to_mark)
            self._mark_results(result_to_mark)

    def _mark_results(self, result) -> None:
        """Mark the child results or the result of the level itself as `result`.

        The marking of results should be done in overloaded methods.
        """

    def update_teardown(self, result: Result, error: Exception | None = None) -> None:
        """Store the teardown result.

        Args:
            result: The result of the teardown.
            error: The error that occurred in case of a failure.
        """
        self.teardown_result.result = result
        self.teardown_result.error = error

    def _get_setup_teardown_errors(self) -> list[Exception]:
        errors = []
        if self.setup_result.error:
            errors.append(self.setup_result.error)
        if self.teardown_result.error:
            errors.append(self.teardown_result.error)
        return errors

    def _get_child_errors(self) -> list[Exception]:
        return [error for child_result in self.child_results for error in child_result.get_errors()]

    def get_errors(self) -> list[Exception]:
        """Compile errors from the whole result hierarchy.

        Returns:
            The errors from setup, teardown and all errors found in the whole result hierarchy.
        """
        return self._get_setup_teardown_errors() + self._get_child_errors()

    def to_dict(self):
        """Convert the results hierarchy into a dictionary representation."""

    def add_result(self, results: dict[str, int]):
        """Collate the test case result to the given result hierarchy.

        Args:
            results: The dictionary in which results will be collated.
        """
        for child_result in self.child_results:
            child_result.add_result(results)

    def generate_pass_rate_dict(self, test_run_summary) -> dict[str, float]:
        """Generate a dictionary with the PASS/FAIL ratio of all test cases.

        Args:
            test_run_summary: The summary dictionary containing test result counts.

        Returns:
            A dictionary with the PASS/FAIL ratio of all test cases.
        """
        return {
            "PASS_RATE": (
                float(test_run_summary[Result.PASS.name])
                * 100
                / sum(test_run_summary[result.name] for result in Result if result != Result.SKIP)
            )
        }


class DTSResult(BaseResult):
    """Stores environment information and test results from a DTS run.

        * Test run level information, such as testbed, the test suite list and
          DPDK build configuration (compiler, target OS and cpu),
        * Test suite and test case results,
        * All errors that are caught and recorded during DTS execution.

    The information is stored hierarchically. This is the first level of the hierarchy
    and as such is where the data form the whole hierarchy is collated or processed.

    The internal list stores the results of all test runs.
    """

    _output_dir: str
    _logger: DTSLogger
    _errors: list[Exception]
    _return_code: ErrorSeverity

    def __init__(self, output_dir: str, logger: DTSLogger):
        """Extend the constructor with top-level specifics.

        Args:
            output_dir: The directory where DTS logs and results are saved.
            logger: The logger instance the whole result will use.
        """
        super().__init__()
        self._output_dir = output_dir
        self._logger = logger
        self._errors = []
        self._return_code = ErrorSeverity.NO_ERR

    def add_test_run(self, test_run_config: TestRunConfiguration) -> "TestRunResult":
        """Add and return the child result (test run).

        Args:
            test_run_config: A test run configuration.

        Returns:
            The test run's result.
        """
        result = TestRunResult(test_run_config)
        self.child_results.append(result)
        return result

    def add_error(self, error: Exception) -> None:
        """Record an error that occurred outside any test run.

        Args:
            error: The exception to record.
        """
        self._errors.append(error)

    def process(self) -> None:
        """Process the data after a whole DTS run.

        The data is added to child objects during runtime and this object is not updated
        at that time. This requires us to process the child data after it's all been gathered.

        The processing gathers all errors and the statistics of test case results.
        """
        self._errors += self.get_errors()
        if self._errors and self._logger:
            self._logger.debug("Summary of errors:")
            for error in self._errors:
                self._logger.debug(repr(error))

        TextSummary(self).save(Path(self._output_dir, "results_summary.txt"))
        JsonResults(self).save(Path(self._output_dir, "results.json"))

    def get_return_code(self) -> int:
        """Go through all stored Exceptions and return the final DTS error code.

        Returns:
            The highest error code found.
        """
        for error in self._errors:
            error_return_code = ErrorSeverity.GENERIC_ERR
            if isinstance(error, DTSError):
                error_return_code = error.severity

            if error_return_code > self._return_code:
                self._return_code = error_return_code

        return int(self._return_code)

    def to_dict(self) -> DtsRunResultDict:
        """Convert DTS result into a dictionary format.

        The dictionary contains test runs and summary of test runs.

        Returns:
            A dictionary representation of the DTS result
        """

        def merge_test_run_summaries(test_run_summaries: list[dict[str, int]]) -> dict[str, int]:
            """Merge multiple test run summaries into one dictionary.

            Args:
                test_run_summaries: List of test run summary dictionaries.

            Returns:
                A merged dictionary containing the aggregated summary.
            """
            return {
                key.name: sum(test_run_summary[key.name] for test_run_summary in test_run_summaries)
                for key in Result
            }

        test_runs = [child.to_dict() for child in self.child_results]
        test_run_summary = merge_test_run_summaries([test_run["summary"] for test_run in test_runs])

        return {
            "test_runs": test_runs,
            "summary": test_run_summary | self.generate_pass_rate_dict(test_run_summary),
        }


class TestRunResult(BaseResult):
    """The test run specific result.

    The internal list stores the results of all test suites in a given test run.

    Attributes:
        compiler_version: The DPDK build compiler version.
        dpdk_version: The built DPDK version.
        sut_os_name: The operating system of the SUT node.
        sut_os_version: The operating system version of the SUT node.
        sut_kernel_version: The operating system kernel version of the SUT node.
    """

    _config: TestRunConfiguration
    _test_suites_with_cases: list[TestSuiteWithCases]
    _ports: list[Port]
    _sut_info: OSSessionInfo | None
    _dpdk_build_info: DPDKBuildInfo | None

    def __init__(self, test_run_config: TestRunConfiguration):
        """Extend the constructor with the test run's config.

        Args:
            test_run_config: A test run configuration.
        """
        super().__init__()
        self._config = test_run_config
        self._test_suites_with_cases = []
        self._ports = []
        self._sut_info = None
        self._dpdk_build_info = None

    def add_test_suite(
        self,
        test_suite_with_cases: TestSuiteWithCases,
    ) -> "TestSuiteResult":
        """Add and return the child result (test suite).

        Args:
            test_suite_with_cases: The test suite with test cases.

        Returns:
            The test suite's result.
        """
        result = TestSuiteResult(test_suite_with_cases)
        self.child_results.append(result)
        return result

    @property
    def test_suites_with_cases(self) -> list[TestSuiteWithCases]:
        """The test suites with test cases to be executed in this test run.

        The test suites can only be assigned once.

        Returns:
            The list of test suites with test cases. If an error occurs between
            the initialization of :class:`TestRunResult` and assigning test cases to the instance,
            return an empty list, representing that we don't know what to execute.
        """
        return self._test_suites_with_cases

    @test_suites_with_cases.setter
    def test_suites_with_cases(self, test_suites_with_cases: list[TestSuiteWithCases]) -> None:
        if self._test_suites_with_cases:
            raise ValueError(
                "Attempted to assign test suites to a test run result "
                "which already has test suites."
            )
        self._test_suites_with_cases = test_suites_with_cases

    @property
    def ports(self) -> list[Port]:
        """Get the list of ports associated with this test run."""
        return self._ports

    @ports.setter
    def ports(self, ports: list[Port]) -> None:
        """Set the list of ports associated with this test run.

        Args:
            ports: The list of ports to associate with this test run.

        Raises:
            ValueError: If the ports have already been assigned to this test run.
        """
        if self._ports:
            raise ValueError(
                "Attempted to assign `ports` to a test run result which already has `ports`."
            )
        self._ports = ports

    @property
    def sut_info(self) -> OSSessionInfo | None:
        """Get the SUT OS session information associated with this test run."""
        return self._sut_info

    @sut_info.setter
    def sut_info(self, sut_info: OSSessionInfo) -> None:
        """Set the SUT node information associated with this test run.

        Args:
            sut_info: The SUT node information to associate with this test run.

        Raises:
            ValueError: If the SUT information has already been assigned to this test run.
        """
        if self._sut_info:
            raise ValueError(
                "Attempted to assign `sut_info` to a test run result which already has `sut_info`."
            )
        self._sut_info = sut_info

    @property
    def dpdk_build_info(self) -> DPDKBuildInfo | None:
        """Get the DPDK build information associated with this test run."""
        return self._dpdk_build_info

    @dpdk_build_info.setter
    def dpdk_build_info(self, dpdk_build_info: DPDKBuildInfo) -> None:
        """Set the DPDK build information associated with this test run.

        Args:
            dpdk_build_info: The DPDK build information to associate with this test run.

        Raises:
            ValueError: If the DPDK build information has already been assigned to this test run.
        """
        if self._dpdk_build_info:
            raise ValueError(
                "Attempted to assign `dpdk_build_info` to a test run result which already "
                "has `dpdk_build_info`."
            )
        self._dpdk_build_info = dpdk_build_info

    def to_dict(self) -> TestRunResultDict:
        """Convert the test run result into a dictionary.

        The dictionary contains test suites in this test run, and a summary of the test run and
        information about the DPDK version, compiler version and associated ports.

        Returns:
            TestRunResultDict: A dictionary representation of the test run result.
        """
        results = {result.name: 0 for result in Result}
        self.add_result(results)

        compiler_version = None
        dpdk_version = None

        if self.dpdk_build_info:
            compiler_version = self.dpdk_build_info.compiler_version
            dpdk_version = self.dpdk_build_info.dpdk_version

        return {
            "compiler_version": compiler_version,
            "dpdk_version": dpdk_version,
            "ports": [asdict(port) for port in self.ports],
            "test_suites": [child.to_dict() for child in self.child_results],
            "summary": results | self.generate_pass_rate_dict(results),
        }

    def _mark_results(self, result) -> None:
        """Mark the test suite results as `result`."""
        for test_suite_with_cases in self._test_suites_with_cases:
            child_result = self.add_test_suite(test_suite_with_cases)
            child_result.update_setup(result)


class TestSuiteResult(BaseResult):
    """The test suite specific result.

    The internal list stores the results of all test cases in a given test suite.

    Attributes:
        test_suite_name: The test suite name.
    """

    test_suite_name: str
    _test_suite_with_cases: TestSuiteWithCases
    _child_configs: list[str]

    def __init__(self, test_suite_with_cases: TestSuiteWithCases):
        """Extend the constructor with test suite's config.

        Args:
            test_suite_with_cases: The test suite with test cases.
        """
        super().__init__()
        self.test_suite_name = test_suite_with_cases.test_suite_class.__name__
        self._test_suite_with_cases = test_suite_with_cases

    def add_test_case(self, test_case_name: str) -> "TestCaseResult":
        """Add and return the child result (test case).

        Args:
            test_case_name: The name of the test case.

        Returns:
            The test case's result.
        """
        result = TestCaseResult(test_case_name)
        self.child_results.append(result)
        return result

    def to_dict(self) -> TestSuiteResultDict:
        """Convert the test suite result into a dictionary.

        The dictionary contains a test suite name and test cases given in this test suite.
        """
        return {
            "test_suite_name": self.test_suite_name,
            "test_cases": [child.to_dict() for child in self.child_results],
        }

    def _mark_results(self, result) -> None:
        """Mark the test case results as `result`."""
        for test_case_method in self._test_suite_with_cases.test_cases:
            child_result = self.add_test_case(test_case_method.__name__)
            child_result.update_setup(result)


class TestCaseResult(BaseResult, FixtureResult):
    r"""The test case specific result.

    Stores the result of the actual test case. This is done by adding an extra superclass
    in :class:`FixtureResult`. The setup and teardown results are :class:`FixtureResult`\s and
    the class is itself a record of the test case.

    Attributes:
        test_case_name: The test case name.
    """

    test_case_name: str

    def __init__(self, test_case_name: str):
        """Extend the constructor with test case's name.

        Args:
            test_case_name: The test case's name.
        """
        super().__init__()
        self.test_case_name = test_case_name

    def update(self, result: Result, error: Exception | None = None) -> None:
        """Update the test case result.

        This updates the result of the test case itself and doesn't affect
        the results of the setup and teardown steps in any way.

        Args:
            result: The result of the test case.
            error: The error that occurred in case of a failure.
        """
        self.result = result
        self.error = error

    def _get_child_errors(self) -> list[Exception]:
        if self.error:
            return [self.error]
        return []

    def to_dict(self) -> TestCaseResultDict:
        """Convert the test case result into a dictionary.

        The dictionary contains a test case name and the result name.
        """
        return {"test_case_name": self.test_case_name, "result": self.result.name}

    def add_result(self, results: dict[str, int]):
        r"""Add the test case result to the results.

        The base method goes through the hierarchy recursively and this method is here to stop
        the recursion, as the :class:`TestCaseResult` are the leaves of the hierarchy tree.

        Args:
            results: The dictionary to which results will be collated.
        """
        results[self.result.name] += 1

    def _mark_results(self, result) -> None:
        r"""Mark the result as `result`."""
        self.update(result)

    def __bool__(self) -> bool:
        """The test case passed only if setup, teardown and the test case itself passed."""
        return bool(self.setup_result) and bool(self.teardown_result) and bool(self.result)


class TextSummary:
    """Generates and saves textual summaries of DTS run results.

    The summary includes:
    * Results of test cases,
    * Compiler version of the DPDK build,
    * DPDK version of the DPDK source tree,
    * Overall summary of results when multiple test runs are present.
    """

    _dict_result: DtsRunResultDict
    _summary: dict[str, int | float]
    _text: str

    def __init__(self, dts_run_result: DTSResult):
        """Initializes with a DTSResult object and converts it to a dictionary format.

        Args:
            dts_run_result: The DTS result.
        """
        self._dict_result = dts_run_result.to_dict()
        self._summary = self._dict_result["summary"]
        self._text = ""

    @property
    def _outdent(self) -> str:
        """Appropriate indentation based on multiple test run results."""
        return "\t" if len(self._dict_result["test_runs"]) > 1 else ""

    def save(self, output_path: Path):
        """Generate and save text statistics to a file.

        Args:
            output_path: The path where the text file will be saved.
        """
        if self._dict_result["test_runs"]:
            with open(f"{output_path}", "w") as fp:
                self._add_test_runs_dict_decorator(self._add_test_run_dict)
                fp.write(self._text)

    def _add_test_runs_dict_decorator(self, func: Callable):
        """Handles multiple test runs and appends results to the summary.

        Adds headers for each test run and overall result when multiple
        test runs are provided.

        Args:
            func: Function to process and add results from each test run.
        """
        if len(self._dict_result["test_runs"]) > 1:
            for idx, test_run_result in enumerate(self._dict_result["test_runs"]):
                self._text += f"TEST_RUN_{idx}\n"
                func(test_run_result)

            self._add_overall_results()
        else:
            func(self._dict_result["test_runs"][0])

    def _add_test_run_dict(self, test_run_dict: TestRunResultDict):
        """Adds the results and the test run attributes of a single test run to the summary.

        Args:
            test_run_dict: Dictionary containing the test run results.
        """
        self._add_column(
            DPDK_VERSION=test_run_dict["dpdk_version"],
            COMPILER_VERSION=test_run_dict["compiler_version"],
            **test_run_dict["summary"],
        )
        self._text += "\n"

    def _add_column(self, **rows):
        """Formats and adds key-value pairs to the summary text.

        Handles cases where values might be None by replacing them with "N/A".

        Args:
            **rows: Arbitrary key-value pairs representing the result data.
        """
        rows = {k: "N/A" if v is None else v for k, v in rows.items()}
        max_length = len(max(rows, key=len))
        for key, value in rows.items():
            self._text += f"{self._outdent}{key:<{max_length}} = {value}\n"

    def _add_overall_results(self):
        """Add overall summary of test runs."""
        self._text += "OVERALL\n"
        self._add_column(**self._summary)


class JsonResults:
    """Save DTS run result in JSON format."""

    _dict_result: DtsRunResultDict

    def __init__(self, dts_run_result: DTSResult):
        """Initializes with a DTSResult object and converts it to a dictionary format.

        Args:
            dts_run_result: The DTS result.
        """
        self._dict_result = dts_run_result.to_dict()

    def save(self, output_path: Path):
        """Save the result to a file as JSON.

        Args:
            output_path: The path where the JSON file will be saved.
        """
        with open(f"{output_path}", "w") as fp:
            json.dump(self._dict_result, fp, indent=4)