@@ -6,14 +6,14 @@
import sys
from .config import CONFIGURATION, BuildTargetConfiguration, ExecutionConfiguration
-from .exception import DTSError, ErrorSeverity
from .logger import DTSLOG, getLogger
+from .test_result import BuildTargetResult, DTSResult, ExecutionResult, Result
from .test_suite import get_test_suites
from .testbed_model import SutNode
from .utils import check_dts_python_version
dts_logger: DTSLOG = getLogger("dts_runner")
-errors = []
+result: DTSResult = DTSResult(dts_logger)
def run_all() -> None:
@@ -22,7 +22,7 @@ def run_all() -> None:
config file.
"""
global dts_logger
- global errors
+ global result
# check the python version of the server that run dts
check_dts_python_version()
@@ -39,29 +39,31 @@ def run_all() -> None:
# the SUT has not been initialized yet
try:
sut_node = SutNode(execution.system_under_test)
+ result.update_setup(Result.PASS)
except Exception as e:
dts_logger.exception(
f"Connection to node {execution.system_under_test} failed."
)
- errors.append(e)
+ result.update_setup(Result.FAIL, e)
else:
nodes[sut_node.name] = sut_node
if sut_node:
- _run_execution(sut_node, execution)
+ _run_execution(sut_node, execution, result)
except Exception as e:
dts_logger.exception("An unexpected error has occurred.")
- errors.append(e)
+ result.add_error(e)
raise
finally:
try:
for node in nodes.values():
node.close()
+ result.update_teardown(Result.PASS)
except Exception as e:
dts_logger.exception("Final cleanup of nodes failed.")
- errors.append(e)
+ 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
@@ -72,61 +74,72 @@ def run_all() -> None:
_exit_dts()
-def _run_execution(sut_node: SutNode, execution: ExecutionConfiguration) -> None:
+def _run_execution(
+ sut_node: SutNode, execution: ExecutionConfiguration, result: DTSResult
+) -> None:
"""
Run the given execution. This involves running the execution setup as well as
running all build targets in the given execution.
"""
dts_logger.info(f"Running execution with SUT '{execution.system_under_test.name}'.")
+ execution_result = result.add_execution(sut_node.config)
try:
sut_node.set_up_execution(execution)
+ execution_result.update_setup(Result.PASS)
except Exception as e:
dts_logger.exception("Execution setup failed.")
- errors.append(e)
+ execution_result.update_setup(Result.FAIL, e)
else:
for build_target in execution.build_targets:
- _run_build_target(sut_node, build_target, execution)
+ _run_build_target(sut_node, build_target, execution, execution_result)
finally:
try:
sut_node.tear_down_execution()
+ execution_result.update_teardown(Result.PASS)
except Exception as e:
dts_logger.exception("Execution teardown failed.")
- errors.append(e)
+ execution_result.update_teardown(Result.FAIL, e)
def _run_build_target(
sut_node: SutNode,
build_target: BuildTargetConfiguration,
execution: ExecutionConfiguration,
+ execution_result: ExecutionResult,
) -> None:
"""
Run the given build target.
"""
dts_logger.info(f"Running build target '{build_target.name}'.")
+ build_target_result = execution_result.add_build_target(build_target)
try:
sut_node.set_up_build_target(build_target)
+ result.dpdk_version = sut_node.dpdk_version
+ build_target_result.update_setup(Result.PASS)
except Exception as e:
dts_logger.exception("Build target setup failed.")
- errors.append(e)
+ build_target_result.update_setup(Result.FAIL, e)
else:
- _run_suites(sut_node, execution)
+ _run_suites(sut_node, execution, build_target_result)
finally:
try:
sut_node.tear_down_build_target()
+ build_target_result.update_teardown(Result.PASS)
except Exception as e:
dts_logger.exception("Build target teardown failed.")
- errors.append(e)
+ build_target_result.update_teardown(Result.FAIL, e)
def _run_suites(
sut_node: SutNode,
execution: ExecutionConfiguration,
+ build_target_result: BuildTargetResult,
) -> None:
"""
Use the given build_target to run execution's test suites
@@ -143,12 +156,15 @@ def _run_suites(
)
except Exception as e:
dts_logger.exception("An error occurred when searching for test suites.")
- errors.append(e)
+ result.update_setup(Result.ERROR, e)
else:
for test_suite_class in test_suite_classes:
test_suite = test_suite_class(
- sut_node, test_suite_config.test_cases, execution.func, errors
+ sut_node,
+ test_suite_config.test_cases,
+ execution.func,
+ build_target_result,
)
test_suite.run()
@@ -157,20 +173,8 @@ def _exit_dts() -> None:
"""
Process all errors and exit with the proper exit code.
"""
- if errors and dts_logger:
- dts_logger.debug("Summary of errors:")
- for error in errors:
- dts_logger.debug(repr(error))
-
- return_code = ErrorSeverity.NO_ERR
- for error in errors:
- error_return_code = ErrorSeverity.GENERIC_ERR
- if isinstance(error, DTSError):
- error_return_code = error.severity
-
- if error_return_code > return_code:
- return_code = error_return_code
+ result.process()
if dts_logger:
dts_logger.info("DTS execution has ended.")
- sys.exit(return_code)
+ sys.exit(result.get_return_code())
new file mode 100644
@@ -0,0 +1,316 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2023 PANTHEON.tech s.r.o.
+
+"""
+Generic result container and reporters
+"""
+
+import os.path
+from collections.abc import MutableSequence
+from enum import Enum, auto
+
+from .config import (
+ OS,
+ Architecture,
+ BuildTargetConfiguration,
+ Compiler,
+ CPUType,
+ NodeConfiguration,
+)
+from .exception import DTSError, ErrorSeverity
+from .logger import DTSLOG
+from .settings import SETTINGS
+
+
+class Result(Enum):
+ """
+ An Enum defining the possible states that
+ a setup, a teardown or a test case may end up in.
+ """
+
+ PASS = auto()
+ FAIL = auto()
+ ERROR = auto()
+ SKIP = auto()
+
+ def __bool__(self) -> bool:
+ return self is self.PASS
+
+
+class FixtureResult(object):
+ """
+ A record that stored the result of a setup or a teardown.
+ The default is FAIL because immediately after creating the object
+ the setup of the corresponding stage will be executed, which also guarantees
+ the execution of teardown.
+ """
+
+ result: Result
+ error: Exception | None = None
+
+ def __init__(
+ self,
+ result: Result = Result.FAIL,
+ error: Exception | None = None,
+ ):
+ self.result = result
+ self.error = error
+
+ def __bool__(self) -> bool:
+ return bool(self.result)
+
+
+class Statistics(dict):
+ """
+ A helper class used to store the number of test cases by its result
+ along a few other basic information.
+ Using a dict provides a convenient way to format the data.
+ """
+
+ def __init__(self, dpdk_version):
+ super(Statistics, self).__init__()
+ for result in Result:
+ self[result.name] = 0
+ self["PASS RATE"] = 0.0
+ self["DPDK VERSION"] = dpdk_version
+
+ def __iadd__(self, other: Result) -> "Statistics":
+ """
+ Add a Result to the final count.
+ """
+ self[other.name] += 1
+ self["PASS RATE"] = (
+ float(self[Result.PASS.name])
+ * 100
+ / sum(self[result.name] for result in Result)
+ )
+ return self
+
+ def __str__(self) -> str:
+ """
+ Provide a string representation of the data.
+ """
+ stats_str = ""
+ for key, value in self.items():
+ stats_str += f"{key:<12} = {value}\n"
+ # according to docs, we should use \n when writing to text files
+ # on all platforms
+ return stats_str
+
+
+class BaseResult(object):
+ """
+ The Base class for all results. Stores the results of
+ the setup and teardown portions of the corresponding stage
+ and a list of results from each inner stage in _inner_results.
+ """
+
+ setup_result: FixtureResult
+ teardown_result: FixtureResult
+ _inner_results: MutableSequence["BaseResult"]
+
+ def __init__(self):
+ self.setup_result = FixtureResult()
+ self.teardown_result = FixtureResult()
+ self._inner_results = []
+
+ def update_setup(self, result: Result, error: Exception | None = None) -> None:
+ self.setup_result.result = result
+ self.setup_result.error = error
+
+ def update_teardown(self, result: Result, error: Exception | None = None) -> None:
+ 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_inner_errors(self) -> list[Exception]:
+ return [
+ error
+ for inner_result in self._inner_results
+ for error in inner_result.get_errors()
+ ]
+
+ def get_errors(self) -> list[Exception]:
+ return self._get_setup_teardown_errors() + self._get_inner_errors()
+
+ def add_stats(self, statistics: Statistics) -> None:
+ for inner_result in self._inner_results:
+ inner_result.add_stats(statistics)
+
+
+class TestCaseResult(BaseResult, FixtureResult):
+ """
+ The test case specific result.
+ Stores the result of the actual test case.
+ Also stores the test case name.
+ """
+
+ test_case_name: str
+
+ def __init__(self, test_case_name: str):
+ super(TestCaseResult, self).__init__()
+ self.test_case_name = test_case_name
+
+ def update(self, result: Result, error: Exception | None = None) -> None:
+ self.result = result
+ self.error = error
+
+ def _get_inner_errors(self) -> list[Exception]:
+ if self.error:
+ return [self.error]
+ return []
+
+ def add_stats(self, statistics: Statistics) -> None:
+ statistics += self.result
+
+ def __bool__(self) -> bool:
+ return (
+ bool(self.setup_result) and bool(self.teardown_result) and bool(self.result)
+ )
+
+
+class TestSuiteResult(BaseResult):
+ """
+ The test suite specific result.
+ The _inner_results list stores results of test cases in a given test suite.
+ Also stores the test suite name.
+ """
+
+ suite_name: str
+
+ def __init__(self, suite_name: str):
+ super(TestSuiteResult, self).__init__()
+ self.suite_name = suite_name
+
+ def add_test_case(self, test_case_name: str) -> TestCaseResult:
+ test_case_result = TestCaseResult(test_case_name)
+ self._inner_results.append(test_case_result)
+ return test_case_result
+
+
+class BuildTargetResult(BaseResult):
+ """
+ The build target specific result.
+ The _inner_results list stores results of test suites in a given build target.
+ Also stores build target specifics, such as compiler used to build DPDK.
+ """
+
+ arch: Architecture
+ os: OS
+ cpu: CPUType
+ compiler: Compiler
+
+ def __init__(self, build_target: BuildTargetConfiguration):
+ super(BuildTargetResult, self).__init__()
+ self.arch = build_target.arch
+ self.os = build_target.os
+ self.cpu = build_target.cpu
+ self.compiler = build_target.compiler
+
+ def add_test_suite(self, test_suite_name: str) -> TestSuiteResult:
+ test_suite_result = TestSuiteResult(test_suite_name)
+ self._inner_results.append(test_suite_result)
+ return test_suite_result
+
+
+class ExecutionResult(BaseResult):
+ """
+ The execution specific result.
+ The _inner_results list stores results of build targets in a given execution.
+ Also stores the SUT node configuration.
+ """
+
+ sut_node: NodeConfiguration
+
+ def __init__(self, sut_node: NodeConfiguration):
+ super(ExecutionResult, self).__init__()
+ self.sut_node = sut_node
+
+ def add_build_target(
+ self, build_target: BuildTargetConfiguration
+ ) -> BuildTargetResult:
+ build_target_result = BuildTargetResult(build_target)
+ self._inner_results.append(build_target_result)
+ return build_target_result
+
+
+class DTSResult(BaseResult):
+ """
+ Stores environment information and test results from a DTS run, which are:
+ * Execution level information, such as SUT and TG hardware.
+ * Build target level information, such as compiler, target OS and cpu.
+ * Test suite results.
+ * All errors that are caught and recorded during DTS execution.
+
+ The information is stored in nested objects.
+
+ The class is capable of computing the return code used to exit DTS with
+ from the stored error.
+
+ It also provides a brief statistical summary of passed/failed test cases.
+ """
+
+ dpdk_version: str | None
+ _logger: DTSLOG
+ _errors: list[Exception]
+ _return_code: ErrorSeverity
+ _stats_result: Statistics | None
+ _stats_filename: str
+
+ def __init__(self, logger: DTSLOG):
+ super(DTSResult, self).__init__()
+ self.dpdk_version = None
+ self._logger = logger
+ self._errors = []
+ self._return_code = ErrorSeverity.NO_ERR
+ self._stats_result = None
+ self._stats_filename = os.path.join(SETTINGS.output_dir, "statistics.txt")
+
+ def add_execution(self, sut_node: NodeConfiguration) -> ExecutionResult:
+ execution_result = ExecutionResult(sut_node)
+ self._inner_results.append(execution_result)
+ return execution_result
+
+ def add_error(self, error) -> None:
+ self._errors.append(error)
+
+ def process(self) -> None:
+ """
+ Process the data after a DTS run.
+ The data is added to nested objects during runtime and this parent object
+ is not updated at that time. This requires us to process the nested data
+ after it's all been gathered.
+
+ The processing gathers all errors and the result statistics of test cases.
+ """
+ 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))
+
+ self._stats_result = Statistics(self.dpdk_version)
+ self.add_stats(self._stats_result)
+ with open(self._stats_filename, "w+") as stats_file:
+ stats_file.write(str(self._stats_result))
+
+ def get_return_code(self) -> int:
+ """
+ Go through all stored Exceptions and return 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)
@@ -9,12 +9,12 @@
import importlib
import inspect
import re
-from collections.abc import MutableSequence
from types import MethodType
from .exception import ConfigurationError, SSHTimeoutError, TestCaseVerifyError
from .logger import DTSLOG, getLogger
from .settings import SETTINGS
+from .test_result import BuildTargetResult, Result, TestCaseResult, TestSuiteResult
from .testbed_model import SutNode
@@ -40,21 +40,21 @@ class TestSuite(object):
_logger: DTSLOG
_test_cases_to_run: list[str]
_func: bool
- _errors: MutableSequence[Exception]
+ _result: TestSuiteResult
def __init__(
self,
sut_node: SutNode,
test_cases: list[str],
func: bool,
- errors: MutableSequence[Exception],
+ build_target_result: BuildTargetResult,
):
self.sut_node = sut_node
self._logger = getLogger(self.__class__.__name__)
self._test_cases_to_run = test_cases
self._test_cases_to_run.extend(SETTINGS.test_cases)
self._func = func
- self._errors = errors
+ self._result = build_target_result.add_test_suite(self.__class__.__name__)
def set_up_suite(self) -> None:
"""
@@ -97,10 +97,11 @@ def run(self) -> None:
try:
self._logger.info(f"Starting test suite setup: {test_suite_name}")
self.set_up_suite()
+ self._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}")
- self._errors.append(e)
+ self._result.update_setup(Result.ERROR, e)
else:
self._execute_test_suite()
@@ -109,13 +110,14 @@ def run(self) -> None:
try:
self.tear_down_suite()
self.sut_node.kill_cleanup_dpdk_apps()
+ self._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, "
f"the next test suite may be affected."
)
- self._errors.append(e)
+ self._result.update_setup(Result.ERROR, e)
def _execute_test_suite(self) -> None:
"""
@@ -123,17 +125,18 @@ def _execute_test_suite(self) -> None:
"""
if self._func:
for test_case_method in self._get_functional_test_cases():
+ test_case_name = test_case_method.__name__
+ test_case_result = self._result.add_test_case(test_case_name)
all_attempts = SETTINGS.re_run + 1
attempt_nr = 1
- while (
- not self._run_test_case(test_case_method)
- and attempt_nr <= all_attempts
- ):
+ self._run_test_case(test_case_method, 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_method.__name__}'. "
+ f"Re-running FAILED test case '{test_case_name}'. "
f"Attempt number {attempt_nr} out of {all_attempts}."
)
+ self._run_test_case(test_case_method, test_case_result)
def _get_functional_test_cases(self) -> list[MethodType]:
"""
@@ -166,68 +169,69 @@ def _should_be_executed(self, test_case_name: str, test_case_regex: str) -> bool
return match
- def _run_test_case(self, test_case_method: MethodType) -> bool:
+ def _run_test_case(
+ self, test_case_method: MethodType, test_case_result: TestCaseResult
+ ) -> None:
"""
Setup, execute and teardown a test case in this suite.
- Exceptions are caught and recorded in logs.
+ Exceptions are caught and recorded in logs and results.
"""
test_case_name = test_case_method.__name__
- result = False
try:
# run set_up function for each case
self.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}")
- self._errors.append(e)
+ test_case_result.update_setup(Result.FAIL, e)
except Exception as e:
self._logger.exception(f"Test case setup ERROR: {test_case_name}")
- self._errors.append(e)
+ test_case_result.update_setup(Result.ERROR, e)
else:
# run test case if setup was successful
- result = self._execute_test_case(test_case_method)
+ self._execute_test_case(test_case_method, test_case_result)
finally:
try:
self.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."
)
- self._errors.append(e)
- result = False
+ test_case_result.update_teardown(Result.ERROR, e)
+ test_case_result.update(Result.ERROR)
- return result
-
- def _execute_test_case(self, test_case_method: MethodType) -> bool:
+ def _execute_test_case(
+ self, test_case_method: MethodType, test_case_result: TestCaseResult
+ ) -> None:
"""
Execute one test case and handle failures.
"""
test_case_name = test_case_method.__name__
- result = False
try:
self._logger.info(f"Starting test case execution: {test_case_name}")
test_case_method()
- result = True
+ 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}")
- self._errors.append(e)
+ test_case_result.update(Result.FAIL, e)
except Exception as e:
self._logger.exception(f"Test case execution ERROR: {test_case_name}")
- self._errors.append(e)
+ 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")
- return result
-
def get_test_suites(testsuite_module_path: str) -> list[type[TestSuite]]:
def is_test_suite(object) -> bool: