@@ -36,7 +36,7 @@
import json
import os.path
import pathlib
-from dataclasses import dataclass
+from dataclasses import dataclass, fields
from enum import auto, unique
from typing import Union
@@ -506,6 +506,28 @@ def from_dict(
vdevs=vdevs,
)
+ def copy_and_modify(self, **kwargs) -> "ExecutionConfiguration":
+ """Create a shallow copy with any of the fields modified.
+
+ The only new data are those passed to this method.
+ The rest are copied from the object's fields calling the method.
+
+ Args:
+ **kwargs: The names and types of keyword arguments are defined
+ by the fields of the :class:`ExecutionConfiguration` class.
+
+ Returns:
+ The copied and modified execution configuration.
+ """
+ new_config = {}
+ for field in fields(self):
+ if field.name in kwargs:
+ new_config[field.name] = kwargs[field.name]
+ else:
+ new_config[field.name] = getattr(self, field.name)
+
+ return ExecutionConfiguration(**new_config)
+
@dataclass(slots=True, frozen=True)
class Configuration:
@@ -197,7 +197,7 @@
},
"cases": {
"type": "array",
- "description": "If specified, only this subset of test suite's test cases will be run. Unknown test cases will be silently ignored.",
+ "description": "If specified, only this subset of test suite's test cases will be run.",
"items": {
"type": "string"
},
@@ -17,17 +17,27 @@
and the test case stage runs test cases individually.
"""
+import importlib
+import inspect
import logging
+import re
import sys
from types import MethodType
+from typing import Iterable
from .config import (
BuildTargetConfiguration,
+ Configuration,
ExecutionConfiguration,
TestSuiteConfig,
load_config,
)
-from .exception import BlockingTestSuiteError, SSHTimeoutError, TestCaseVerifyError
+from .exception import (
+ BlockingTestSuiteError,
+ ConfigurationError,
+ SSHTimeoutError,
+ TestCaseVerifyError,
+)
from .logger import DTSLOG, getLogger
from .settings import SETTINGS
from .test_result import (
@@ -37,8 +47,9 @@
Result,
TestCaseResult,
TestSuiteResult,
+ TestSuiteWithCases,
)
-from .test_suite import TestSuite, get_test_suites
+from .test_suite import TestSuite
from .testbed_model import SutNode, TGNode
@@ -59,13 +70,23 @@ class DTSRunner:
given execution, the next execution begins.
"""
+ _configuration: Configuration
_logger: DTSLOG
_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 logger and result."""
+ """Initialize the instance with configuration, logger, result and string constants."""
+ self._configuration = load_config()
self._logger = getLogger("DTSRunner")
self._result = DTSResult(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):
"""Run all build targets in all executions from the test run configuration.
@@ -106,29 +127,32 @@ def run(self):
try:
# check the python version of the server that runs dts
self._check_dts_python_version()
+ self._result.update_setup(Result.PASS)
# for all Execution sections
- for execution in load_config().executions:
- sut_node = sut_nodes.get(execution.system_under_test_node.name)
- tg_node = tg_nodes.get(execution.traffic_generator_node.name)
-
+ for execution in self._configuration.executions:
+ self._logger.info(
+ f"Running execution with SUT '{execution.system_under_test_node.name}'."
+ )
+ execution_result = self._result.add_execution(execution.system_under_test_node)
+ # we don't want to modify the original config, so create a copy
+ execution_test_suites = list(execution.test_suites)
+ if not execution.skip_smoke_tests:
+ execution_test_suites[:0] = [TestSuiteConfig.from_dict("smoke_tests")]
try:
- if not sut_node:
- sut_node = SutNode(execution.system_under_test_node)
- sut_nodes[sut_node.name] = sut_node
- if not tg_node:
- tg_node = TGNode(execution.traffic_generator_node)
- tg_nodes[tg_node.name] = tg_node
- self._result.update_setup(Result.PASS)
+ test_suites_with_cases = self._get_test_suites_with_cases(
+ execution_test_suites, execution.func, execution.perf
+ )
except Exception as e:
- failed_node = execution.system_under_test_node.name
- if sut_node:
- failed_node = execution.traffic_generator_node.name
- self._logger.exception(f"The Creation of node {failed_node} failed.")
- self._result.update_setup(Result.FAIL, e)
+ self._logger.exception(
+ f"Invalid test suite configuration found: " f"{execution_test_suites}."
+ )
+ execution_result.update_setup(Result.FAIL, e)
else:
- self._run_execution(sut_node, tg_node, execution)
+ self._connect_nodes_and_run_execution(
+ sut_nodes, tg_nodes, execution, execution_result, test_suites_with_cases
+ )
except Exception as e:
self._logger.exception("An unexpected error has occurred.")
@@ -163,11 +187,206 @@ def _check_dts_python_version(self) -> None:
)
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]:
+ """Test suites with test cases discovery.
+
+ The test suites with test cases defined in the user configuration are discovered
+ and stored for future use so that we don't import the modules twice and so that
+ the list of test suites with test cases is available for recording right away.
+
+ 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 discovered test suites, each with test cases.
+ """
+ test_suites_with_cases = []
+
+ for test_suite_config in test_suite_configs:
+ test_suite_class = self._get_test_suite_class(test_suite_config.test_suite)
+ test_cases = []
+ func_test_cases, perf_test_cases = self._filter_test_cases(
+ test_suite_class, set(test_suite_config.test_cases + SETTINGS.test_cases)
+ )
+ 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 _get_test_suite_class(self, module_name: str) -> type[TestSuite]:
+ """Find the :class:`TestSuite` class in `module_name`.
+
+ The full module name is `module_name` prefixed with `self._test_suite_module_prefix`.
+ The module name is a standard filename with words separated with underscores.
+ Search the `module_name` for a :class:`TestSuite` class which starts
+ with `self._test_suite_class_prefix`, continuing with CamelCase `module_name`.
+ The first matching class is returned.
+
+ The CamelCase convention is not tested, only lowercase strings are compared.
+
+ Args:
+ module_name: The module name without prefix where to search for the test suite.
+
+ Returns:
+ The found test suite class.
+
+ Raises:
+ ConfigurationError: If the corresponding module is not found or
+ a valid :class:`TestSuite` is not found in the module.
+ """
+
+ def is_test_suite(object) -> bool:
+ """Check whether `object` is a :class:`TestSuite`.
+
+ The `object` is a subclass of :class:`TestSuite`, but not :class:`TestSuite` itself.
+
+ Args:
+ object: The object to be checked.
+
+ Returns:
+ :data:`True` if `object` is a subclass of `TestSuite`.
+ """
+ try:
+ if issubclass(object, TestSuite) and object is not TestSuite:
+ return True
+ except TypeError:
+ return False
+ return False
+
+ testsuite_module_path = f"{self._test_suite_module_prefix}{module_name}"
+ try:
+ test_suite_module = importlib.import_module(testsuite_module_path)
+ except ModuleNotFoundError as e:
+ raise ConfigurationError(
+ f"Test suite module '{testsuite_module_path}' not found."
+ ) from e
+
+ lowercase_suite_to_find = (
+ f"{self._test_suite_class_prefix}{module_name.replace('_', '')}".lower()
+ )
+ for class_name, class_obj in inspect.getmembers(test_suite_module, is_test_suite):
+ if (
+ class_name.startswith(self._test_suite_class_prefix)
+ and lowercase_suite_to_find == class_name.lower()
+ ):
+ return class_obj
+ raise ConfigurationError(
+ f"Couldn't find any valid test suites in {test_suite_module.__name__}."
+ )
+
+ def _filter_test_cases(
+ self, test_suite_class: type[TestSuite], test_cases_to_run: set[str]
+ ) -> tuple[list[MethodType], list[MethodType]]:
+ """Filter `test_cases_to_run` from `test_suite_class`.
+
+ There are two rounds of filtering if `test_cases_to_run` is not empty.
+ The first filters `test_cases_to_run` from all methods of `test_suite_class`.
+ Then the methods are separated into functional and performance test cases.
+ If a method matches neither the functional nor performance name prefix, it's an error.
+
+ Args:
+ test_suite_class: The class of the test suite.
+ test_cases_to_run: Test case names to filter from `test_suite_class`.
+ If empty, return all matching test cases.
+
+ Returns:
+ A list of test case methods that should be executed.
+
+ Raises:
+ ConfigurationError: If a test case from `test_cases_to_run` is not found
+ or it doesn't match either the functional nor performance name prefix.
+ """
+ func_test_cases = []
+ perf_test_cases = []
+ name_method_tuples = inspect.getmembers(test_suite_class, inspect.isfunction)
+ if test_cases_to_run:
+ name_method_tuples = [
+ (name, method) for name, method in name_method_tuples if name in test_cases_to_run
+ ]
+ if len(name_method_tuples) < len(test_cases_to_run):
+ missing_test_cases = test_cases_to_run - {name for name, _ in name_method_tuples}
+ raise ConfigurationError(
+ f"Test cases {missing_test_cases} not found among methods "
+ f"of {test_suite_class.__name__}."
+ )
+
+ for test_case_name, test_case_method in name_method_tuples:
+ if re.match(self._func_test_case_regex, test_case_name):
+ func_test_cases.append(test_case_method)
+ elif re.match(self._perf_test_case_regex, test_case_name):
+ perf_test_cases.append(test_case_method)
+ elif test_cases_to_run:
+ raise ConfigurationError(
+ f"Method '{test_case_name}' matches neither "
+ f"a functional nor a performance test case name."
+ )
+
+ return func_test_cases, perf_test_cases
+
+ def _connect_nodes_and_run_execution(
+ self,
+ sut_nodes: dict[str, SutNode],
+ tg_nodes: dict[str, TGNode],
+ execution: ExecutionConfiguration,
+ execution_result: ExecutionResult,
+ test_suites_with_cases: Iterable[TestSuiteWithCases],
+ ) -> None:
+ """Connect nodes, then continue to run the given execution.
+
+ Connect the :class:`SutNode` and the :class:`TGNode` of this `execution`.
+ 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.
+ execution: An execution's test run configuration.
+ execution_result: The execution's result.
+ test_suites_with_cases: The test suites with test cases to run.
+ """
+ sut_node = sut_nodes.get(execution.system_under_test_node.name)
+ tg_node = tg_nodes.get(execution.traffic_generator_node.name)
+
+ try:
+ if not sut_node:
+ sut_node = SutNode(execution.system_under_test_node)
+ sut_nodes[sut_node.name] = sut_node
+ if not tg_node:
+ tg_node = TGNode(execution.traffic_generator_node)
+ tg_nodes[tg_node.name] = tg_node
+ except Exception as e:
+ failed_node = execution.system_under_test_node.name
+ if sut_node:
+ failed_node = execution.traffic_generator_node.name
+ self._logger.exception(f"The Creation of node {failed_node} failed.")
+ execution_result.update_setup(Result.FAIL, e)
+
+ else:
+ self._run_execution(
+ sut_node, tg_node, execution, execution_result, test_suites_with_cases
+ )
+
def _run_execution(
self,
sut_node: SutNode,
tg_node: TGNode,
execution: ExecutionConfiguration,
+ execution_result: ExecutionResult,
+ test_suites_with_cases: Iterable[TestSuiteWithCases],
) -> None:
"""Run the given execution.
@@ -178,11 +397,11 @@ def _run_execution(
sut_node: The execution's SUT node.
tg_node: The execution's TG node.
execution: An execution's test run configuration.
+ execution_result: The execution's result.
+ test_suites_with_cases: The test suites with test cases to run.
"""
self._logger.info(f"Running execution with SUT '{execution.system_under_test_node.name}'.")
- execution_result = self._result.add_execution(sut_node.config)
execution_result.add_sut_info(sut_node.node_info)
-
try:
sut_node.set_up_execution(execution)
execution_result.update_setup(Result.PASS)
@@ -192,7 +411,10 @@ def _run_execution(
else:
for build_target in execution.build_targets:
- self._run_build_target(sut_node, tg_node, build_target, execution, execution_result)
+ build_target_result = execution_result.add_build_target(build_target)
+ self._run_build_target(
+ sut_node, tg_node, build_target, build_target_result, test_suites_with_cases
+ )
finally:
try:
@@ -207,8 +429,8 @@ def _run_build_target(
sut_node: SutNode,
tg_node: TGNode,
build_target: BuildTargetConfiguration,
- execution: ExecutionConfiguration,
- execution_result: ExecutionResult,
+ build_target_result: BuildTargetResult,
+ test_suites_with_cases: Iterable[TestSuiteWithCases],
) -> None:
"""Run the given build target.
@@ -220,11 +442,11 @@ def _run_build_target(
sut_node: The execution's sut node.
tg_node: The execution's tg node.
build_target: A build target's test run configuration.
- execution: The build target's execution's test run configuration.
- execution_result: The execution level result object associated with the execution.
+ build_target_result: The build target level result object associated
+ with the current build target.
+ test_suites_with_cases: The test suites with test cases to run.
"""
self._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)
@@ -236,7 +458,7 @@ def _run_build_target(
build_target_result.update_setup(Result.FAIL, e)
else:
- self._run_test_suites(sut_node, tg_node, execution, build_target_result)
+ self._run_test_suites(sut_node, tg_node, build_target_result, test_suites_with_cases)
finally:
try:
@@ -250,10 +472,10 @@ def _run_test_suites(
self,
sut_node: SutNode,
tg_node: TGNode,
- execution: ExecutionConfiguration,
build_target_result: BuildTargetResult,
+ test_suites_with_cases: Iterable[TestSuiteWithCases],
) -> None:
- """Run the execution's (possibly a subset of) test suites using the current build target.
+ """Run `test_suites_with_cases` with the current build target.
The method assumes the build target we're testing has already been built on the SUT node.
The current build target thus corresponds to the current DPDK build present on the SUT node.
@@ -264,22 +486,20 @@ def _run_test_suites(
Args:
sut_node: The execution's SUT node.
tg_node: The execution's TG node.
- execution: The execution's test run configuration associated
- with the current build target.
build_target_result: The build target level result object associated
with the current build target.
+ test_suites_with_cases: The test suites with test cases to run.
"""
end_build_target = False
- if not execution.skip_smoke_tests:
- execution.test_suites[:0] = [TestSuiteConfig.from_dict("smoke_tests")]
- for test_suite_config in execution.test_suites:
+ for test_suite_with_cases in test_suites_with_cases:
+ test_suite_result = build_target_result.add_test_suite(
+ test_suite_with_cases.test_suite_class.__name__
+ )
try:
- self._run_test_suite_module(
- sut_node, tg_node, execution, build_target_result, test_suite_config
- )
+ self._run_test_suite(sut_node, tg_node, test_suite_result, test_suite_with_cases)
except BlockingTestSuiteError as e:
self._logger.exception(
- f"An error occurred within {test_suite_config.test_suite}. "
+ f"An error occurred within {test_suite_with_cases.test_suite_class.__name__}. "
"Skipping build target..."
)
self._result.add_error(e)
@@ -288,15 +508,14 @@ def _run_test_suites(
if end_build_target:
break
- def _run_test_suite_module(
+ def _run_test_suite(
self,
sut_node: SutNode,
tg_node: TGNode,
- execution: ExecutionConfiguration,
- build_target_result: BuildTargetResult,
- test_suite_config: TestSuiteConfig,
+ test_suite_result: TestSuiteResult,
+ test_suite_with_cases: TestSuiteWithCases,
) -> None:
- """Set up, execute and tear down all test suites in a single test suite module.
+ """Set up, execute and tear down `test_suite_with_cases`.
The method assumes the build target we're testing has already been built on the SUT node.
The current build target thus corresponds to the current DPDK build present on the SUT node.
@@ -306,92 +525,79 @@ def _run_test_suite_module(
Record the setup and the teardown and handle failures.
- The test cases to execute are discovered when creating the :class:`TestSuite` object.
-
Args:
sut_node: The execution's SUT node.
tg_node: The execution's TG node.
- execution: The execution's test run configuration associated
- with the current build target.
- build_target_result: The build target level result object associated
- with the current build target.
- test_suite_config: Test suite test run configuration specifying the test suite module
- and possibly a subset of test cases of test suites in that module.
+ 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__
+ test_suite = test_suite_with_cases.test_suite_class(sut_node, tg_node)
try:
- full_suite_path = f"tests.TestSuite_{test_suite_config.test_suite}"
- test_suite_classes = get_test_suites(full_suite_path)
- suites_str = ", ".join((x.__name__ for x in test_suite_classes))
- self._logger.debug(f"Found test suites '{suites_str}' in '{full_suite_path}'.")
+ 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("An error occurred when searching for test suites.")
- self._result.update_setup(Result.ERROR, e)
+ self._logger.exception(f"Test suite setup ERROR: {test_suite_name}")
+ test_suite_result.update_setup(Result.ERROR, e)
else:
- for test_suite_class in test_suite_classes:
- test_suite = test_suite_class(sut_node, tg_node, test_suite_config.test_cases)
-
- test_suite_name = test_suite.__class__.__name__
- test_suite_result = build_target_result.add_test_suite(test_suite_name)
- 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(execution.func, test_suite, test_suite_result)
-
- finally:
- try:
- 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, "
- f"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)
+ self._execute_test_suite(
+ test_suite,
+ test_suite_with_cases.test_cases,
+ test_suite_result,
+ )
+ finally:
+ try:
+ 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, func: bool, test_suite: TestSuite, test_suite_result: TestSuiteResult
+ self,
+ test_suite: TestSuite,
+ test_cases: Iterable[MethodType],
+ test_suite_result: TestSuiteResult,
) -> None:
- """Execute all discovered test cases in `test_suite`.
+ """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:
- func: Whether to execute functional test cases.
test_suite: The test suite object.
+ test_cases: The list of test case methods.
test_suite_result: The test suite level result object associated
with the current test suite.
"""
- if func:
- for test_case_method in test_suite._get_functional_test_cases():
- test_case_name = test_case_method.__name__
- test_case_result = test_suite_result.add_test_case(test_case_name)
- all_attempts = SETTINGS.re_run + 1
- attempt_nr = 1
+ for test_case_method in test_cases:
+ test_case_name = test_case_method.__name__
+ test_case_result = test_suite_result.add_test_case(test_case_name)
+ all_attempts = SETTINGS.re_run + 1
+ attempt_nr = 1
+ self._run_test_case(test_suite, 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_name}'. "
+ f"Attempt number {attempt_nr} out of {all_attempts}."
+ )
self._run_test_case(test_suite, 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_name}'. "
- f"Attempt number {attempt_nr} out of {all_attempts}."
- )
- self._run_test_case(test_suite, test_case_method, test_case_result)
def _run_test_case(
self,
@@ -399,7 +605,7 @@ def _run_test_case(
test_case_method: MethodType,
test_case_result: TestCaseResult,
) -> None:
- """Setup, execute and teardown a test case in `test_suite`.
+ """Setup, execute and teardown `test_case_method` from `test_suite`.
Record the result of the setup and the teardown and handle failures.
@@ -424,7 +630,7 @@ def _run_test_case(
else:
# run test case if setup was successful
- self._execute_test_case(test_case_method, test_case_result)
+ self._execute_test_case(test_suite, test_case_method, test_case_result)
finally:
try:
@@ -440,11 +646,15 @@ def _run_test_case(
test_case_result.update(Result.ERROR)
def _execute_test_case(
- self, test_case_method: MethodType, test_case_result: TestCaseResult
+ self,
+ test_suite: TestSuite,
+ test_case_method: MethodType,
+ test_case_result: TestCaseResult,
) -> None:
- """Execute one test case, record the result and handle failures.
+ """Execute `test_case_method` from `test_suite`, record the result and handle failures.
Args:
+ test_suite: The test suite object.
test_case_method: The test case method.
test_case_result: The test case level result object associated
with the current test case.
@@ -452,7 +662,7 @@ def _execute_test_case(
test_case_name = test_case_method.__name__
try:
self._logger.info(f"Starting test case execution: {test_case_name}")
- test_case_method()
+ test_case_method(test_suite)
test_case_result.update(Result.PASS)
self._logger.info(f"Test case execution PASSED: {test_case_name}")
@@ -253,8 +253,7 @@ def _get_parser() -> argparse.ArgumentParser:
"--test-cases",
action=_env_arg("DTS_TESTCASES"),
default="",
- help="[DTS_TESTCASES] Comma-separated list of test cases to execute. "
- "Unknown test cases will be silently ignored.",
+ help="[DTS_TESTCASES] Comma-separated list of test cases to execute.",
)
parser.add_argument(
@@ -25,7 +25,9 @@
import os.path
from collections.abc import MutableSequence
+from dataclasses import dataclass
from enum import Enum, auto
+from types import MethodType
from .config import (
OS,
@@ -36,10 +38,42 @@
CPUType,
NodeConfiguration,
NodeInfo,
+ TestSuiteConfig,
)
from .exception import DTSError, ErrorSeverity
from .logger import DTSLOG
from .settings import SETTINGS
+from .test_suite import TestSuite
+
+
+@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.
+ """
+
+ test_suite_class: type[TestSuite]
+ test_cases: list[MethodType]
+
+ 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],
+ )
class Result(Enum):
@@ -11,25 +11,17 @@
* Testbed (SUT, TG) configuration,
* Packet sending and verification,
* Test case verification.
-
-The module also defines a function, :func:`get_test_suites`,
-for gathering test suites from a Python module.
"""
-import importlib
-import inspect
-import re
from ipaddress import IPv4Interface, IPv6Interface, ip_interface
-from types import MethodType
-from typing import Any, ClassVar, Union
+from typing import ClassVar, Union
from scapy.layers.inet import IP # type: ignore[import]
from scapy.layers.l2 import Ether # type: ignore[import]
from scapy.packet import Packet, Padding # type: ignore[import]
-from .exception import ConfigurationError, TestCaseVerifyError
+from .exception import TestCaseVerifyError
from .logger import DTSLOG, getLogger
-from .settings import SETTINGS
from .testbed_model import Port, PortLink, SutNode, TGNode
from .utils import get_packet_summaries
@@ -37,7 +29,6 @@
class TestSuite(object):
"""The base class with building blocks needed by most test cases.
- * Test case filtering and collection,
* Test suite setup/cleanup methods to override,
* Test case setup/cleanup methods to override,
* Test case verification,
@@ -71,7 +62,6 @@ class TestSuite(object):
#: will block the execution of all subsequent test suites in the current build target.
is_blocking: ClassVar[bool] = False
_logger: DTSLOG
- _test_cases_to_run: list[str]
_port_links: list[PortLink]
_sut_port_ingress: Port
_sut_port_egress: Port
@@ -86,24 +76,19 @@ def __init__(
self,
sut_node: SutNode,
tg_node: TGNode,
- test_cases: list[str],
):
"""Initialize the test suite testbed information and basic configuration.
- Process what test cases to run, find links between ports and set up
- default IP addresses to be used when configuring them.
+ Find links between ports and set up default IP addresses to be used when
+ configuring them.
Args:
sut_node: The SUT node where the test suite will run.
tg_node: The TG node where the test suite will run.
- test_cases: The list of test cases to execute.
- If empty, all test cases will be executed.
"""
self.sut_node = sut_node
self.tg_node = tg_node
self._logger = getLogger(self.__class__.__name__)
- self._test_cases_to_run = test_cases
- self._test_cases_to_run.extend(SETTINGS.test_cases)
self._port_links = []
self._process_links()
self._sut_port_ingress, self._tg_port_egress = (
@@ -364,65 +349,3 @@ def _verify_l3_packet(self, received_packet: IP, expected_packet: IP) -> bool:
if received_packet.src != expected_packet.src or received_packet.dst != expected_packet.dst:
return False
return True
-
- def _get_functional_test_cases(self) -> list[MethodType]:
- """Get all functional test cases defined in this TestSuite.
-
- Returns:
- The list of functional test cases of this TestSuite.
- """
- return self._get_test_cases(r"test_(?!perf_)")
-
- def _get_test_cases(self, test_case_regex: str) -> list[MethodType]:
- """Return a list of test cases matching test_case_regex.
-
- Returns:
- The list of test cases matching test_case_regex of this TestSuite.
- """
- self._logger.debug(f"Searching for test cases in {self.__class__.__name__}.")
- filtered_test_cases = []
- for test_case_name, test_case in inspect.getmembers(self, inspect.ismethod):
- if self._should_be_executed(test_case_name, test_case_regex):
- filtered_test_cases.append(test_case)
- cases_str = ", ".join((x.__name__ for x in filtered_test_cases))
- self._logger.debug(f"Found test cases '{cases_str}' in {self.__class__.__name__}.")
- return filtered_test_cases
-
- def _should_be_executed(self, test_case_name: str, test_case_regex: str) -> bool:
- """Check whether the test case should be scheduled to be executed."""
- match = bool(re.match(test_case_regex, test_case_name))
- if self._test_cases_to_run:
- return match and test_case_name in self._test_cases_to_run
-
- return match
-
-
-def get_test_suites(testsuite_module_path: str) -> list[type[TestSuite]]:
- r"""Find all :class:`TestSuite`\s in a Python module.
-
- Args:
- testsuite_module_path: The path to the Python module.
-
- Returns:
- The list of :class:`TestSuite`\s found within the Python module.
-
- Raises:
- ConfigurationError: The test suite module was not found.
- """
-
- def is_test_suite(object: Any) -> bool:
- try:
- if issubclass(object, TestSuite) and object is not TestSuite:
- return True
- except TypeError:
- return False
- return False
-
- try:
- testcase_module = importlib.import_module(testsuite_module_path)
- except ModuleNotFoundError as e:
- raise ConfigurationError(f"Test suite '{testsuite_module_path}' not found.") from e
- return [
- test_suite_class
- for _, test_suite_class in inspect.getmembers(testcase_module, is_test_suite)
- ]
@@ -51,6 +51,9 @@ linters = "mccabe,pycodestyle,pydocstyle,pyflakes"
format = "pylint"
max_line_length = 100
+[tool.pylama.linter.pycodestyle]
+ignore = "E203,W503"
+
[tool.pylama.linter.pydocstyle]
convention = "google"
@@ -21,7 +21,7 @@
from framework.utils import REGEX_FOR_PCI_ADDRESS
-class SmokeTests(TestSuite):
+class TestSmokeTests(TestSuite):
"""DPDK and infrastructure smoke test suite.
The test cases validate the most basic DPDK functionality needed for all other test suites.