On Wed, Jun 24, 2026 at 5:33 PM Koushik Bhargav Nimoji <[email protected]> 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 <[email protected]> > --- > 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 <[email protected]>
