"""
import json
@@ -12,11 +38,20 @@
import pathlib
from dataclasses import dataclass
from enum import auto, unique
-from typing import Any, TypedDict, Union
+from typing import Union
import warlock # type: ignore[import]
import yaml
+from framework.config.types import (
+ BuildTargetConfigDict,
+ ConfigurationDict,
+ ExecutionConfigDict,
+ NodeConfigDict,
+ PortConfigDict,
+ TestSuiteConfigDict,
+ TrafficGeneratorConfigDict,
+)
from framework.exception import ConfigurationError
from framework.settings import SETTINGS
from framework.utils import StrEnum
@@ -24,55 +59,97 @@
@unique
class Architecture(StrEnum):
+ r"""The supported architectures of
:class:`~framework.testbed_model.node.Node`\s."""
+
+ #:
i686 = auto()
+ #:
x86_64 = auto()
+ #:
x86_32 = auto()
+ #:
arm64 = auto()
+ #:
ppc64le = auto()
@unique
class OS(StrEnum):
+ r"""The supported operating systems of
:class:`~framework.testbed_model.node.Node`\s."""
+
+ #:
linux = auto()
+ #:
freebsd = auto()
+ #:
windows = auto()
@unique
class CPUType(StrEnum):
+ r"""The supported CPUs of :class:`~framework.testbed_model.node.Node`\s."""
+
+ #:
native = auto()
+ #:
armv8a = auto()
+ #:
dpaa2 = auto()
+ #:
thunderx = auto()
+ #:
xgene1 = auto()
@unique
class Compiler(StrEnum):
+ r"""The supported compilers of
:class:`~framework.testbed_model.node.Node`\s."""
+
+ #:
gcc = auto()
+ #:
clang = auto()
+ #:
icc = auto()
+ #:
msvc = auto()
@unique
class TrafficGeneratorType(StrEnum):
+ """The supported traffic generators."""
+
+ #:
SCAPY = auto()
-# Slots enables some optimizations, by pre-allocating space for the defined
-# attributes in the underlying data structure.
-#
-# Frozen makes the object immutable. This enables further optimizations,
-# and makes it thread safe should we every want to move in that direction.
@dataclass(slots=True, frozen=True)
class HugepageConfiguration:
+ r"""The hugepage configuration of
:class:`~framework.testbed_model.node.Node`\s.
+
+ Attributes:
+ amount: The number of hugepages.
+ force_first_numa: If :data:`True`, the hugepages will be configured on
the first NUMA node.
+ """
+
amount: int
force_first_numa: bool
@dataclass(slots=True, frozen=True)
class PortConfig:
+ r"""The port configuration of
:class:`~framework.testbed_model.node.Node`\s.
+
+ Attributes:
+ node: The :class:`~framework.testbed_model.node.Node` where this port
exists.
+ pci: The PCI address of the port.
+ os_driver_for_dpdk: The operating system driver name for use with DPDK.
+ os_driver: The operating system driver name when the operating system
controls the port.
+ peer_node: The :class:`~framework.testbed_model.node.Node` of the port
+ connected to this port.
+ peer_pci: The PCI address of the port connected to this port.
+ """
+
node: str
pci: str
os_driver_for_dpdk: str
@@ -81,18 +158,44 @@ class PortConfig:
peer_pci: str
@staticmethod
- def from_dict(node: str, d: dict) -> "PortConfig":
+ def from_dict(node: str, d: PortConfigDict) -> "PortConfig":
+ """A convenience method that creates the object from fewer inputs.
+
+ Args:
+ node: The node where this port exists.
+ d: The configuration dictionary.
+
+ Returns:
+ The port configuration instance.
+ """
return PortConfig(node=node, **d)
@dataclass(slots=True, frozen=True)
class TrafficGeneratorConfig:
+ """The configuration of traffic generators.
+
+ The class will be expanded when more configuration is needed.
+
+ Attributes:
+ traffic_generator_type: The type of the traffic generator.
+ """
+
traffic_generator_type: TrafficGeneratorType
@staticmethod
- def from_dict(d: dict) -> "ScapyTrafficGeneratorConfig":
- # This looks useless now, but is designed to allow expansion to traffic
- # generators that require more configuration later.
+ def from_dict(d: TrafficGeneratorConfigDict) ->
"ScapyTrafficGeneratorConfig":
+ """A convenience method that produces traffic generator config of the
proper type.
+
+ Args:
+ d: The configuration dictionary.
+
+ Returns:
+ The traffic generator configuration instance.
+
+ Raises:
+ ConfigurationError: An unknown traffic generator type was
encountered.
+ """
match TrafficGeneratorType(d["type"]):
case TrafficGeneratorType.SCAPY:
return ScapyTrafficGeneratorConfig(
@@ -106,11 +209,31 @@ def from_dict(d: dict) -> "ScapyTrafficGeneratorConfig":
@dataclass(slots=True, frozen=True)
class ScapyTrafficGeneratorConfig(TrafficGeneratorConfig):
+ """Scapy traffic generator specific configuration."""
+
pass
@dataclass(slots=True, frozen=True)
class NodeConfiguration:
+ r"""The configuration of :class:`~framework.testbed_model.node.Node`\s.
+
+ Attributes:
+ name: The name of the :class:`~framework.testbed_model.node.Node`.
+ hostname: The hostname of the
:class:`~framework.testbed_model.node.Node`.
+ Can be an IP or a domain name.
+ user: The name of the user used to connect to
+ the :class:`~framework.testbed_model.node.Node`.
+ password: The password of the user. The use of passwords is heavily
discouraged.
+ Please use keys instead.
+ arch: The architecture of the
:class:`~framework.testbed_model.node.Node`.
+ os: The operating system of the
:class:`~framework.testbed_model.node.Node`.
+ lcores: A comma delimited list of logical cores to use when running
DPDK.
+ use_first_core: If :data:`True`, the first logical core won't be used.
+ hugepages: An optional hugepage configuration.
+ ports: The ports that can be used in testing.
+ """
+
name: str
hostname: str
user: str
@@ -123,57 +246,91 @@ class NodeConfiguration:
ports: list[PortConfig]
@staticmethod
- def from_dict(d: dict) -> Union["SutNodeConfiguration",
"TGNodeConfiguration"]:
- hugepage_config = d.get("hugepages")
- if hugepage_config:
- if "force_first_numa" not in hugepage_config:
- hugepage_config["force_first_numa"] = False
- hugepage_config = HugepageConfiguration(**hugepage_config)
-
- common_config = {
- "name": d["name"],
- "hostname": d["hostname"],
- "user": d["user"],
- "password": d.get("password"),
- "arch": Architecture(d["arch"]),
- "os": OS(d["os"]),
- "lcores": d.get("lcores", "1"),
- "use_first_core": d.get("use_first_core", False),
- "hugepages": hugepage_config,
- "ports": [PortConfig.from_dict(d["name"], port) for port in
d["ports"]],
- }
-
+ def from_dict(
+ d: NodeConfigDict,
+ ) -> Union["SutNodeConfiguration", "TGNodeConfiguration"]:
+ """A convenience method that processes the inputs before creating a
specialized instance.
+
+ Args:
+ d: The configuration dictionary.
+
+ Returns:
+ Either an SUT or TG configuration instance.
+ """
+ hugepage_config = None
+ if "hugepages" in d:
+ hugepage_config_dict = d["hugepages"]
+ if "force_first_numa" not in hugepage_config_dict:
+ hugepage_config_dict["force_first_numa"] = False
+ hugepage_config = HugepageConfiguration(**hugepage_config_dict)
+
+ # The calls here contain duplicated code which is here because Mypy
doesn't
+ # properly support dictionary unpacking with TypedDicts
if "traffic_generator" in d:
return TGNodeConfiguration(
+ name=d["name"],
+ hostname=d["hostname"],
+ user=d["user"],
+ password=d.get("password"),
+ arch=Architecture(d["arch"]),
+ os=OS(d["os"]),
+ lcores=d.get("lcores", "1"),
+ use_first_core=d.get("use_first_core", False),
+ hugepages=hugepage_config,
+ ports=[PortConfig.from_dict(d["name"], port) for port in
d["ports"]],
traffic_generator=TrafficGeneratorConfig.from_dict(
d["traffic_generator"]
),
- **common_config,
)
else:
return SutNodeConfiguration(
- memory_channels=d.get("memory_channels", 1), **common_config
+ name=d["name"],
+ hostname=d["hostname"],
+ user=d["user"],
+ password=d.get("password"),
+ arch=Architecture(d["arch"]),
+ os=OS(d["os"]),
+ lcores=d.get("lcores", "1"),
+ use_first_core=d.get("use_first_core", False),
+ hugepages=hugepage_config,
+ ports=[PortConfig.from_dict(d["name"], port) for port in
d["ports"]],
+ memory_channels=d.get("memory_channels", 1),
)
@dataclass(slots=True, frozen=True)
class SutNodeConfiguration(NodeConfiguration):
+ """:class:`~framework.testbed_model.sut_node.SutNode` specific
configuration.
+
+ Attributes:
+ memory_channels: The number of memory channels to use when running
DPDK.
+ """
+
memory_channels: int
@dataclass(slots=True, frozen=True)
class TGNodeConfiguration(NodeConfiguration):
+ """:class:`~framework.testbed_model.tg_node.TGNode` specific configuration.
+
+ Attributes:
+ traffic_generator: The configuration of the traffic generator present
on the TG node.
+ """
+
traffic_generator: ScapyTrafficGeneratorConfig
@dataclass(slots=True, frozen=True)
class NodeInfo:
- """Class to hold important versions within the node.
-
- This class, unlike the NodeConfiguration class, cannot be generated at the
start.
- This is because we need to initialize a connection with the node before we
can
- collect the information needed in this class. Therefore, it cannot be a
part of
- the configuration class above.
+ """Supplemental node information.
+
+ Attributes:
+ os_name: The name of the running operating system of
+ the :class:`~framework.testbed_model.node.Node`.
+ os_version: The version of the running operating system of
+ the :class:`~framework.testbed_model.node.Node`.
+ kernel_version: The kernel version of the running operating system of
+ the :class:`~framework.testbed_model.node.Node`.
"""
os_name: str
@@ -183,6 +340,20 @@ class NodeInfo:
@dataclass(slots=True, frozen=True)
class BuildTargetConfiguration:
+ """DPDK build configuration.
+
+ The configuration used for building DPDK.
+
+ Attributes:
+ arch: The target architecture to build for.
+ os: The target os to build for.
+ cpu: The target CPU to build for.
+ compiler: The compiler executable to use.
+ compiler_wrapper: This string will be put in front of the compiler when
+ executing the build. Useful for adding wrapper commands, such as
``ccache``.
+ name: The name of the compiler.
+ """
+
arch: Architecture
os: OS
cpu: CPUType
@@ -191,7 +362,18 @@ class BuildTargetConfiguration:
name: str
@staticmethod
- def from_dict(d: dict) -> "BuildTargetConfiguration":
+ def from_dict(d: BuildTargetConfigDict) -> "BuildTargetConfiguration":
+ r"""A convenience method that processes the inputs before creating an
instance.
+
+ `arch`, `os`, `cpu` and `compiler` are converted to :class:`Enum`\s and
+ `name` is constructed from `arch`, `os`, `cpu` and `compiler`.
+
+ Args:
+ d: The configuration dictionary.
+
+ Returns:
+ The build target configuration instance.
+ """
return BuildTargetConfiguration(
arch=Architecture(d["arch"]),
os=OS(d["os"]),
@@ -204,23 +386,29 @@ def from_dict(d: dict) -> "BuildTargetConfiguration":
@dataclass(slots=True, frozen=True)
class BuildTargetInfo:
- """Class to hold important versions within the build target.
+ """Various versions and other information about a build target.
- This is very similar to the NodeInfo class, it just instead holds information
- for the build target.
+ Attributes:
+ dpdk_version: The DPDK version that was built.
+ compiler_version: The version of the compiler used to build DPDK.
"""
dpdk_version: str
compiler_version: str
-class TestSuiteConfigDict(TypedDict):
- suite: str
- cases: list[str]
-
-
@dataclass(slots=True, frozen=True)
class TestSuiteConfig:
+ """Test suite configuration.
+
+ Information about a single test suite to be executed.
+
+ Attributes:
+ test_suite: The name of the test suite module without the starting
``TestSuite_``.
+ test_cases: The names of test cases from this test suite to execute.
+ If empty, all test cases will be executed.
+ """
+
test_suite: str
test_cases: list[str]
@@ -228,6 +416,14 @@ class TestSuiteConfig:
def from_dict(
entry: str | TestSuiteConfigDict,
) -> "TestSuiteConfig":
+ """Create an instance from two different types.
+
+ Args:
+ entry: Either a suite name or a dictionary containing the config.
+
+ Returns:
+ The test suite configuration instance.
+ """
if isinstance(entry, str):
return TestSuiteConfig(test_suite=entry, test_cases=[])
elif isinstance(entry, dict):
@@ -238,19 +434,49 @@ def from_dict(
@dataclass(slots=True, frozen=True)
class ExecutionConfiguration:
+ """The configuration of an execution.
+
+ The configuration contains testbed information, what tests to execute
+ and with what DPDK build.
+
+ Attributes:
+ build_targets: A list of DPDK builds to test.
+ perf: Whether to run performance tests.
+ func: Whether to run functional tests.
+ skip_smoke_tests: Whether to skip smoke tests.
+ test_suites: The names of test suites and/or test cases to execute.
+ system_under_test_node: The SUT node to use in this execution.
+ traffic_generator_node: The TG node to use in this execution.
+ vdevs: The names of virtual devices to test.
+ """
+
build_targets: list[BuildTargetConfiguration]
perf: bool
func: bool
+ skip_smoke_tests: bool
test_suites: list[TestSuiteConfig]
system_under_test_node: SutNodeConfiguration
traffic_generator_node: TGNodeConfiguration
vdevs: list[str]
- skip_smoke_tests: bool
@staticmethod
def from_dict(
- d: dict, node_map: dict[str, Union[SutNodeConfiguration |
TGNodeConfiguration]]
+ d: ExecutionConfigDict,
+ node_map: dict[str, Union[SutNodeConfiguration | TGNodeConfiguration]],
) -> "ExecutionConfiguration":
+ """A convenience method that processes the inputs before creating an
instance.
+
+ The build target and the test suite config is transformed into their
respective objects.
