path: root/dts/framework/runner.py

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

"""Test suite runner module.

The module is responsible for running DTS in a series of stages:

    #. Test run stage,
    #. DPDK build stage,
    #. Test suite stage,
    #. Test case stage.

The test run stage sets up the environment before running test suites.
The test suite stage sets up steps common to all test cases
and the test case stage runs test cases individually.
"""

import os
import random
import sys
from pathlib import Path
from types import MethodType
from typing import Iterable

from framework.testbed_model.capability import Capability, get_supported_capabilities
from framework.testbed_model.sut_node import SutNode
from framework.testbed_model.tg_node import TGNode

from .config import (
    Configuration,
    DPDKPrecompiledBuildConfiguration,
    SutNodeConfiguration,
    TestRunConfiguration,
    TestSuiteConfig,
    TGNodeConfiguration,
    load_config,
)
from .exception import BlockingTestSuiteError, SSHTimeoutError, TestCaseVerifyError
from .logger import DTSLogger, DtsStage, get_dts_logger
from .settings import SETTINGS
from .test_result import (
    DTSResult,
    Result,
    TestCaseResult,
    TestRunResult,
    TestSuiteResult,
    TestSuiteWithCases,
)
from .test_suite import TestCase, TestSuite
from .testbed_model.topology import Topology


