new file mode 100644
@@ -0,0 +1,6 @@
+executions:
+ - system_under_test: "SUT 1"
+nodes:
+ - name: "SUT 1"
+ hostname: sut1.change.me.localhost
+ user: root
new file mode 100644
@@ -0,0 +1,99 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2010-2021 Intel Corporation
+# Copyright(c) 2022 University of New Hampshire
+#
+
+"""
+Generic port and topology nodes configuration file load function
+"""
+import json
+import os.path
+import pathlib
+from dataclasses import dataclass
+from typing import Any, Optional
+
+import warlock
+import yaml
+
+from framework.settings import SETTINGS
+
+
+# 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 NodeConfiguration:
+ name: str
+ hostname: str
+ user: str
+ password: Optional[str]
+
+ @staticmethod
+ def from_dict(d: dict) -> "NodeConfiguration":
+ return NodeConfiguration(
+ name=d["name"],
+ hostname=d["hostname"],
+ user=d["user"],
+ password=d.get("password"),
+ )
+
+
+@dataclass(slots=True, frozen=True)
+class ExecutionConfiguration:
+ system_under_test: NodeConfiguration
+
+ @staticmethod
+ def from_dict(d: dict, node_map: dict) -> "ExecutionConfiguration":
+ sut_name = d["system_under_test"]
+ assert sut_name in node_map, f"Unknown SUT {sut_name} in execution {d}"
+
+ return ExecutionConfiguration(
+ system_under_test=node_map[sut_name],
+ )
+
+
+@dataclass(slots=True, frozen=True)
+class Configuration:
+ executions: list[ExecutionConfiguration]
+
+ @staticmethod
+ def from_dict(d: dict) -> "Configuration":
+ nodes: list[NodeConfiguration] = list(
+ map(NodeConfiguration.from_dict, d["nodes"])
+ )
+ assert len(nodes) > 0, "There must be a node to test"
+
+ node_map = {node.name: node for node in nodes}
+ assert len(nodes) == len(node_map), "Duplicate node names are not allowed"
+
+ executions: list[ExecutionConfiguration] = list(
+ map(
+ ExecutionConfiguration.from_dict, d["executions"], [node_map for _ in d]
+ )
+ )
+
+ return Configuration(executions=executions)
+
+
+def load_config() -> Configuration:
+ """
+ Loads the configuration file and the configuration file schema,
+ validates the configuration file, and creates a configuration object.
+ """
+ with open(SETTINGS.config_file_path, "r") as f:
+ config_data = yaml.safe_load(f)
+
+ schema_path = os.path.join(
+ pathlib.Path(__file__).parent.resolve(), "conf_yaml_schema.json"
+ )
+
+ with open(schema_path, "r") as f:
+ schema = json.load(f)
+ config: dict[str, Any] = warlock.model_factory(schema, name="_Config")(config_data)
+ config_obj: Configuration = Configuration.from_dict(dict(config))
+ return config_obj
+
+
+CONFIGURATION = load_config()
new file mode 100644
@@ -0,0 +1,73 @@
+{
+ "$schema": "https://json-schema.org/draft-07/schema",
+ "title": "DPDK DTS Config Schema",
+ "definitions": {
+ "node_name": {
+ "type": "string",
+ "description": "A unique identifier for a node"
+ },
+ "node_role": {
+ "type": "string",
+ "description": "The role a node plays in DTS",
+ "enum": [
+ "system_under_test",
+ "traffic_generator"
+ ]
+ }
+ },
+ "type": "object",
+ "properties": {
+ "nodes": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "name": {
+ "type": "string",
+ "description": "A unique identifier for this node"
+ },
+ "hostname": {
+ "type": "string",
+ "description": "A hostname from which the node running DTS can access this node. This can also be an IP address."
+ },
+ "user": {
+ "type": "string",
+ "description": "The user to access this node with."
+ },
+ "password": {
+ "type": "string",
+ "description": "The password to use on this node. Use only as a last resort. SSH keys are STRONGLY preferred."
+ }
+ },
+ "additionalProperties": false,
+ "required": [
+ "name",
+ "hostname",
+ "user"
+ ]
+ },
+ "minimum": 1
+ },
+ "executions": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "system_under_test": {
+ "$ref": "#/definitions/node_name"
+ }
+ },
+ "additionalProperties": false,
+ "required": [
+ "system_under_test"
+ ]
+ },
+ "minimum": 1
+ }
+ },
+ "required": [
+ "executions",
+ "nodes"
+ ],
+ "additionalProperties": false
+}
new file mode 100644
@@ -0,0 +1,23 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2010-2014 Intel Corporation
+# Copyright(c) 2022 PANTHEON.tech s.r.o.
+# Copyright(c) 2022 University of New Hampshire
+#
+
+"""
+User-defined exceptions used across the framework.
+"""
+
+
+class ConfigParseException(Exception):
+ """
+ Configuration file parse failure exception.
+ """
+
+ config: str
+
+ def __init__(self, conf_file: str):
+ self.config = conf_file
+
+ def __str__(self) -> str:
+ return f"Failed to parse config file [{self.config}]"
new file mode 100644
@@ -0,0 +1,84 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2010-2021 Intel Corporation
+# Copyright(c) 2022 PANTHEON.tech s.r.o.
+# Copyright(c) 2022 University of New Hampshire
+#
+
+import argparse
+import os
+from dataclasses import dataclass
+from typing import Any, Callable, Iterable, Optional, Sequence, TypeVar
+
+_T = TypeVar("_T")
+
+
+def _env_arg(env_var: str) -> Any:
+ class _EnvironmentArgument(argparse.Action):
+ def __init__(
+ self,
+ option_strings: Sequence[str],
+ dest: str,
+ nargs: Optional[str | int] = None,
+ const: Optional[str] = None,
+ default: str = None,
+ type: Callable[[str], Optional[_T | argparse.FileType]] = None,
+ choices: Optional[Iterable[_T]] = None,
+ required: bool = True,
+ help: Optional[str] = None,
+ metavar: Optional[str | tuple[str, ...]] = None,
+ ) -> None:
+ env_var_value = os.environ.get(env_var)
+ default = env_var_value or default
+ super(_EnvironmentArgument, self).__init__(
+ option_strings,
+ dest,
+ nargs=nargs,
+ const=const,
+ default=default,
+ type=type,
+ choices=choices,
+ required=required,
+ help=help,
+ metavar=metavar,
+ )
+
+ def __call__(
+ self,
+ parser: argparse.ArgumentParser,
+ namespace: argparse.Namespace,
+ values: Any,
+ option_string: str = None,
+ ) -> None:
+ setattr(namespace, self.dest, values)
+
+ return _EnvironmentArgument
+
+
+@dataclass(slots=True, frozen=True)
+class _Settings:
+ config_file_path: str
+
+
+def _get_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(description="DPDK test framework.")
+
+ parser.add_argument(
+ "--config-file",
+ action=_env_arg("DTS_CFG_FILE"),
+ default="conf.yaml",
+ required=False,
+ help="[DTS_CFG_FILE] configuration file that describes the test cases, SUTs "
+ "and targets.",
+ )
+
+ return parser
+
+
+def _get_settings() -> _Settings:
+ parsed_args = _get_parser().parse_args()
+ return _Settings(
+ config_file_path=parsed_args.config_file,
+ )
+
+
+SETTINGS: _Settings = _get_settings()