@@ -21,11 +21,10 @@
import inspect
import os
import random
-import re
import sys
from pathlib import Path
-from types import FunctionType
-from typing import Iterable, Sequence
+from types import MethodType
+from typing import Iterable
from framework.testbed_model.sut_node import SutNode
from framework.testbed_model.tg_node import TGNode
@@ -54,7 +53,7 @@
TestSuiteResult,
TestSuiteWithCases,
)
-from .test_suite import TestSuite
+from .test_suite import TestCase, TestSuite
class DTSRunner:
@@ -234,9 +233,9 @@ def _get_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, test_suite_config.test_cases
+ test_cases: list[type[TestCase]] = []
+ func_test_cases, perf_test_cases = test_suite_class.get_test_cases(
+ test_suite_config.test_cases
)
if func:
test_cases.extend(func_test_cases)
@@ -311,57 +310,6 @@ def is_test_suite(object) -> bool:
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: Sequence[str]
- ) -> tuple[list[FunctionType], list[FunctionType]]:
- """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 = set(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_test_run(
self,
sut_nodes: dict[str, SutNode],
@@ -609,7 +557,7 @@ def _run_test_suite(
def _execute_test_suite(
self,
test_suite: TestSuite,
- test_cases: Iterable[FunctionType],
+ test_cases: Iterable[type[TestCase]],
test_suite_result: TestSuiteResult,
) -> None:
"""Execute all `test_cases` in `test_suite`.
@@ -620,29 +568,29 @@ def _execute_test_suite(
Args:
test_suite: The test suite object.
- test_cases: The list of test case methods.
+ 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_method in test_cases:
- test_case_name = test_case_method.__name__
+ 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
- self._run_test_case(test_suite, test_case_method, test_case_result)
+ 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_method, test_case_result)
+ self._run_test_case(test_suite, test_case, test_case_result)
def _run_test_case(
self,
test_suite: TestSuite,
- test_case_method: FunctionType,
+ test_case: type[TestCase],
test_case_result: TestCaseResult,
) -> None:
"""Setup, execute and teardown `test_case_method` from `test_suite`.
@@ -651,11 +599,11 @@ def _run_test_case(
Args:
test_suite: The test suite object.
- test_case_method: The test case method.
+ 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_method.__name__
+ test_case_name = test_case.__name__
try:
# run set_up function for each case
@@ -670,7 +618,7 @@ def _run_test_case(
else:
# run test case if setup was successful
- self._execute_test_case(test_suite, test_case_method, test_case_result)
+ self._execute_test_case(test_suite, test_case, test_case_result)
finally:
try:
@@ -688,21 +636,22 @@ def _run_test_case(
def _execute_test_case(
self,
test_suite: TestSuite,
- test_case_method: FunctionType,
+ 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_method: The test case method.
+ 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_method.__name__
+ test_case_name = test_case.__name__
try:
self._logger.info(f"Starting test case execution: {test_case_name}")
- test_case_method(test_suite)
+ # 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}")
@@ -27,7 +27,6 @@
from collections.abc import MutableSequence
from dataclasses import dataclass
from enum import Enum, auto
-from types import FunctionType
from typing import Union
from .config import (
@@ -44,7 +43,7 @@
from .exception import DTSError, ErrorSeverity
from .logger import DTSLogger
from .settings import SETTINGS
-from .test_suite import TestSuite
+from .test_suite import TestCase, TestSuite
@dataclass(slots=True, frozen=True)
@@ -63,7 +62,7 @@ class is to hold a subset of test cases (which could be all test cases) because
"""
test_suite_class: type[TestSuite]
- test_cases: list[FunctionType]
+ test_cases: list[type[TestCase]]
def create_config(self) -> TestSuiteConfig:
"""Generate a :class:`TestSuiteConfig` from the stored test suite with test cases.
@@ -13,9 +13,12 @@
* Test case verification.
"""
+import inspect
from collections import Counter
+from collections.abc import Callable, Sequence
+from enum import Enum, auto
from ipaddress import IPv4Interface, IPv6Interface, ip_interface
-from typing import ClassVar, Union
+from typing import ClassVar, Protocol, TypeVar, Union, cast
from scapy.layers.inet import IP # type: ignore[import-untyped]
from scapy.layers.l2 import Ether # type: ignore[import-untyped]
@@ -28,7 +31,7 @@
PacketFilteringConfig,
)
-from .exception import TestCaseVerifyError
+from .exception import ConfigurationError, TestCaseVerifyError
from .logger import DTSLogger, get_dts_logger
from .utils import get_packet_summaries
@@ -121,6 +124,68 @@ def _process_links(self) -> None:
):
self._port_links.append(PortLink(sut_port=sut_port, tg_port=tg_port))
+ @classmethod
+ def get_test_cases(
+ cls, test_case_sublist: Sequence[str] | None = None
+ ) -> tuple[set[type["TestCase"]], set[type["TestCase"]]]:
+ """Filter `test_case_subset` from this class.
+
+ Test cases are regular (or bound) methods decorated with :func:`func_test`
+ or :func:`perf_test`.
+
+ Args:
+ test_case_sublist: Test case names to filter from this class.
+ If empty or :data:`None`, return all test cases.
+
+ Returns:
+ The filtered test case functions. This method returns functions as opposed to methods,
+ as methods are bound to instances and this method only has access to the class.
+
+ Raises:
+ ConfigurationError: If a test case from `test_case_subset` is not found.
+ """
+
+ def is_test_case(function: Callable) -> bool:
+ if inspect.isfunction(function):
+ # TestCase is not used at runtime, so we can't use isinstance() with `function`.
+ # But function.test_type exists.
+ if hasattr(function, "test_type"):
+ return isinstance(function.test_type, TestCaseType)
+ return False
+
+ if test_case_sublist is None:
+ test_case_sublist = []
+
+ # the copy is needed so that the condition "elif test_case_sublist" doesn't
+ # change mid-cycle
+ test_case_sublist_copy = list(test_case_sublist)
+ func_test_cases = set()
+ perf_test_cases = set()
+
+ for test_case_name, test_case_function in inspect.getmembers(cls, is_test_case):
+ if test_case_name in test_case_sublist_copy:
+ # if test_case_sublist_copy is non-empty, remove the found test case
+ # so that we can look at the remainder at the end
+ test_case_sublist_copy.remove(test_case_name)
+ elif test_case_sublist:
+ # the original list not being empty means we're filtering test cases
+ # since we didn't remove test_case_name in the previous branch,
+ # it doesn't match the filter and we don't want to remove it
+ continue
+
+ match test_case_function.test_type:
+ case TestCaseType.PERFORMANCE:
+ perf_test_cases.add(test_case_function)
+ case TestCaseType.FUNCTIONAL:
+ func_test_cases.add(test_case_function)
+
+ if test_case_sublist_copy:
+ raise ConfigurationError(
+ f"Test cases {test_case_sublist_copy} not found among functions of {cls.__name__}."
+ )
+
+ return func_test_cases, perf_test_cases
+
def set_up_suite(self) -> None:
"""Set up test fixtures common to all test cases.
@@ -425,3 +490,59 @@ 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
+
+
+#: The generic type for a method of an instance of TestSuite
+TestSuiteMethodType = TypeVar("TestSuiteMethodType", bound=Callable[[TestSuite], None])
+
+
+class TestCaseType(Enum):
+ """The types of test cases."""
+
+ #:
+ FUNCTIONAL = auto()
+ #:
+ PERFORMANCE = auto()
+
+
+class TestCase(Protocol[TestSuiteMethodType]):
+ """Definition of the test case type for static type checking purposes.
+
+ The type is applied to test case functions through a decorator, which casts the decorated
+ test case function to :class:`TestCase` and sets common variables.
+ """
+
+ #:
+ test_type: ClassVar[TestCaseType]
+ #: necessary for mypy so that it can treat this class as the function it's shadowing
+ __call__: TestSuiteMethodType
+
+ @classmethod
+ def make_decorator(
+ cls, test_case_type: TestCaseType
+ ) -> Callable[[TestSuiteMethodType], type["TestCase"]]:
+ """Create a decorator for test suites.
+
+ The decorator casts the decorated function as :class:`TestCase`,
+ sets it as `test_case_type`
+ and initializes common variables defined in :class:`RequiresCapabilities`.
+
+ Args:
+ test_case_type: Either a functional or performance test case.
+
+ Returns:
+ The decorator of a functional or performance test case.
+ """
+
+ def _decorator(func: TestSuiteMethodType) -> type[TestCase]:
+ test_case = cast(type[TestCase], func)
+ test_case.test_type = test_case_type
+ return test_case
+
+ return _decorator
+
+
+#: The decorator for functional test cases.
+func_test: Callable = TestCase.make_decorator(TestCaseType.FUNCTIONAL)
+#: The decorator for performance test cases.
+perf_test: Callable = TestCase.make_decorator(TestCaseType.PERFORMANCE)
@@ -8,7 +8,7 @@
"""
from framework.remote_session.dpdk_shell import compute_eal_params
-from framework.test_suite import TestSuite
+from framework.test_suite import TestSuite, func_test
from framework.testbed_model.cpu import (
LogicalCoreCount,
LogicalCoreCountFilter,
@@ -27,7 +27,8 @@ def set_up_suite(self) -> None:
"""
self.app_helloworld_path = self.sut_node.build_dpdk_app("helloworld")
- def test_hello_world_single_core(self) -> None:
+ @func_test
+ def hello_world_single_core(self) -> None:
"""Single core test case.
Steps:
@@ -46,7 +47,8 @@ def test_hello_world_single_core(self) -> None:
f"helloworld didn't start on lcore{lcores[0]}",
)
- def test_hello_world_all_cores(self) -> None:
+ @func_test
+ def hello_world_all_cores(self) -> None:
"""All cores test case.
Steps:
@@ -10,7 +10,7 @@
from scapy.layers.inet import IP, UDP # type: ignore[import-untyped]
from scapy.layers.l2 import Ether # type: ignore[import-untyped]
-from framework.test_suite import TestSuite
+from framework.test_suite import TestSuite, func_test
class TestOsUdp(TestSuite):
@@ -26,6 +26,7 @@ def set_up_suite(self) -> None:
self.sut_node.bind_ports_to_driver(for_dpdk=False)
self.configure_testbed_ipv4()
+ @func_test
def test_os_udp(self) -> None:
"""Basic UDP IPv4 traffic test case.
@@ -24,7 +24,7 @@
from framework.params.testpmd import SimpleForwardingModes
from framework.remote_session.testpmd_shell import TestPmdShell
-from framework.test_suite import TestSuite
+from framework.test_suite import TestSuite, func_test
class TestPmdBufferScatter(TestSuite):
@@ -123,6 +123,7 @@ def pmd_scatter(self, mbsize: int) -> None:
f"{offset}.",
)
+ @func_test
def test_scatter_mbuf_2048(self) -> None:
"""Run the :meth:`pmd_scatter` test with `mbsize` set to 2048."""
self.pmd_scatter(mbsize=2048)
@@ -17,7 +17,7 @@
from framework.config import PortConfig
from framework.remote_session.testpmd_shell import TestPmdShell
from framework.settings import SETTINGS
-from framework.test_suite import TestSuite
+from framework.test_suite import TestSuite, func_test
from framework.utils import REGEX_FOR_PCI_ADDRESS
@@ -47,6 +47,7 @@ def set_up_suite(self) -> None:
self.dpdk_build_dir_path = self.sut_node.remote_dpdk_build_dir
self.nics_in_node = self.sut_node.config.ports
+ @func_test
def test_unit_tests(self) -> None:
"""DPDK meson ``fast-tests`` unit tests.
@@ -63,6 +64,7 @@ def test_unit_tests(self) -> None:
privileged=True,
)
+ @func_test
def test_driver_tests(self) -> None:
"""DPDK meson ``driver-tests`` unit tests.
@@ -91,6 +93,7 @@ def test_driver_tests(self) -> None:
privileged=True,
)
+ @func_test
def test_devices_listed_in_testpmd(self) -> None:
"""Testpmd device discovery.
@@ -108,6 +111,7 @@ def test_devices_listed_in_testpmd(self) -> None:
"please check your configuration",
)
+ @func_test
def test_device_bound_to_driver(self) -> None:
"""Device driver in OS.