Dpdk: annotation fixes for mypy

Author: mcgov

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: