[PATCH v4] dts: report dut/NIC info during DTS run

[Patrick Robb]

On Wed, Jun 24, 2026 at 5:33 PM Koushik Bhargav Nimoji <knimoji at iol.unh.edu>
wrote:

> This patch gathers NIC info during a DTS run and writes it to an output
> json file. This allows the json file to be used when reporting results
> on the DTS results dashboard.
>
> Signed-off-by: Koushik Bhargav Nimoji <knimoji at iol.unh.edu>
> ---
> v2:
>     *Resolved merge conflicts
> v3:
>     *Fixed an issue with retrieving
>      the NIC's hardware version
> v4:
>     *Moved nic info gathering step before the nics get
>      binded to their respective drivers
>     *Condensed some areas of code in order to make them
>      more readable
>     *Removed redundant None checks and added some where
>      required
>     *Fixed LshwOutput class to better reflect the lshw
>      command output
> ---
>  dts/framework/test_run.py                    |  8 +++
>  dts/framework/testbed_model/linux_session.py | 68 ++++++++++++++++++++
>  dts/framework/testbed_model/os_session.py    | 11 ++++
>  3 files changed, 87 insertions(+)
>
> diff --git a/dts/framework/test_run.py b/dts/framework/test_run.py
> index 94dc6023a7..c92fe90f2e 100644
> --- a/dts/framework/test_run.py
> +++ b/dts/framework/test_run.py
> @@ -98,6 +98,7 @@
>          "InternalError" -> "exit":ew
>  """
>
> +import json
>  import random
>  from collections import deque
>  from collections.abc import Iterable
> @@ -347,6 +348,12 @@ def next(self) -> State | None:
>          test_run.ctx.dpdk.setup()
>          test_run.ctx.topology.setup()
>
> +        used_nic_info: list[dict[str, str]] =
> self.test_run.ctx.sut_node.main_session.get_nic_info()
>

drop "used" for nic_info or change to testrun_nic_info?


> +        with open(f"{SETTINGS.output_dir}/dut_info.json", "w") as file:
> +            json.dump(used_nic_info, file, indent=3)
> +
> +        self.logger.info(f"DUT NIC info written to:
> {SETTINGS.output_dir}/dut_info.json")
> +
>          if test_run.config.use_virtual_functions:
>              test_run.ctx.topology.instantiate_vf_ports()
>          if test_run.ctx.sut_node.cryptodevs and test_run.config.crypto:
> @@ -370,6 +377,7 @@ def next(self) -> State | None:
>          test_run.supported_capabilities = get_supported_capabilities(
>              test_run.ctx.sut_node, test_run.ctx.topology,
> test_run.required_capabilities
>          )
> +
>          return TestRunExecution(test_run, self.result)
>
>      def on_error(self, ex: BaseException) -> State | None:
> diff --git a/dts/framework/testbed_model/linux_session.py
> b/dts/framework/testbed_model/linux_session.py
> index 3a6e97974b..9e9146c372 100644
> --- a/dts/framework/testbed_model/linux_session.py
> +++ b/dts/framework/testbed_model/linux_session.py
> @@ -38,6 +38,8 @@ class LshwConfigurationOutput(TypedDict):
>      driver: str
>      #:
>      link: str
> +    #:
> +    firmware: str
>
>
>  class LshwOutput(TypedDict):
> @@ -61,6 +63,12 @@ class LshwOutput(TypedDict):
>              ...
>      """
>
> +    #:
> +    vendor: NotRequired[str]
> +    #:
> +    product: NotRequired[str]
> +    #:
> +    version: NotRequired[str]
>      #:
>      businfo: str
>      #:
> @@ -197,6 +205,66 @@ def unbind_ports(self, ports: list[Port]):
>          if self._lshw_net_info:
>              del self._lshw_net_info
>
> +    def get_nic_info(self) -> list[dict[str, str]]:
> +        """Overrides :meth`~.os_session.OSSession.get_nic_info`.
> +
> +        Raises:
> +            ConfigurationError: If the NIC info could not be found.
> +        """
> +        port_data = {
> +            port.get("businfo"): port for port in self._lshw_net_info if
> port.get("businfo")
> +        }
> +
> +        all_nic_info: list[dict[str, str]] = []
> +        for port in self._config.ports:
> +            pci_addr = port.pci
> +
> +            command_result = self.send_command(
>

rename to lshw_result please.


> +                f"sudo lshw -c network -businfo | grep '{pci_addr}' | cut
> -d'@' -f1"
> +            )
> +            if command_result.return_code != 0 and command_result.stdout
> == "":
> +                raise ConfigurationError(f"Unable to get bus type for
> port {pci_addr}.")
> +            bus_type = command_result.stdout
> +
> +            bus_info = f"{bus_type}@{pci_addr}"
> +            nic_port: LshwOutput | None = port_data[bus_info]
> +            if nic_port is None:
> +                raise ConfigurationError(f"Port {pci_addr} could not be
> found on the node.")
> +
> +            config: LshwConfigurationOutput | None =
> nic_port["configuration"]
> +            if config is None:
> +                raise ConfigurationError(
> +                    f"Configuration info for port {pci_addr} could not be
> found on the node."
> +                )
> +
> +            if "logicalname" not in nic_port:
> +                raise ConfigurationError(
> +                    f"Logical name for port {pci_addr} could not be found
> on the node."
> +                )
> +
> +            command_result = self.send_command(
>

ethtool_result


> +                f"ethtool {nic_port['logicalname']} | grep 'Speed:' | awk
> '{{print $2}}'"
> +            )

+            if command_result.return_code == 0 and command_result.stdout:
> +                nic_speed = command_result.stdout
> +            else:
> +                self._logger.error(f"Unable to get speed for NIC:
> {pci_addr}")
> +                nic_speed = None
> +
> +            dut_json = {
> +                "make": nic_port["vendor"] if "vendor" in nic_port else
> "Unknown",
> +                "model": nic_port["product"] if "product" in nic_port
> else "Unknown",
> +                "hardware version": nic_port["version"] if "version" in
> nic_port else "Unknown",
> +                "firmware version": config["firmware"] if "firmware" in
> config else "Unknown",
> +                "deviceBusType": bus_type,
> +                "deviceId": nic_port["serial"] if "serial" in nic_port
> else "Unknown",
> +                "pmd": config["driver"] if "driver" in config else
> "Unknown",
> +                "speed": nic_speed or "Unknown",
> +            }
> +            all_nic_info.append(dut_json)
> +
> +        return all_nic_info
> +
>

What is the intended behavior for cryptodev tests? I realize the ports list
will be empty and we will not enter the initial loop, but is this intended?
Do we want to gether cryptodev info too?


>      def bind_ports_to_driver(self, ports: list[Port], driver_name: str)
> -> None:
>
>
Reviewed-by: Patrick Robb <patrickrobb1997 at gmail.com>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://mails.dpdk.org/archives/dev/attachments/20260624/116d68e7/attachment-0001.htm>