@@ -10,13 +10,18 @@ executions:
compiler_wrapper: ccache
perf: false
func: true
+ nic: #physical devices to be used for testing
+ address: "0000:00:10.1"
+ driver: "vfio-pci"
test_suites:
+ - smoke_tests
- hello_world
system_under_test: "SUT 1"
nodes:
- name: "SUT 1"
- hostname: sut1.change.me.localhost
+ hostname: host.example.name
user: root
+ password: ""
arch: x86_64
os: linux
lcores: ""
@@ -106,6 +106,18 @@ def from_dict(d: dict) -> "NodeConfiguration":
hugepages=hugepage_config,
)
+@dataclass(slots=True, frozen=True)
+class NICConfiguration:
+ address: str
+ driver: str
+
+ @staticmethod
+ def from_dict(d:dict) -> "NICConfiguration":
+ return NICConfiguration(
+ address=d.get("address"),
+ driver=d.get("driver")
+ )
+
@dataclass(slots=True, frozen=True)
class BuildTargetConfiguration:
@@ -157,6 +169,7 @@ class ExecutionConfiguration:
func: bool
test_suites: list[TestSuiteConfig]
system_under_test: NodeConfiguration
+ nic: NICConfiguration
@staticmethod
def from_dict(d: dict, node_map: dict) -> "ExecutionConfiguration":
@@ -166,6 +179,7 @@ def from_dict(d: dict, node_map: dict) -> "ExecutionConfiguration":
test_suites: list[TestSuiteConfig] = list(
map(TestSuiteConfig.from_dict, d["test_suites"])
)
+ nic_conf: NICConfiguration = NICConfiguration.from_dict(d['nic'])
sut_name = d["system_under_test"]
assert sut_name in node_map, f"Unknown SUT {sut_name} in execution {d}"
@@ -174,6 +188,7 @@ def from_dict(d: dict, node_map: dict) -> "ExecutionConfiguration":
perf=d["perf"],
func=d["func"],
test_suites=test_suites,
+ nic=nic_conf,
system_under_test=node_map[sut_name],
)
@@ -97,7 +97,8 @@
"test_suite": {
"type": "string",
"enum": [
- "hello_world"
+ "hello_world",
+ "smoke_tests"
]
},
"test_target": {
@@ -211,6 +212,19 @@
]
}
},
+ "nic": {
+ "type": "object",
+ "properties": {
+ "address": {
+ "type":"string",
+ "description": "PCI address of a physical device to test"
+ },
+ "driver": {
+ "type": "string",
+ "description": "The name of the driver that the physical device should be bound to"
+ }
+ }
+ },
"system_under_test": {
"$ref": "#/definitions/node_name"
}
@@ -5,6 +5,8 @@
import sys
+from .exception import BlockingTestSuiteError
+
from .config import CONFIGURATION, BuildTargetConfiguration, ExecutionConfiguration
from .logger import DTSLOG, getLogger
from .test_result import BuildTargetResult, DTSResult, ExecutionResult, Result
@@ -49,6 +51,7 @@ def run_all() -> None:
nodes[sut_node.name] = sut_node
if sut_node:
+ #SMOKE TEST EXECUTION GOES HERE!
_run_execution(sut_node, execution, result)
except Exception as e:
@@ -118,7 +121,7 @@ def _run_build_target(
try:
sut_node.set_up_build_target(build_target)
- result.dpdk_version = sut_node.dpdk_version
+ # result.dpdk_version = sut_node.dpdk_version
build_target_result.update_setup(Result.PASS)
except Exception as e:
dts_logger.exception("Build target setup failed.")
@@ -146,6 +149,7 @@ def _run_suites(
with possibly only a subset of test cases.
If no subset is specified, run all test cases.
"""
+ end_execution = False
for test_suite_config in execution.test_suites:
try:
full_suite_path = f"tests.TestSuite_{test_suite_config.test_suite}"
@@ -160,13 +164,24 @@ def _run_suites(
else:
for test_suite_class in test_suite_classes:
+ #HERE NEEDS CHANGING
test_suite = test_suite_class(
sut_node,
test_suite_config.test_cases,
execution.func,
build_target_result,
+ sut_node._build_target_config,
+ result
)
- test_suite.run()
+ try:
+ test_suite.run()
+ except BlockingTestSuiteError as e:
+ dts_logger.exception("An error occurred within a blocking TestSuite, execution will now end.")
+ result.add_error(e)
+ end_execution = True
+ #if a blocking test failed and we need to bail out of suite executions
+ if end_execution:
+ break
def _exit_dts() -> None:
@@ -25,6 +25,7 @@ class ErrorSeverity(IntEnum):
SSH_ERR = 4
DPDK_BUILD_ERR = 10
TESTCASE_VERIFY_ERR = 20
+ BLOCKING_TESTSUITE_ERR = 25
class DTSError(Exception):
@@ -144,3 +145,13 @@ def __init__(self, value: str):
def __str__(self) -> str:
return repr(self.value)
+
+class BlockingTestSuiteError(DTSError):
+ suite_name: str
+ severity: ClassVar[ErrorSeverity] = ErrorSeverity.BLOCKING_TESTSUITE_ERR
+
+ def __init__(self, suite_name:str) -> None:
+ self.suite_name = suite_name
+
+ def __str__(self) -> str:
+ return f"Blocking suite {self.suite_name} failed."
@@ -8,6 +8,7 @@
import os.path
from collections.abc import MutableSequence
from enum import Enum, auto
+from typing import Dict
from .config import (
OS,
@@ -67,12 +68,13 @@ class Statistics(dict):
Using a dict provides a convenient way to format the data.
"""
- def __init__(self, dpdk_version):
+ def __init__(self, output_info: Dict[str, str] | None):
super(Statistics, self).__init__()
for result in Result:
self[result.name] = 0
self["PASS RATE"] = 0.0
- self["DPDK VERSION"] = dpdk_version
+ if output_info:
+ for info_key, info_val in output_info.items(): self[info_key] = info_val
def __iadd__(self, other: Result) -> "Statistics":
"""
@@ -258,6 +260,7 @@ class DTSResult(BaseResult):
"""
dpdk_version: str | None
+ output: dict | None
_logger: DTSLOG
_errors: list[Exception]
_return_code: ErrorSeverity
@@ -267,6 +270,7 @@ class DTSResult(BaseResult):
def __init__(self, logger: DTSLOG):
super(DTSResult, self).__init__()
self.dpdk_version = None
+ self.output = None
self._logger = logger
self._errors = []
self._return_code = ErrorSeverity.NO_ERR
@@ -296,7 +300,10 @@ def process(self) -> None:
for error in self._errors:
self._logger.debug(repr(error))
- self._stats_result = Statistics(self.dpdk_version)
+ self._stats_result = Statistics(self.output)
+ #add information gathered from the smoke tests to the statistics
+ # for info_key, info_val in smoke_test_info.items(): self._stats_result[info_key] = info_val
+ # print(self._stats_result)
self.add_stats(self._stats_result)
with open(self._stats_filename, "w+") as stats_file:
stats_file.write(str(self._stats_result))
@@ -10,11 +10,14 @@
import inspect
import re
from types import MethodType
+from typing import Dict
-from .exception import ConfigurationError, SSHTimeoutError, TestCaseVerifyError
+from .config import BuildTargetConfiguration
+
+from .exception import BlockingTestSuiteError, ConfigurationError, SSHTimeoutError, TestCaseVerifyError
from .logger import DTSLOG, getLogger
from .settings import SETTINGS
-from .test_result import BuildTargetResult, Result, TestCaseResult, TestSuiteResult
+from .test_result import BuildTargetResult, DTSResult, Result, TestCaseResult, TestSuiteResult
from .testbed_model import SutNode
@@ -37,10 +40,12 @@ class TestSuite(object):
"""
sut_node: SutNode
+ is_blocking = False
_logger: DTSLOG
_test_cases_to_run: list[str]
_func: bool
_result: TestSuiteResult
+ _dts_result: DTSResult
def __init__(
self,
@@ -48,6 +53,8 @@ def __init__(
test_cases: list[str],
func: bool,
build_target_result: BuildTargetResult,
+ build_target_conf: BuildTargetConfiguration,
+ dts_result: DTSResult
):
self.sut_node = sut_node
self._logger = getLogger(self.__class__.__name__)
@@ -55,6 +62,8 @@ def __init__(
self._test_cases_to_run.extend(SETTINGS.test_cases)
self._func = func
self._result = build_target_result.add_test_suite(self.__class__.__name__)
+ self.build_target_info = build_target_conf
+ self._dts_result = dts_result
def set_up_suite(self) -> None:
"""
@@ -118,6 +127,9 @@ def run(self) -> None:
f"the next test suite may be affected."
)
self._result.update_setup(Result.ERROR, e)
+ if len(self._result.get_errors()) > 0 and self.is_blocking:
+ raise BlockingTestSuiteError(test_suite_name)
+
def _execute_test_suite(self) -> None:
"""
@@ -137,6 +149,7 @@ def _execute_test_suite(self) -> None:
f"Attempt number {attempt_nr} out of {all_attempts}."
)
self._run_test_case(test_case_method, test_case_result)
+
def _get_functional_test_cases(self) -> list[MethodType]:
"""
@@ -232,6 +245,11 @@ def _execute_test_case(
test_case_result.update(Result.SKIP)
raise KeyboardInterrupt("Stop DTS")
+ def write_to_statistics_file(self, output: Dict[str, str]):
+ if self._dts_result.output != None:
+ self._dts_result.output.update(output)
+ else:
+ self._dts_result.output = output
def get_test_suites(testsuite_module_path: str) -> list[type[TestSuite]]:
def is_test_suite(object) -> bool:
@@ -252,3 +270,5 @@ def is_test_suite(object) -> bool:
test_suite_class
for _, test_suite_class in inspect.getmembers(testcase_module, is_test_suite)
]
+
+
new file mode 100644
@@ -0,0 +1,63 @@
+from framework.test_suite import TestSuite
+from framework.testbed_model.sut_node import SutNode
+
+
+def get_compiler_version(compiler_name: str, sut_node: SutNode) -> str:
+ match compiler_name:
+ case "gcc":
+ return sut_node.main_session.send_command(f"{compiler_name} --version", 60).stdout.split("\n")[0]
+ case "clang":
+ return sut_node.main_session.send_command(f"{compiler_name} --version", 60).stdout.split("\n")[0]
+ case "msvc":
+ return sut_node.main_session.send_command(f"cl", 60).stdout
+ case "icc":
+ return sut_node.main_session.send_command(f"{compiler_name} -V", 60).stdout
+
+class SmokeTests(TestSuite):
+ is_blocking = True
+
+
+ def set_up_suite(self) -> None:
+ """
+ Setup:
+ build all DPDK
+ """
+ self.dpdk_build_dir_path = self.sut_node.remote_dpdk_build_dir
+
+
+ def test_unit_tests(self) -> None:
+ """
+ Test:
+ run the fast-test unit-test suite through meson
+ """
+ self.sut_node.main_session.send_command(f"meson test -C {self.dpdk_build_dir_path} --suite fast-tests", 300, True)
+
+ def test_driver_tests(self) -> None:
+ """
+ Test:
+ run the driver-test unit-test suite through meson
+ """
+ self.sut_node.main_session.send_command(f"meson test -C {self.dpdk_build_dir_path} --suite driver-tests", 300)
+
+ def test_gather_info(self) -> None:
+ out = {}
+
+ out['OS'] = self.sut_node.main_session.send_command("awk -F= '$1==\"NAME\" {print $2}' /etc/os-release", 60).stdout
+ out["OS VERSION"] = self.sut_node.main_session.send_command("awk -F= '$1==\"VERSION\" {print $2}' /etc/os-release", 60, True).stdout
+ out["COMPILER VERSION"] = get_compiler_version(self.build_target_info.compiler.name, self.sut_node)
+ out["DPDK VERSION"] = self.sut_node.dpdk_version
+ if self.build_target_info.os.name == "linux":
+ out['KERNEL VERSION'] = self.sut_node.main_session.send_command("uname -r", 60).stdout
+ elif self.build_target_info.os.name == "windows":
+ out['KERNEL VERSION'] = self.sut_node.main_session.send_command("uname -a", 60).stdout
+ self.write_to_statistics_file(out)
+
+ def test_start_testpmd(self) -> None:
+ """
+ Still heavily in development
+ """
+ self.sut_node.main_session.send_command(f"{self.dpdk_build_dir_path}/app/dpdk-testpmd -- -i", 60)
+ out = self.sut_node.main_session.send_command("show port summary all")
+ self.sut_node.main_session.send_command("quit")
+ print(out.stdout)
+
\ No newline at end of file