[PATCH v2 3/5] dts: add offload configuration querying to testpmd

[jspewock@iol.unh.edu]

From: Jeremy Spewock <jspewock@iol.unh.edu>

Testpmd offers methods for querying the runtime configuration of
offloads on a device that are useful for verification, but bindings to
reach these methods do not exist in the Testpmd API offered in the
framework. This patch creates methods that can query this configuration
and also generalizes the OffloadCapability class to allow it to account
for parsing the configuration output as well since the flag values will
be the same.

Signed-off-by: Jeremy Spewock <jspewock@iol.unh.edu>
---
 dts/framework/remote_session/testpmd_shell.py | 92 ++++++++++++++++++-
 1 file changed, 89 insertions(+), 3 deletions(-)

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index df4ed7ce5c..71859c63da 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -680,21 +680,45 @@ def from_string(cls, line: str) -> Self:
         return flag
 
     @classmethod
-    def make_parser(cls, per_port: bool) -> ParserFn:
+    def from_list(cls, lines: list[str]) -> list[Self]:
+        """Make a list of instances from a list of strings that contain flag names.
+
+        The strings are expected to separate the flag names by whitespace.
+
+        Args:
+            lines: The list of strings to parse.
+
+        Returns:
+            A list of instances parsed from each string in `lines`.
+        """
+        return [cls.from_string(line) for line in lines]
+
+    @classmethod
+    def make_parser(cls, per_port: bool, find_multiple: bool = False) -> ParserFn:
         """Make a parser function.
 
         Args:
             per_port: If :data:`True`, will return capabilities per port. If :data:`False`,
                 will return capabilities per queue.
+            find_multiple: If :data:`True`, will use :func:`TextParser.find_all` to find all
+                matches for the regex query and return a list of instances based on those matches.
+                If :data:`False`, will return a single instance of the flag based off a single
+                match.
 
         Returns:
             ParserFn: A dictionary for the `dataclasses.field` metadata argument containing a
                 parser function that makes an instance of this flag from text.
         """
         granularity = "Port" if per_port else "Queue"
+        parser_func: Callable[..., ParserFn] | Callable[..., ParserFn] = (
+            TextParser.find_all if find_multiple else TextParser.find
+        )
+        instance_func: Callable[..., list[OffloadCapability]] | Callable[..., OffloadCapability] = (
+            cls.from_list if find_multiple else cls.from_string
+        )
         return TextParser.wrap(
-            TextParser.find(rf"Per {granularity}\s+:(.*)$", re.MULTILINE),
-            cls.from_string,
+            parser_func(rf"{granularity}[\s\[\]\d]+:(.*)$", re.MULTILINE),
+            instance_func,
         )
 
 
@@ -822,6 +846,42 @@ class TxOffloadCapabilities(OffloadCapabilities):
     per_port: TxOffloadCapability = field(metadata=TxOffloadCapability.make_parser(True))
 
 
+@dataclass
+class OffloadConfiguration(TextParser):
+    """The result of testpmd's ``show port <port_id> rx/tx_offload configuration`` command."""
+
+    #:
+    port_id: int = field(metadata=TextParser.find_int(r"Offloading Configuration of port (\d+) :"))
+    #: Queue offload configurations.
+    queues: list[RxOffloadCapability] | list[TxOffloadCapability]
+    #: Port offload configuration.
+    port: RxOffloadCapability | TxOffloadCapability
+
+
+@dataclass
+class RxOffloadConfiguration(OffloadConfiguration):
+    """Extends :class:`OffloadingConfiguration` with Rx specific functionality."""
+
+    #:
+    queues: list[RxOffloadCapability] = field(
+        metadata=RxOffloadCapability.make_parser(False, find_multiple=True)
+    )
+    #:
+    port: RxOffloadCapability = field(metadata=RxOffloadCapability.make_parser(True))
+
+
+@dataclass
+class TxOffloadConfiguration(OffloadConfiguration):
+    """Extends :class:`OffloadingConfiguration` with Tx specific functionality."""
+
+    #:
+    queues: list[TxOffloadCapability] = field(
+        metadata=TxOffloadCapability.make_parser(False, find_multiple=True)
+    )
+    #:
+    port: TxOffloadCapability = field(metadata=TxOffloadCapability.make_parser(True))
+
+
 T = TypeVarTuple("T")  # type: ignore[misc]
 
 
@@ -1590,6 +1650,32 @@ def show_port_tx_offload_capabilities(self, port_id: int) -> TxOffloadCapabiliti
         offload_capabilities_out = self.send_command(command)
         return TxOffloadCapabilities.parse(offload_capabilities_out)
 
+    def show_port_rx_offload_configuration(self, port_id: int) -> RxOffloadConfiguration:
+        """Get the Rx offload configuration on a given port.
+
+        Args:
+            port_id: The ID of the port to query the configuration of.
+
+        Returns:
+            An instance of :class:`RxOffloadConfiguration` containing the offload configuration of
+            the port.
+        """
+        output = self.send_command(f"show port {port_id} rx_offload configuration")
+        return RxOffloadConfiguration.parse(output)
+
+    def show_port_tx_offload_configuration(self, port_id: int) -> TxOffloadConfiguration:
+        """Get the Tx offload configuration on a given port.
+
+        Args:
+            port_id: The ID of the port to query the configuration of.
+
+        Returns:
+            An instance of :class:`TxOffloadConfiguration` containing the offload configuration of
+            the port.
+        """
+        output = self.send_command(f"show port {port_id} tx_offload configuration")
+        return TxOffloadConfiguration.parse(output)
+
     def _stop_port(self, port_id: int, verify: bool = True) -> None:
         """Stop port with `port_id` in testpmd.
 
-- 
2.46.0