@@ -238,7 +238,6 @@ class ExecutionConfiguration:
system_under_test_node: SutNodeConfiguration
traffic_generator_node: TGNodeConfiguration
vdevs: list[str]
- skip_smoke_tests: bool
@staticmethod
def from_dict(
@@ -247,9 +246,10 @@ def from_dict(
build_targets: list[BuildTargetConfiguration] = list(
map(BuildTargetConfiguration.from_dict, d["build_targets"])
)
+ if not d.get("skip_smoke_tests", False):
+ d["test_suites"].insert(0, "smoke_tests")
test_suites: list[TestSuiteConfig] = list(map(TestSuiteConfig.from_dict, d["test_suites"]))
sut_name = d["system_under_test_node"]["node_name"]
- skip_smoke_tests = d.get("skip_smoke_tests", False)
assert sut_name in node_map, f"Unknown SUT {sut_name} in execution {d}"
system_under_test_node = node_map[sut_name]
assert isinstance(
@@ -270,7 +270,6 @@ def from_dict(
build_targets=build_targets,
perf=d["perf"],
func=d["func"],
- skip_smoke_tests=skip_smoke_tests,
test_suites=test_suites,
system_under_test_node=system_under_test_node,
traffic_generator_node=traffic_generator_node,
@@ -3,9 +3,14 @@
# Copyright(c) 2022-2023 PANTHEON.tech s.r.o.
# Copyright(c) 2022-2023 University of New Hampshire
+import importlib
+import inspect
import logging
+import re
import sys
-from types import MethodType
+from copy import deepcopy
+from dataclasses import dataclass
+from types import MethodType, ModuleType
from .config import (
BuildTargetConfiguration,
@@ -13,7 +18,12 @@
ExecutionConfiguration,
TestSuiteConfig,
)
-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 (
@@ -24,25 +34,55 @@
TestCaseResult,
TestSuiteResult,
)
-from .test_suite import TestSuite, get_test_suites
+from .test_suite import TestSuite
from .testbed_model import SutNode, TGNode
from .utils import check_dts_python_version
+@dataclass
+class TestSuiteSetup:
+ test_suite: type[TestSuite]
+ test_cases: list[MethodType]
+
+ def processed_config(self) -> TestSuiteConfig:
+ return TestSuiteConfig(
+ test_suite=self.test_suite.__name__,
+ test_cases=[test_case.__name__ for test_case in self.test_cases],
+ )
+
+
+@dataclass
+class Execution:
+ config: ExecutionConfiguration
+ test_suite_setups: list[TestSuiteSetup]
+
+ def processed_config(self) -> ExecutionConfiguration:
+ """
+ Creating copy of execution config witch add test-case names.
+ """
+ modified_execution_config = deepcopy(self.config)
+ modified_execution_config.test_suites[:] = [
+ test_suite.processed_config() for test_suite in self.test_suite_setups
+ ]
+ return modified_execution_config
+
+
class DTSRunner:
_logger: DTSLOG
_result: DTSResult
- _configuration: Configuration
+ _executions: list[Execution]
def __init__(self, configuration: Configuration):
self._logger = getLogger("DTSRunner")
self._result = DTSResult(self._logger)
- self._configuration = configuration
+ self._executions = create_executions(configuration.executions)
def run(self):
"""
The main process of DTS. Runs all build targets in all executions from the main
config file.
+ Suite execution consists of running all test cases scheduled to be executed.
+ A test case run consists of setup, execution and teardown of said test case.
"""
# check the python version of the server that run dts
check_dts_python_version()
@@ -50,22 +90,22 @@ def run(self):
tg_nodes: dict[str, TGNode] = {}
try:
# for all Execution sections
- for execution in self._configuration.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._executions:
+ sut_node = sut_nodes.get(execution.config.system_under_test_node.name)
+ tg_node = tg_nodes.get(execution.config.traffic_generator_node.name)
try:
if not sut_node:
- sut_node = SutNode(execution.system_under_test_node)
+ sut_node = SutNode(execution.config.system_under_test_node)
sut_nodes[sut_node.name] = sut_node
if not tg_node:
- tg_node = TGNode(execution.traffic_generator_node)
+ tg_node = TGNode(execution.config.traffic_generator_node)
tg_nodes[tg_node.name] = tg_node
self._result.update_setup(Result.PASS)
except Exception as e:
- failed_node = execution.system_under_test_node.name
+ failed_node = execution.config.system_under_test_node.name
if sut_node:
- failed_node = execution.traffic_generator_node.name
+ failed_node = execution.config.traffic_generator_node.name
self._logger.exception(
f"The Creation of node {failed_node} failed."
)
@@ -100,29 +140,34 @@ def _run_execution(
self,
sut_node: SutNode,
tg_node: TGNode,
- execution: ExecutionConfiguration,
+ execution: Execution,
) -> None:
"""
Run the given execution. This involves running the execution setup as well as
running all build targets in the given execution.
"""
self._logger.info(
- f"Running execution with SUT '{execution.system_under_test_node.name}'."
+ "Running execution with SUT "
+ f"'{execution.config.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)
+ sut_node.set_up_execution(execution.config)
execution_result.update_setup(Result.PASS)
except Exception as e:
self._logger.exception("Execution setup failed.")
execution_result.update_setup(Result.FAIL, e)
else:
- for build_target in execution.build_targets:
+ for build_target in execution.config.build_targets:
self._run_build_target(
- sut_node, tg_node, build_target, execution, execution_result
+ sut_node,
+ tg_node,
+ build_target,
+ execution,
+ execution_result,
)
finally:
@@ -138,7 +183,7 @@ def _run_build_target(
sut_node: SutNode,
tg_node: TGNode,
build_target: BuildTargetConfiguration,
- execution: ExecutionConfiguration,
+ execution: Execution,
execution_result: ExecutionResult,
) -> None:
"""
@@ -171,7 +216,7 @@ def _run_test_suites(
self,
sut_node: SutNode,
tg_node: TGNode,
- execution: ExecutionConfiguration,
+ execution: Execution,
build_target_result: BuildTargetResult,
) -> None:
"""
@@ -180,16 +225,18 @@ def _run_test_suites(
If no subset is specified, run all test cases.
"""
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_setup in execution.test_suite_setups:
try:
self._run_test_suite(
- sut_node, tg_node, execution, build_target_result, test_suite_config
+ sut_node,
+ tg_node,
+ test_suite_setup,
+ build_target_result,
)
except BlockingTestSuiteError as e:
self._logger.exception(
- f"An error occurred within {test_suite_config.test_suite}. "
+ "An error occurred within "
+ f"{test_suite_setup.test_suite.__name__}. "
"Skipping build target..."
)
self._result.add_error(e)
@@ -202,14 +249,10 @@ def _run_test_suite(
self,
sut_node: SutNode,
tg_node: TGNode,
- execution: ExecutionConfiguration,
+ test_suite_setup: TestSuiteSetup,
build_target_result: BuildTargetResult,
- test_suite_config: TestSuiteConfig,
) -> None:
"""Runs a single test suite.
- Setup, execute and teardown the whole suite.
- Suite execution consists of running all test cases scheduled to be executed.
- A test cast run consists of setup, execution and teardown of said test case.
Args:
sut_node: Node to run tests on.
@@ -220,84 +263,67 @@ def _run_test_suite(
Raises:
BlockingTestSuiteError: If a test suite that was marked as blocking fails.
"""
+ test_suite = test_suite_setup.test_suite(sut_node, tg_node)
+ test_suite_name = test_suite_setup.test_suite.__name__
+ test_suite_result = build_target_result.add_test_suite(test_suite_name)
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
- )
+ self._execute_test_suite(
+ test_suite,
+ test_suite_setup.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, "
- 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)
+ 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: list[MethodType],
+ test_suite_result: TestSuiteResult,
) -> None:
"""
Execute all test cases scheduled to be executed in this 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
- 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)
+ 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_case_method, test_suite, 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_case_method, test_suite, test_case_result)
def _run_test_case(
self,
- test_suite: TestSuite,
test_case_method: MethodType,
+ test_suite: TestSuite,
test_case_result: TestCaseResult,
) -> None:
"""
@@ -305,7 +331,6 @@ def _run_test_case(
Exceptions are caught and recorded in logs and results.
"""
test_case_name = test_case_method.__name__
-
try:
# run set_up function for each case
test_suite.set_up_test_case()
@@ -319,7 +344,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_case_method, test_suite, test_case_result)
finally:
try:
@@ -335,7 +360,10 @@ 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_case_method: MethodType,
+ test_suite: TestSuite,
+ test_case_result: TestCaseResult,
) -> None:
"""
Execute one test case and handle failures.
@@ -343,7 +371,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}")
@@ -371,3 +399,92 @@ def _exit_dts(self) -> None:
logging.shutdown()
sys.exit(self._result.get_return_code())
+
+
+def create_executions(
+ execution_configs: list[ExecutionConfiguration],
+) -> list[Execution]:
+ executions: list[Execution] = []
+ for execution_config in execution_configs:
+ test_suite_setups: list[TestSuiteSetup] = []
+
+ for test_suite_config in execution_config.test_suites:
+ testsuite_module_path = f"tests.TestSuite_{test_suite_config.test_suite}"
+ try:
+ suite_module = importlib.import_module(testsuite_module_path)
+ except ModuleNotFoundError as e:
+ raise ConfigurationError(
+ f"Test suite '{testsuite_module_path}' not found."
+ ) from e
+
+ test_suite = _get_suite_class(suite_module, test_suite_config.test_suite)
+
+ test_cases_to_run = test_suite_config.test_cases
+ test_cases_to_run.extend(SETTINGS.test_cases)
+
+ test_cases = []
+ if execution_config.func:
+ # add functional test cases
+ test_cases.extend(
+ _get_test_cases(test_suite, r"test_(?!perf_)", test_cases_to_run)
+ )
+
+ if execution_config.perf:
+ # add performance test cases
+ test_cases.extend(
+ _get_test_cases(test_suite, r"test_perf_", test_cases_to_run)
+ )
+
+ test_suite_setups.append(
+ TestSuiteSetup(test_suite=test_suite, test_cases=test_cases)
+ )
+
+ executions.append(
+ Execution(
+ config=execution_config,
+ test_suite_setups=test_suite_setups,
+ )
+ )
+
+ return executions
+
+
+def _get_suite_class(suite_module: ModuleType, suite_name: str) -> type[TestSuite]:
+ def is_test_suite(object) -> bool:
+ try:
+ if issubclass(object, TestSuite) and object is not TestSuite:
+ return True
+ except TypeError:
+ return False
+ return False
+
+ suite_name_regex = suite_name.replace("_", "").lower()
+ for class_name, suite_class in inspect.getmembers(suite_module, is_test_suite):
+ if not class_name.startswith("Test"):
+ continue
+
+ if suite_name_regex == class_name[4:].lower():
+ return suite_class
+ raise ConfigurationError(
+ f"Cannot find valid test suite in {suite_module.__name__}."
+ )
+
+
+def _get_test_cases(
+ suite_class: type[TestSuite], test_case_regex: str, test_cases_to_run: list[str]
+) -> list[MethodType]:
+ def should_be_executed(test_case_name: str) -> bool:
+ match = bool(re.match(test_case_regex, test_case_name))
+ if test_cases_to_run:
+ return match and test_case_name in test_cases_to_run
+
+ return match
+
+ test_cases = []
+ for test_case_name, test_case_method in inspect.getmembers(
+ suite_class, inspect.isfunction
+ ):
+ if should_be_executed(test_case_name):
+ test_cases.append(test_case_method)
+
+ return test_cases
@@ -6,20 +6,15 @@
Base class for creating DTS test cases.
"""
-import importlib
-import inspect
-import re
from ipaddress import IPv4Interface, IPv6Interface, ip_interface
-from types import MethodType
from typing import 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 SutNode, TGNode
from .testbed_model.hw.port import Port, PortLink
from .utils import get_packet_summaries
@@ -47,7 +42,6 @@ class TestSuite(object):
tg_node: TGNode
is_blocking = False
_logger: DTSLOG
- _test_cases_to_run: list[str]
_port_links: list[PortLink]
_sut_port_ingress: Port
_sut_port_egress: Port
@@ -62,13 +56,10 @@ def __init__(
self,
sut_node: SutNode,
tg_node: TGNode,
- test_cases_to_run: list[str],
):
self.sut_node = sut_node
self.tg_node = tg_node
self._logger = getLogger(self.__class__.__name__)
- self._test_cases_to_run = test_cases_to_run
- self._test_cases_to_run.extend(SETTINGS.test_cases)
self._port_links = []
self._process_links()
self._sut_port_ingress, self._tg_port_egress = (
@@ -268,51 +259,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.
- """
- 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.
- """
- 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 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]]:
- def is_test_suite(object) -> 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)
- ]
@@ -10,7 +10,7 @@
from framework.utils import REGEX_FOR_PCI_ADDRESS
-class SmokeTests(TestSuite):
+class TestSmokeTests(TestSuite):
is_blocking = True
# dicts in this list are expected to have two keys:
# "pci_address" and "current_driver"