From b87eb65dd9ca8c065c349c6699b28740254fbcfe Mon Sep 17 00:00:00 2001
From: Matthew G McGovern <mamcgove@microsoft.com>
Date: Tue, 19 Nov 2024 10:23:05 -0800
Subject: [PATCH] Dpdk: annotation fixes for mypy
---
microsoft/testsuites/dpdk/dpdkutil.py | 29 +++++++++++++--------------
1 file changed, 14 insertions(+), 15 deletions(-)
diff --git a/microsoft/testsuites/dpdk/dpdkutil.py b/microsoft/testsuites/dpdk/dpdkutil.py
index 5c9f312c93..d57df3a05f 100644
--- a/microsoft/testsuites/dpdk/dpdkutil.py
+++ b/microsoft/testsuites/dpdk/dpdkutil.py
@@ -518,7 +518,8 @@ def verify_dpdk_build(
except (NotEnoughMemoryException, UnsupportedOperationException) as err:
raise SkippedException(err)
testpmd = test_kit.testpmd
-
+ if result is not None:
+ annotate_dpdk_test_result(test_kit, test_result=result, log=log)
# grab a nic and run testpmd
test_nic = node.nics.get_secondary_nic()
@@ -534,8 +535,7 @@ def verify_dpdk_build(
assert_that(tx_pps).described_as(
f"TX-PPS ({tx_pps}) should have been greater than 2^20 (~1m) PPS."
).is_greater_than(2**20)
- if result is not None:
- annotate_dpdk_test_result(test_kit, test_result=result, log=log)
+
return test_kit
@@ -1126,7 +1126,7 @@ class NicType(Enum):
# Short name for nic types
-NIC_SHORT_NAMES = {
+NIC_SHORT_NAMES: Dict[NicType, str] = {
NicType.CX3: "cx3",
NicType.CX4: "cx4",
NicType.CX5: "cx5",
@@ -1139,22 +1139,21 @@ def get_node_nic_short_name(node: Node) -> str:
devices = node.tools[Lspci].get_devices_by_type(DEVICE_TYPE_SRIOV)
if node.nics.is_mana_device_present():
return NIC_SHORT_NAMES[NicType.MANA]
- short_names = map(lambda x: x.value, [NicType.CX3, NicType.CX4, NicType.CX5])
- for nic_name in short_names:
- if any([str(nic_name) in x.device_info for x in devices]):
+ mlx_nics = [NicType.CX3, NicType.CX4, NicType.CX5]
+ for nic_name in mlx_nics:
+ if any([nic_name.value in x.device_info for x in devices]):
return NIC_SHORT_NAMES[nic_name]
# We assert much earlier to enforce that SRIOV is enabled,
# so we should never hit this unless someone is testing a new platform.
# Instead of asserting, just log that the short name was not found.
- known_nic_types = ",".join(short_names)
+ known_nic_types = ",".join([x.value for x in mlx_nics])
found_nic_types = ",".join(map(str, [x.device_id for x in devices]))
- node.log.debug(
+ # assert, this will be caught in annotate_dpdk_test_result
+ # and logged, rather than failing the test.
+ raise AssertionError(
"Unknown NIC hardware was detected during DPDK test case. "
f"Expected one of: {known_nic_types}. Found {found_nic_types}. "
)
- # this is just a function for annotating a result, so don't assert
- # if there's
- return found_nic_types
# Add dpdk/rdma/nic info to dpdk test result
@@ -1170,19 +1169,19 @@ def annotate_dpdk_test_result(
test_result.information["dpdk_version"] = str(dpdk_version)
log.debug(f"Found dpdk version: {dpdk_version}")
except AssertionError as err:
- test_kit.node.log.debug(f"Could not fetch DPDK version info: {str(err)}")
+ log.debug(f"Could not fetch DPDK version info: {str(err)}")
try:
rdma_version = test_kit.rdma_core.get_installed_version()
test_result.information["rdma_version"] = str(rdma_version)
log.debug(f"Found rdma version: {rdma_version}")
except AssertionError as err:
- test_kit.node.log.debug(f"Could not fetch RDMA version info: {str(err)}")
+ log.debug(f"Could not fetch RDMA version info: {str(err)}")
try:
nic_hw = get_node_nic_short_name(test_kit.node)
test_result.information["nic_hw"] = nic_hw
log.debug(f"Found nic version: {nic_hw}")
except AssertionError as err:
- test_kit.node.log.debug(f"Could not fetch NIC short name: {str(err)}")
+ log.debug(f"Could not fetch NIC short name: {str(err)}")
def skip_32bit_test_on_unsupported_distros(os: OperatingSystem) -> None: