@@ -32,7 +32,7 @@
from typing_extensions import TypeVarTuple
-from framework.exception import InteractiveCommandExecutionError
+from framework.exception import InteractiveCommandExecutionError, InternalError
from framework.params.testpmd import SimpleForwardingModes, TestPmdParams
from framework.params.types import TestPmdParamsDict
from framework.parser import ParserFn, TextParser
@@ -2002,6 +2002,146 @@ def set_verbose(self, level: int, verify: bool = True) -> None:
f"Testpmd failed to set verbose level to {level}."
)
+ @stop_then_start_port()
+ def _set_offload(
+ self,
+ port_id: int,
+ is_rx: bool,
+ offloads: OffloadCapability,
+ on: bool,
+ queue_id: int | None = None,
+ verify: bool = True,
+ ) -> None:
+ """Base method for configuring offloads on ports and queues.
+
+ If `queue_id` is not specified then it is assumed that you want to set the offloads on the
+ port rather than a queue.
+ """
+ for offload in type(offloads):
+ if offload not in offloads or offload.name is None:
+ continue
+ port_type = "rx" if is_rx else "tx"
+ command = [
+ "port",
+ f"{port_id}",
+ f"{port_type}_offload",
+ f"{offload.name.lower()}",
+ f"{'on' if on else 'off'}",
+ ]
+ if queue_id is not None:
+ # If modifying queues the command is "port <id> rxq <q_id> ..."
+ command.insert(2, f"{port_type}q {queue_id}")
+ else:
+ # If modifying a port the command is "port config <id> ..."
+ command.insert(1, "config")
+
+ self.send_command(" ".join(command))
+ if verify:
+ # verification of ports has to be done based on if it was applied to all queues or
+ # not because the "Per Port" line doesn't get modified until the port is started.
+ current_offload_conf: OffloadConfiguration = (
+ self.show_port_rx_offload_configuration(port_id)
+ if is_rx
+ else self.show_port_tx_offload_configuration(port_id)
+ )
+ # Casting to the generic type is required for mypy
+ queues_capabilities = cast(list[OffloadCapability], current_offload_conf.queues)
+ if queue_id is not None and len(current_offload_conf.queues) < queue_id + 1:
+ raise InternalError(f"Queue {queue_id} does not exist in testpmd")
+ capability_is_set = (
+ len(current_offload_conf.queues) > 0
+ and (queue_id is not None and offload in queues_capabilities[queue_id])
+ or all(offload in conf for conf in queues_capabilities)
+ )
+ if capability_is_set != on:
+ self._logger.debug(
+ f"Test pmd failed to modify capabilities on port {port_id}:\n"
+ f"{current_offload_conf.queues}"
+ )
+ raise InteractiveCommandExecutionError(
+ f"Test pmd failed to {'add' if on else 'remove'} capability {offload.name} "
+ f"{'to' if on else 'from'} port {port_id}."
+ )
+
+ def set_port_offload(
+ self,
+ port_id: int,
+ is_rx: bool,
+ offload: OffloadCapability,
+ on: bool,
+ verify: bool = True,
+ ) -> None:
+ """Configure Rx/Tx offload on a port.
+
+ Args:
+ port_id: The ID of the port to set configure the offload on.
+ is_rx: A flag that signifies which type of offload to set. If :data:`True` an Rx
+ offload will be set, otherwise a Tx offload will be set.
+ offload: The offload to set on the port.
+ on: If :data:`True` the specified offload will be set turned on, otherwise the offload
+ will be turned off.
+ verify: If :data:`True` an additional command will be sent to check the configuration
+ of offloads on the port to verify `offload` was set properly. Defaults to
+ :data:`True`.
+
+ Raises:
+ InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd failed to
+ set the offload on the port.
+ """
+ self._set_offload(port_id, is_rx, offload, on, None, verify)
+
+ def set_queue_offload(
+ self,
+ port_id: int,
+ queue_id: int,
+ is_rx: bool,
+ offload: OffloadCapability,
+ on: bool,
+ verify: bool = True,
+ ) -> None:
+ """Configure Rx/Tx offload on a queue that resides on a specified port.
+
+ Args:
+ port_id: The ID of the port where the queue resides.
+ queue_id: The ID of the queue on the port.
+ is_rx: A flag that signifies which type of offload to set. If :data:`True` an Rx
+ offload will be set, otherwise a Tx offload will be set.
+ offload: The offload to set on a port.
+ on: If :data:`True` the offload will be turned on, otherwise the offload will be turned
+ off.
+ verify: If :data:`True` an additional command will be sent to check the configuration
+ of the queue to validate that the offload was configured properly. Defaults to
+ :data:`True`.
+
+ Raises:
+ InteractiveCommandExecutionError: If `verify` is :data:`True` and testpmd failed to
+ set the offload on the queue.
+ """
+ self._set_offload(port_id, is_rx, offload, on, queue_id, verify)
+
+ def is_port_offload_configured(
+ self, port_id: int, is_rx: bool, offload: OffloadCapability
+ ) -> bool:
+ """Verify whether or not an Rx/Tx offload is currently configured on a port.
+
+ Args:
+ port_id: The ID of the port to check the configuration of.
+ is_rx: If :data:`True` the Rx offloads of the port will be checked, otherwise the Tx
+ offloads will be checked.
+ offload: The offload to search for on the port.
+
+ Returns:
+ :data:`True` if the offload is configured on the port, :data:`False` otherwise.
+ """
+ offload_config: OffloadConfiguration = (
+ self.show_port_rx_offload_configuration(port_id)
+ if is_rx
+ else self.show_port_tx_offload_configuration(port_id)
+ )
+ # Cast to the generic type for mypy
+ port_capabilities = cast(OffloadCapability, offload_config.port)
+ return offload in port_capabilities
+
def flow_create(self, cmd: FlowRule, verify: bool = True) -> None:
"""Creates a flow rule in the testpmd session.