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

[Koushik Bhargav Nimoji]

On Wed, Jun 24, 2026 at 10:06 PM Patrick Robb <patrickrobb1997 at gmail.com>
wrote:

>
>
> 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?
>
>
The intended behavior here is to skip cryptodev devices. Not entering the
initial loop, and therefore returning an empty list is the expected
behavior when running cryptodev tests.

>

>      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/20260625/ce6f01e8/attachment-0001.htm>