[PATCH v4] dts: add flow rule dataclass to testpmd shell

[Dean Marx <dmarx@iol.unh.edu>]

Add dataclass for passing in flow rule creation arguments, as well as a
__str__ method for converting to a sendable testpmd command. Add
flow_create method to TestPmdShell class for initializing flow rules.

Signed-off-by: Dean Marx <dmarx@iol.unh.edu>
---
 dts/framework/remote_session/testpmd_shell.py | 57 +++++++++++++++++++
 1 file changed, 57 insertions(+)

diff --git a/dts/framework/remote_session/testpmd_shell.py b/dts/framework/remote_session/testpmd_shell.py
index b9e7875615..aaae7e5515 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -672,6 +672,44 @@ class TestPmdPortStats(TextParser):
     tx_bps: int = field(metadata=TextParser.find_int(r"Tx-bps:\s+(\d+)"))
 
 
+@dataclass
+class FlowRule:
+    """Dataclass for setting flow rule parameters."""
+
+    #:
+    port_id: int
+    #:
+    ingress: bool
+    #:
+    pattern: str
+    #:
+    actions: str
+
+    #:
+    group_id: int | None = None
+    #:
+    priority_level: int | None = None
+    #:
+    user_id: int | None = None
+
+    def __str__(self) -> str:
+        """Returns the string representation of a flow_func instance.
+
+        In this case, a properly formatted flow create command that can be sent to testpmd.
+        """
+        ret = f"flow create {self.port_id} "
+        if self.group_id is not None:
+            ret += f"group {self.group_id} "
+        if self.priority_level is not None:
+            ret += f"priority {self.priority_level} "
+        ret += "ingress " if self.ingress else "egress "
+        if self.user_id is not None:
+            ret += f"user_id {self.user_id} "
+        ret += f"pattern {self.pattern} / end "
+        ret += f"actions {self.actions} / end"
+        return ret
+
+
 class PacketOffloadFlag(Flag):
     """Flag representing the Packet Offload Features Flags in DPDK.
 
@@ -1717,6 +1755,25 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
 
         return TestPmdPortStats.parse(output)
 
+    def flow_create(self, flow_rule: FlowRule, verify: bool = True) -> None:
+        """Creates a flow rule in the testpmd session.
+
+        Args:
+            flow_rule: FlowRule object used for creating testpmd flow rule.
+            verify: If :data:`True`, the output of the command is scanned
+            to ensure the flow rule was created successfully.
+
+        Raises:
+            InteractiveCommandExecutionError: If flow rule is invalid.
+        """
+        flow_output = self.send_command(str(flow_rule))
+        if verify:
+            if "created" not in flow_output:
+                self._logger.debug(f"Failed to create flow rule:\n{flow_output}")
+                raise InteractiveCommandExecutionError(
+                    f"Failed to create flow rule:\n{flow_output}"
+                )
+
     @requires_stopped_ports
     def set_port_mtu(self, port_id: int, mtu: int, verify: bool = True) -> None:
         """Change the MTU of a port using testpmd.
-- 
2.44.0