class DTSRunner:
    r"""Test suite runner class.

    The class is responsible for running tests on testbeds defined in the test run configuration.
    Each setup or teardown of each stage is recorded in a :class:`~framework.test_result.DTSResult`
    or one of its subclasses. The test case results are also recorded.

    If an error occurs, the current stage is aborted, the error is recorded, everything in
    the inner stages is marked as blocked and the run continues in the next iteration
    of the same stage. The return code is the highest `severity` of all
    :class:`~.framework.exception.DTSError`\s.

    Example:
        An error occurs in a test suite setup. The current test suite is aborted,
        all its test cases are marked as blocked and the run continues
        with the next test suite. If the errored test suite was the last one in the
        given test run, the next test run begins.
    """

    _configuration: Configuration
    _logger: DTSLogger
    _result: DTSResult
    _test_suite_class_prefix: str
    _test_suite_module_prefix: str
    _func_test_case_regex: str
    _perf_test_case_regex: str

    def __init__(self):
        """Initialize the instance with configuration, logger, result and string constants."""
        self._configuration = load_config(SETTINGS.config_file_path)
        self._logger = get_dts_logger()
        if not os.path.exists(SETTINGS.output_dir):
            os.makedirs(SETTINGS.output_dir)
        self._logger.add_dts_root_logger_handlers(SETTINGS.verbose, SETTINGS.output_dir)
        self._result = DTSResult(SETTINGS.output_dir, self._logger)
        self._test_suite_class_prefix = "Test"
        self._test_suite_module_prefix = "tests.TestSuite_"
        self._func_test_case_regex = r"test_(?!perf_)"
        self._perf_test_case_regex = r"test_perf_"

    def run(self) -> None:
        """Run all test runs from the test run configuration.

        Before running test suites, test runs are first set up.
        The test runs defined in the test run configuration are iterated over.
        The test runs define which tests to run and where to run them.

        The test suites are set up for each test run and each discovered
        test case within the test suite is set up, executed and torn down. After all test cases
        have been executed, the test suite is torn down and the next test suite will be run. Once
        all test suites have been run, the next test run will be tested.

        In order to properly mark test suites and test cases as blocked in case of a failure,
        we need to have discovered which test suites and test cases to run before any failures
        happen. The discovery happens at the earliest point at the start of each test run.

        All the nested steps look like this:

            #. Test run setup

                #. Test suite setup

                    #. Test case setup
                    #. Test case logic
                    #. Test case teardown

                #. Test suite teardown

            #. Test run teardown

        The test cases are filtered according to the specification in the test run configuration and
        the :option:`--test-suite` command line argument or
        the :envvar:`DTS_TESTCASES` environment variable.
        """
        sut_nodes: dict[str, SutNode] = {}
        tg_nodes: dict[str, TGNode] = {}
        try:
            # check the python version of the server that runs dts
            self._check_dts_python_version()
            self._result.update_setup(Result.PASS)

            # for all test run sections
            for test_run_with_nodes_config in self._configuration.test_runs_with_nodes:
                test_run_config, sut_node_config, tg_node_config = test_run_with_nodes_config
                self._logger.set_stage(DtsStage.test_run_setup)
                self._logger.info(f"Running test run with SUT '{sut_node_config.name}'.")
                self._init_random_seed(test_run_config)
                test_run_result = self._result.add_test_run(test_run_config)
                # we don't want to modify the original config, so create a copy
                test_run_test_suites = list(
                    SETTINGS.test_suites if SETTINGS.test_suites else test_run_config.test_suites
                )
                if not test_run_config.skip_smoke_tests:
                    test_run_test_suites[:0] = [TestSuiteConfig(test_suite="smoke_tests")]
                try:
                    test_suites_with_cases = self._get_test_suites_with_cases(
                        test_run_test_suites, test_run_config.func, test_run_config.perf
                    )
                    test_run_result.test_suites_with_cases = test_suites_with_cases
                except Exception as e:
                    self._logger.exception(
                        f"Invalid test suite configuration found: " f"{test_run_test_suites}."
                    )
                    test_run_result.update_setup(Result.FAIL, e)

                else:
                    self._connect_nodes_and_run_test_run(
                        sut_nodes,
                        tg_nodes,
                        sut_node_config,
                        tg_node_config,
                        test_run_config,
                        test_run_result,
                        test_suites_with_cases,
                    )

        except Exception as e:
            self._logger.exception("An unexpected error has occurred.")
            self._result.add_error(e)
            raise

        finally:
            try:
                self._logger.set_stage(DtsStage.post_run)
                for node in (sut_nodes | tg_nodes).values():
                    node.close()
                self._result.update_teardown(Result.PASS)
            except Exception as e:
                self._logger.exception("The final cleanup of nodes failed.")
                self._result.update_teardown(Result.ERROR, e)

        # we need to put the sys.exit call outside the finally clause to make sure
        # that unexpected exceptions will propagate
        # in that case, the error that should be reported is the uncaught exception as
        # that is a severe error originating from the framework
        # at that point, we'll only have partial results which could be impacted by the
        # error causing the uncaught exception, making them uninterpretable
        self._exit_dts()

    def _check_dts_python_version(self) -> None:
        """Check the required Python version - v3.10."""
        if sys.version_info.major < 3 or (
            sys.version_info.major == 3 and sys.version_info.minor < 10
        ):
            self._logger.warning(
                "DTS execution node's python version is lower than Python 3.10, "
                "is deprecated and will not work in future releases."
            )
            self._logger.warning("Please use Python >= 3.10 instead.")

    def _get_test_suites_with_cases(
        self,
        test_suite_configs: list[TestSuiteConfig],
        func: bool,
        perf: bool,
    ) -> list[TestSuiteWithCases]:
        """Get test suites with selected cases.

        The test suites with test cases defined in the user configuration are selected
        and the corresponding functions and classes are gathered.

        Args:
            test_suite_configs: Test suite configurations.
            func: Whether to include functional test cases in the final list.
            perf: Whether to include performance test cases in the final list.

        Returns:
            The test suites, each with test cases.
        """
        test_suites_with_cases = []

        for test_suite_config in test_suite_configs:
            test_suite_class = test_suite_config.test_suite_spec.class_obj
            test_cases: list[type[TestCase]] = []
            func_test_cases, perf_test_cases = test_suite_class.filter_test_cases(
                test_suite_config.test_cases_names
            )
            if func:
                test_cases.extend(func_test_cases)
            if perf:
                test_cases.extend(perf_test_cases)

            test_suites_with_cases.append(
                TestSuiteWithCases(test_suite_class=test_suite_class, test_cases=test_cases)
            )
        return test_suites_with_cases

    def _connect_nodes_and_run_test_run(
        self,
        sut_nodes: dict[str, SutNode],
        tg_nodes: dict[str, TGNode],
        sut_node_config: SutNodeConfiguration,
        tg_node_config: TGNodeConfiguration,
        test_run_config: TestRunConfiguration,
        test_run_result: TestRunResult,
        test_suites_with_cases: Iterable[TestSuiteWithCases],
    ) -> None:
        """Connect nodes, then continue to run the given test run.

        Connect the :class:`SutNode` and the :class:`TGNode` of this `test_run_config`.
        If either has already been connected, it's going to be in either `sut_nodes` or `tg_nodes`,
        respectively.
        If not, connect and add the node to the respective `sut_nodes` or `tg_nodes` :class:`dict`.

        Args:
            sut_nodes: A dictionary storing connected/to be connected SUT nodes.
            tg_nodes: A dictionary storing connected/to be connected TG nodes.
            sut_node_config: The test run's SUT node configuration.
            tg_node_config: The test run's TG node configuration.
            test_run_config: A test run configuration.
            test_run_result: The test run's result.
            test_suites_with_cases: The test suites with test cases to run.
        """
        sut_node = sut_nodes.get(sut_node_config.name)
        tg_node = tg_nodes.get(tg_node_config.name)

        try:
            if not sut_node:
                sut_node = SutNode(sut_node_config)
                sut_nodes[sut_node.name] = sut_node
            if not tg_node:
                tg_node = TGNode(tg_node_config)
                tg_nodes[tg_node.name] = tg_node
        except Exception as e:
            failed_node = test_run_config.system_under_test_node.node_name
            if sut_node:
                failed_node = test_run_config.traffic_generator_node
            self._logger.exception(f"The Creation of node {failed_node} failed.")
            test_run_result.update_setup(Result.FAIL, e)

        else:
            self._run_test_run(
                sut_node, tg_node, test_run_config, test_run_result, test_suites_with_cases
            )

    def _run_test_run(
        self,
        sut_node: SutNode,
        tg_node: TGNode,
        test_run_config: TestRunConfiguration,
        test_run_result: TestRunResult,
        test_suites_with_cases: Iterable[TestSuiteWithCases],
    ) -> None:
        """Run the given test run.

        This involves running the test run setup as well as running all test suites
        in the given test run. After that, the test run teardown is run.

        Args:
            sut_node: The test run's SUT node.
            tg_node: The test run's TG node.
            test_run_config: A test run configuration.
            test_run_result: The test run's result.
            test_suites_with_cases: The test suites with test cases to run.

        Raises:
            ConfigurationError: If the DPDK sources or build is not set up from config or settings.
        """
        self._logger.info(
            f"Running test run with SUT '{test_run_config.system_under_test_node.node_name}'."
        )
        test_run_result.ports = sut_node.ports
        test_run_result.sut_info = sut_node.node_info
        try:
            dpdk_build_config = test_run_config.dpdk_config
            if new_location := SETTINGS.dpdk_location:
                dpdk_build_config = dpdk_build_config.model_copy(
                    update={"dpdk_location": new_location}
                )
            if dir := SETTINGS.precompiled_build_dir:
                dpdk_build_config = DPDKPrecompiledBuildConfiguration(
                    dpdk_location=dpdk_build_config.dpdk_location, precompiled_build_dir=dir
                )
            sut_node.set_up_test_run(test_run_config, dpdk_build_config)
            test_run_result.dpdk_build_info = sut_node.get_dpdk_build_info()
            tg_node.set_up_test_run(test_run_config, dpdk_build_config)
            test_run_result.update_setup(Result.PASS)
        except Exception as e:
            self._logger.exception("Test run setup failed.")
            test_run_result.update_setup(Result.FAIL, e)

        else:
            self._run_test_suites(sut_node, tg_node, test_run_result, test_suites_with_cases)

        finally:
            try:
                self._logger.set_stage(DtsStage.test_run_teardown)
                sut_node.tear_down_test_run()
                tg_node.tear_down_test_run()
                test_run_result.update_teardown(Result.PASS)
            except Exception as e:
                self._logger.exception("Test run teardown failed.")
                test_run_result.update_teardown(Result.FAIL, e)

    def _get_supported_capabilities(
        self,
        sut_node: SutNode,
        topology_config: Topology,
        test_suites_with_cases: Iterable[TestSuiteWithCases],
    ) -> set[Capability]:

        capabilities_to_check = set()
        for test_suite_with_cases in test_suites_with_cases:
            capabilities_to_check.update(test_suite_with_cases.required_capabilities)

        self._logger.debug(f"Found capabilities to check: {capabilities_to_check}")

        return get_supported_capabilities(sut_node, topology_config, capabilities_to_check)

    def _run_test_suites(
        self,
        sut_node: SutNode,
        tg_node: TGNode,
        test_run_result: TestRunResult,
        test_suites_with_cases: Iterable[TestSuiteWithCases],
    ) -> None:
        """Run `test_suites_with_cases` with the current test run.

        The method assumes the DPDK we're testing has already been built on the SUT node.

        Before running any suites, the method determines whether they should be skipped
        by inspecting any required capabilities the test suite needs and comparing those
        to capabilities supported by the tested environment. If all capabilities are supported,
        the suite is run. If all test cases in a test suite would be skipped, the whole test suite
        is skipped (the setup and teardown is not run).

        If a blocking test suite (such as the smoke test suite) fails, the rest of the test suites
        in the current test run won't be executed.

        Args:
            sut_node: The test run's SUT node.
            tg_node: The test run's TG node.
            test_run_result: The test run's result.
            test_suites_with_cases: The test suites with test cases to run.
        """
        end_dpdk_build = False
        topology = Topology(sut_node.ports, tg_node.ports)
        supported_capabilities = self._get_supported_capabilities(
            sut_node, topology, test_suites_with_cases
        )
        for test_suite_with_cases in test_suites_with_cases:
            test_suite_with_cases.mark_skip_unsupported(supported_capabilities)
            test_suite_result = test_run_result.add_test_suite(test_suite_with_cases)
            try:
                if not test_suite_with_cases.skip:
                    self._run_test_suite(
                        sut_node,
                        tg_node,
                        topology,
                        test_suite_result,
                        test_suite_with_cases,
                    )
                else:
                    self._logger.info(
                        f"Test suite execution SKIPPED: "
                        f"'{test_suite_with_cases.test_suite_class.__name__}'. Reason: "
                        f"{test_suite_with_cases.test_suite_class.skip_reason}"
                    )
                    test_suite_result.update_setup(Result.SKIP)
            except BlockingTestSuiteError as e:
                self._logger.exception(
                    f"An error occurred within {test_suite_with_cases.test_suite_class.__name__}. "
                    "Skipping the rest of the test suites in this test run."
                )
                self._result.add_error(e)
                end_dpdk_build = True
            # if a blocking test failed and we need to bail out of suite executions
            if end_dpdk_build:
                break

    def _run_test_suite(
        self,
        sut_node: SutNode,
        tg_node: TGNode,
        topology: Topology,
        test_suite_result: TestSuiteResult,
        test_suite_with_cases: TestSuiteWithCases,
    ) -> None:
        """Set up, execute and tear down `test_suite_with_cases`.

        The method assumes the DPDK we're testing has already been built on the SUT node.

        Test suite execution consists of running the discovered test cases.
        A test case run consists of setup, execution and teardown of said test case.

        Record the setup and the teardown and handle failures.

        Args:
            sut_node: The test run's SUT node.
            tg_node: The test run's TG node.
            test_suite_result: The test suite level result object associated
                with the current test suite.
            test_suite_with_cases: The test suite with test cases to run.

        Raises:
            BlockingTestSuiteError: If a blocking test suite fails.
        """
        test_suite_name = test_suite_with_cases.test_suite_class.__name__
        self._logger.set_stage(
            DtsStage.test_suite_setup, Path(SETTINGS.output_dir, test_suite_name)
        )
        test_suite = test_suite_with_cases.test_suite_class(sut_node, tg_node, topology)
        try:
            self._logger.info(f"Starting test suite setup: {test_suite_name}")
            test_suite.set_up_suite()
            test_suite_result.update_setup(Result.PASS)
            self._logger.info(f"Test suite setup successful: {test_suite_name}")
        except Exception as e:
            self._logger.exception(f"Test suite setup ERROR: {test_suite_name}")
            test_suite_result.update_setup(Result.ERROR, e)

        else:
            self._execute_test_suite(
                test_suite,
                test_suite_with_cases.test_cases,
                test_suite_result,
            )
        finally:
            try:
                self._logger.set_stage(DtsStage.test_suite_teardown)
                test_suite.tear_down_suite()
                sut_node.kill_cleanup_dpdk_apps()
                test_suite_result.update_teardown(Result.PASS)
            except Exception as e:
                self._logger.exception(f"Test suite teardown ERROR: {test_suite_name}")
                self._logger.warning(
                    f"Test suite '{test_suite_name}' teardown failed, "
                    "the next test suite may be affected."
                )
                test_suite_result.update_setup(Result.ERROR, e)
            if len(test_suite_result.get_errors()) > 0 and test_suite.is_blocking:
                raise BlockingTestSuiteError(test_suite_name)

    def _execute_test_suite(
        self,
        test_suite: TestSuite,
        test_cases: Iterable[type[TestCase]],
        test_suite_result: TestSuiteResult,
    ) -> None:
        """Execute all `test_cases` in `test_suite`.

        If the :option:`--re-run` command line argument or the :envvar:`DTS_RERUN` environment
        variable is set, in case of a test case failure, the test case will be executed again
        until it passes or it fails that many times in addition of the first failure.

        Args:
            test_suite: The test suite object.
            test_cases: The list of test case functions.
            test_suite_result: The test suite level result object associated
                with the current test suite.
        """
        self._logger.set_stage(DtsStage.test_suite)
        for test_case in test_cases:
            test_case_name = test_case.__name__
            test_case_result = test_suite_result.add_test_case(test_case_name)
            all_attempts = SETTINGS.re_run + 1
            attempt_nr = 1
            if not test_case.skip:
                self._run_test_case(test_suite, test_case, test_case_result)
                while not test_case_result and attempt_nr < all_attempts:
                    attempt_nr += 1
                    self._logger.info(
                        f"Re-running FAILED test case '{test_case_name}'. "
                        f"Attempt number {attempt_nr} out of {all_attempts}."
                    )
                    self._run_test_case(test_suite, test_case, test_case_result)
            else:
                self._logger.info(
                    f"Test case execution SKIPPED: {test_case_name}. Reason: "
                    f"{test_case.skip_reason}"
                )
                test_case_result.update_setup(Result.SKIP)

    def _run_test_case(
        self,
        test_suite: TestSuite,
        test_case: type[TestCase],
        test_case_result: TestCaseResult,
    ) -> None:
        """Setup, execute and teardown `test_case_method` from `test_suite`.

        Record the result of the setup and the teardown and handle failures.

        Args:
            test_suite: The test suite object.
            test_case: The test case function.
            test_case_result: The test case level result object associated
                with the current test case.
        """
        test_case_name = test_case.__name__

        try:
            # run set_up function for each case
            test_suite.set_up_test_case()
            test_case_result.update_setup(Result.PASS)
        except SSHTimeoutError as e:
            self._logger.exception(f"Test case setup FAILED: {test_case_name}")
            test_case_result.update_setup(Result.FAIL, e)
        except Exception as e:
            self._logger.exception(f"Test case setup ERROR: {test_case_name}")
            test_case_result.update_setup(Result.ERROR, e)

        else:
            # run test case if setup was successful
            self._execute_test_case(test_suite, test_case, test_case_result)

        finally:
            try:
                test_suite.tear_down_test_case()
                test_case_result.update_teardown(Result.PASS)
            except Exception as e:
                self._logger.exception(f"Test case teardown ERROR: {test_case_name}")
                self._logger.warning(
                    f"Test case '{test_case_name}' teardown failed, "
                    f"the next test case may be affected."
                )
                test_case_result.update_teardown(Result.ERROR, e)
                test_case_result.update(Result.ERROR)

    def _execute_test_case(
        self,
        test_suite: TestSuite,
        test_case: type[TestCase],
        test_case_result: TestCaseResult,
    ) -> None:
        """Execute `test_case_method` from `test_suite`, record the result and handle failures.

        Args:
            test_suite: The test suite object.
            test_case: The test case function.
            test_case_result: The test case level result object associated
                with the current test case.
        """
        test_case_name = test_case.__name__
        try:
            self._logger.info(f"Starting test case execution: {test_case_name}")
            # Explicit method binding is required, otherwise mypy complains
            MethodType(test_case, test_suite)()
            test_case_result.update(Result.PASS)
            self._logger.info(f"Test case execution PASSED: {test_case_name}")

        except TestCaseVerifyError as e:
            self._logger.exception(f"Test case execution FAILED: {test_case_name}")
            test_case_result.update(Result.FAIL, e)
        except Exception as e:
            self._logger.exception(f"Test case execution ERROR: {test_case_name}")
            test_case_result.update(Result.ERROR, e)
        except KeyboardInterrupt:
            self._logger.error(f"Test case execution INTERRUPTED by user: {test_case_name}")
            test_case_result.update(Result.SKIP)
            raise KeyboardInterrupt("Stop DTS")

    def _exit_dts(self) -> None:
        """Process all errors and exit with the proper exit code."""
        self._result.process()

        if self._logger:
            self._logger.info("DTS execution has ended.")

        sys.exit(self._result.get_return_code())

    def _init_random_seed(self, conf: TestRunConfiguration) -> None:
        """Initialize the random seed to use for the test run."""
        seed = SETTINGS.random_seed or conf.random_seed or random.randrange(0xFFFF_FFFF)
        self._logger.info(f"Initializing test run with random seed {seed}.")
        random.seed(seed)