<div dir="ltr"><div dir="ltr"><br></div><br><div class="gmail_quote gmail_quote_container"><div dir="ltr" class="gmail_attr">On Thu, Jun 26, 2025 at 3:56 PM Dean Marx <<a href="mailto:dmarx@iol.unh.edu">dmarx@iol.unh.edu</a>> wrote:<br></div><blockquote class="gmail_quote" style="margin:0px 0px 0px 0.8ex;border-left:1px solid rgb(204,204,204);padding-left:1ex">Add an RTE Flow API testing suite, which covers some basic<br>
synchronous Flow API rules that should be supported across PMDs.<br>
This suite will be added to over time, as the Flow API is too large<br>
to cover all in one suite, and sending one monolithic series<br>
would be impossible.<br>
<br>
Signed-off-by: Dean Marx <<a href="mailto:dmarx@iol.unh.edu" target="_blank">dmarx@iol.unh.edu</a>><br>
Reviewed-by: Patrick Robb <<a href="mailto:probb@iol.unh.edu" target="_blank">probb@iol.unh.edu</a>><br>
---<br>
 doc/api/dts/tests.TestSuite_rte_flow.rst |   8 +<br>
 dts/tests/TestSuite_rte_flow.py          | 790 +++++++++++++++++++++++<br>
 2 files changed, 798 insertions(+)<br>
 create mode 100644 doc/api/dts/tests.TestSuite_rte_flow.rst<br>
 create mode 100644 dts/tests/TestSuite_rte_flow.py<br>
<br>
diff --git a/doc/api/dts/tests.TestSuite_rte_flow.rst b/doc/api/dts/tests.TestSuite_rte_flow.rst<br>
new file mode 100644<br>
index 0000000000..cad96b2530<br>
--- /dev/null<br>
+++ b/doc/api/dts/tests.TestSuite_rte_flow.rst<br>
@@ -0,0 +1,8 @@<br>
+.. SPDX-License-Identifier: BSD-3-Clause<br>
+<br>
+checksum_offload Test Suite<br></blockquote><div><br></div><div>I'm just seeing this rst misnaming of the testsuite with the docs build now. I'm going to update and force push to next-dts.</div><div> </div><blockquote class="gmail_quote" style="margin:0px 0px 0px 0.8ex;border-left:1px solid rgb(204,204,204);padding-left:1ex">
+===========================<br>
+<br>
+.. automodule:: tests.TestSuite_rte_flow<br>
+   :members:<br>
+   :show-inheritance:<br>
\ No newline at end of file<br>
diff --git a/dts/tests/TestSuite_rte_flow.py b/dts/tests/TestSuite_rte_flow.py<br>
new file mode 100644<br>
index 0000000000..e70f7ea8d1<br>
--- /dev/null<br>
+++ b/dts/tests/TestSuite_rte_flow.py<br>
@@ -0,0 +1,790 @@<br>
+# SPDX-License-Identifier: BSD-3-Clause<br>
+# Copyright(c) 2025 University of New Hampshire<br>
+<br>
+"""RTE Flow testing suite.<br>
+<br>
+This suite verifies a range of flow rules built using patterns<br>
+and actions from the RTE Flow API. It would be impossible to cover<br>
+every valid flow rule, but this suite aims to test the most<br>
+important and common functionalities across PMDs.<br>
+<br>
+"""<br>
+<br>
+from collections.abc import Callable<br>
+from itertools import zip_longest<br>
+from typing import Any, Iterator, cast<br>
+<br>
+from scapy.layers.inet import IP, TCP, UDP<br>
+from scapy.layers.inet6 import IPv6<br>
+from scapy.layers.l2 import Dot1Q, Ether<br>
+from scapy.packet import Packet, Raw<br>
+<br>
+from framework.exception import InteractiveCommandExecutionError<br>
+from framework.remote_session.testpmd_shell import FlowRule, TestPmdShell<br>
+from framework.test_suite import TestSuite, func_test<br>
+from framework.testbed_model.capability import NicCapability, requires<br>
+<br>
+<br>
+@requires(NicCapability.FLOW_CTRL)<br>
+class TestRteFlow(TestSuite):<br>
+    """RTE Flow test suite.<br>
+<br>
+    This suite consists of 12 test cases:<br>
+    1. Queue Action Ethernet: Verifies queue actions with ethernet patterns<br>
+    2. Queue Action IP: Verifies queue actions with IPv4 and IPv6 patterns<br>
+    3. Queue Action L4: Verifies queue actions with TCP and UDP patterns<br>
+    4. Queue Action VLAN: Verifies queue actions with VLAN patterns<br>
+    5. Drop Action Eth: Verifies drop action with ethernet patterns<br>
+    6. Drop Action IP: Verifies drop actions with IPV4 and IPv6 patterns<br>
+    7. Drop Action L4: Verifies drop actions with TCP and UDP patterns<br>
+    8. Drop Action VLAN: Verifies drop actions with VLAN patterns<br>
+    9. Modify Field Action: Verifies packet modification patterns<br>
+    10. Egress Rules: Verifies previously covered rules are still valid as egress<br>
+    11. Jump Action: Verifies packet behavior given grouped flows<br>
+    12. Priority Attribute: Verifies packet behavior given flows with different priorities<br>
+<br>
+    """<br>
+<br>
+    def runner(<br>
+        self,<br>
+        verification_method: Callable[..., Any],<br>
+        flows: list[FlowRule],<br>
+        packets: list[Packet],<br>
+        port_id: int,<br>
+        expected_packets: list[Packet] | None = None,<br>
+        *args: Any,<br>
+        **kwargs: Any,<br>
+    ) -> None:<br>
+        """Runner method that validates each flow using the corresponding verification method.<br>
+<br>
+        Args:<br>
+            verification_method: Callable that performs verification logic.<br>
+            flows: List of flow rules to create and test.<br>
+            packets: List of packets corresponding to each flow.<br>
+            port_id: Number representing the port to create flows on.<br>
+            expected_packets: List of packets to check sent packets against in modification cases.<br>
+            *args: Additional positional arguments to pass to the verification method.<br>
+            **kwargs: Additional keyword arguments to pass to the verification method.<br>
+        """<br>
+<br>
+        def zip_lists(<br>
+            rules: list[FlowRule],<br>
+            packets1: list[Packet],<br>
+            packets2: list[Packet] | None,<br>
+        ) -> Iterator[tuple[FlowRule, Packet, Packet | None]]:<br>
+            """Method that creates an iterable zip containing lists used in runner.<br>
+<br>
+            Args:<br>
+                rules: List of flow rules.<br>
+                packets1: List of packets.<br>
+                packets2: Optional list of packets, excluded from zip if not passed to runner.<br>
+            """<br>
+            return cast(<br>
+                Iterator[tuple[FlowRule, Packet, Packet | None]],<br>
+                zip_longest(rules, packets1, packets2 or [], fillvalue=None),<br>
+            )<br>
+<br>
+        with TestPmdShell(rx_queues=4, tx_queues=4) as testpmd:<br>
+            for flow, packet, expected_packet in zip_lists(flows, packets, expected_packets):<br>
+                is_valid = testpmd.flow_validate(flow_rule=flow, port_id=port_id)<br>
+                self.verify_else_skip(is_valid, "flow rule failed validation.")<br>
+<br>
+                try:<br>
+                    flow_id = testpmd.flow_create(flow_rule=flow, port_id=port_id)<br>
+                except InteractiveCommandExecutionError:<br>
+                    self.log("Flow rule validation passed, but flow creation failed.")<br>
+                    self.verify(False, "Failed flow creation")<br>
+<br>
+                if verification_method == self.send_packet_and_verify:<br>
+                    verification_method(packet=packet, *args, **kwargs)<br>
+<br>
+                elif verification_method == self.send_packet_and_verify_queue:<br>
+                    verification_method(<br>
+                        packet=packet, test_queue=kwargs["test_queue"], testpmd=testpmd<br>
+                    )<br>
+<br>
+                elif verification_method == self.send_packet_and_verify_modification:<br>
+                    verification_method(packet=packet, expected_packet=expected_packet)<br>
+<br>
+                testpmd.flow_delete(flow_id, port_id=port_id)<br>
+<br>
+    def send_packet_and_verify(self, packet: Packet, should_receive: bool = True) -> None:<br>
+        """Generate a packet, send to the DUT, and verify it is forwarded back.<br>
+<br>
+        Args:<br>
+            packet: Scapy packet to send and verify.<br>
+            should_receive: Indicate whether the packet should be received.<br>
+        """<br>
+        received = self.send_packet_and_capture(packet)<br>
+        contains_packet = any(<br>
+            packet.haslayer(Raw) and b"xxxxx" in packet.load for packet in received<br>
+        )<br>
+        self.verify(<br>
+            should_receive == contains_packet,<br>
+            f"Packet was {'dropped' if should_receive else 'received'}",<br>
+        )<br>
+<br>
+    def send_packet_and_verify_queue(<br>
+        self, packet: Packet, test_queue: int, testpmd: TestPmdShell<br>
+    ) -> None:<br>
+        """Send packet and verify queue stats show packet was received.<br>
+<br>
+        Args:<br>
+            packet: Scapy packet to send to the SUT.<br>
+            test_queue: Represents the queue the test packet is being sent to.<br>
+            testpmd: TestPmdShell instance being used to send test packet.<br>
+        """<br>
+        testpmd.set_verbose(level=8)<br>
+        testpmd.start()<br>
+        self.send_packet_and_capture(packet=packet)<br>
+        verbose_output = testpmd.extract_verbose_output(testpmd.stop())<br>
+        received = False<br>
+        for testpmd_packet in verbose_output:<br>
+            if testpmd_packet.queue_id == test_queue:<br>
+                received = True<br>
+        self.verify(received, f"Expected packet was not received on queue {test_queue}")<br>
+<br>
+    def send_packet_and_verify_modification(self, packet: Packet, expected_packet: Packet) -> None:<br>
+        """Send packet and verify the expected modifications are present upon reception.<br>
+<br>
+        Args:<br>
+            packet: Scapy packet to send to the SUT.<br>
+            expected_packet: Scapy packet that should match the received packet.<br>
+        """<br>
+        received = self.send_packet_and_capture(packet)<br>
+<br>
+        # verify reception<br>
+        self.verify(received != [], "Packet was never received.")<br>
+<br>
+        self.log(f"SENT PACKET:     {packet.summary()}")<br>
+        self.log(f"EXPECTED PACKET: {expected_packet.summary()}")<br>
+        for packet in received:<br>
+            self.log(f"RECEIVED PACKET: {packet.summary()}")<br>
+<br>
+        expected_ip_dst = expected_packet[IP].dst if IP in expected_packet else None<br>
+        received_ip_dst = received[IP].dst if IP in received else None<br>
+<br>
+        expected_mac_dst = expected_packet[Ether].dst if Ether in expected_packet else None<br>
+        received_mac_dst = received[Ether].dst if Ether in received else None<br>
+<br>
+        # verify modification<br>
+        if expected_ip_dst is not None:<br>
+            self.verify(<br>
+                received_ip_dst == expected_ip_dst,<br>
+                f"IPv4 dst mismatch: expected {expected_ip_dst}, got {received_ip_dst}",<br>
+            )<br>
+<br>
+        if expected_mac_dst is not None:<br>
+            self.verify(<br>
+                received_mac_dst == expected_mac_dst,<br>
+                f"MAC dst mismatch: expected {expected_mac_dst}, got {received_mac_dst}",<br>
+            )<br>
+<br>
+    def send_packet_and_verify_jump(<br>
+        self,<br>
+        packets: list[Packet],<br>
+        flow_rules: list[FlowRule],<br>
+        test_queues: list[int],<br>
+        testpmd: TestPmdShell,<br>
+    ) -> None:<br>
+        """Create a testpmd session with every rule in the given list, verify jump behavior.<br>
+<br>
+        Args:<br>
+            packets: List of packets to send.<br>
+            flow_rules: List of flow rules to create in the same session.<br>
+            test_queues: List of Rx queue IDs each packet should be received on.<br>
+            testpmd: TestPmdShell instance to create flows on.<br>
+        """<br>
+        testpmd.set_verbose(level=8)<br>
+        for flow in flow_rules:<br>
+            is_valid = testpmd.flow_validate(flow_rule=flow, port_id=0)<br>
+            self.verify_else_skip(is_valid, "flow rule failed validation.")<br>
+<br>
+            try:<br>
+                testpmd.flow_create(flow_rule=flow, port_id=0)<br>
+            except InteractiveCommandExecutionError:<br>
+                self.log("Flow validation passed, but flow creation failed.")<br>
+                self.verify(False, "Failed flow creation")<br>
+<br>
+        for packet, test_queue in zip(packets, test_queues):<br>
+            testpmd.start()<br>
+            self.send_packet_and_capture(packet=packet)<br>
+            verbose_output = testpmd.extract_verbose_output(testpmd.stop())<br>
+            received = False<br>
+            for testpmd_packet in verbose_output:<br>
+                if testpmd_packet.queue_id == test_queue:<br>
+                    received = True<br>
+            self.verify(received, f"Expected packet was not received on queue {test_queue}")<br>
+<br>
+    @func_test<br>
+    def test_queue_action_ETH(self) -> None:<br>
+        """Validate flow rules with queue actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate queue.<br>
+        """<br>
+        packet_list = [<br>
+            Ether(src="02:00:00:00:00:00"),<br>
+            Ether(dst="02:00:00:00:00:00"),<br>
+            Ether(type=0x0800) / IP(),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth src is 02:00:00:00:00:00"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth dst is 02:00:00:00:00:00"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth type is 0x0800"], actions=["queue index 2"]<br>
+            ),<br>
+        ]<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify_queue,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            test_queue=2,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_queue_action_IP(self) -> None:<br>
+        """Validate flow rules with queue actions and IPv4/IPv6 patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate queue.<br>
+        """<br>
+        packet_list = [<br>
+            Ether() / IP(src="192.168.1.1"),<br>
+            Ether() / IP(dst="192.168.1.1"),<br>
+            Ether() / IP(ttl=64),<br>
+            Ether() / IPv6(src="2001:db8::1"),<br>
+            Ether() / IPv6(dst="2001:db8::2"),<br>
+            Ether() / IPv6() / UDP(),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 src is 192.168.1.1"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 dst is 192.168.1.1"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 ttl is 64"], actions=["queue index 2"]<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv6 src is 2001:db8::1"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv6 dst is 2001:db8::2"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv6 proto is 17"], actions=["queue index 2"]<br>
+            ),<br>
+        ]<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify_queue,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            test_queue=2,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_queue_action_L4(self) -> None:<br>
+        """Validate flow rules with queue actions and TCP/UDP patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate queue.<br>
+        """<br>
+        packet_list = [<br>
+            Ether() / IP() / TCP(sport=1234),<br>
+            Ether() / IP() / TCP(dport=80),<br>
+            Ether() / IP() / TCP(flags=0x02),<br>
+            Ether() / IP() / UDP(sport=5000),<br>
+            Ether() / IP() / UDP(dport=53),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 / tcp src is 1234"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 / tcp dst is 80"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 / tcp flags is 0x02"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 / udp src is 5000"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                pattern=["eth / ipv4 / udp dst is 53"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+        ]<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify_queue,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            test_queue=2,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_queue_action_VLAN(self) -> None:<br>
+        """Validate flow rules with queue actions and VLAN patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate queue.<br>
+        """<br>
+        packet_list = [Ether() / Dot1Q(vlan=100), Ether() / Dot1Q(type=0x0800)]<br>
+        flow_list = [<br>
+            FlowRule(direction="ingress", pattern=["eth / vlan"], actions=["queue index 2"]),<br>
+            FlowRule(direction="ingress", pattern=["eth / vlan"], actions=["queue index 2"]),<br>
+        ]<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify_queue,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            test_queue=2,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_drop_action_ETH(self) -> None:<br>
+        """Validate flow rules with drop actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is dropped.<br>
+<br>
+        One packet will be sent as a confidence check, to ensure packets are being<br>
+        received under normal circumstances.<br>
+        """<br>
+        packet_list = [<br>
+            Ether(src="02:00:00:00:00:00") / Raw(load="xxxxx"),<br>
+            Ether(dst="02:00:00:00:00:00") / Raw(load="xxxxx"),<br>
+            Ether(type=0x0800) / Raw(load="xxxxx"),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth src is 02:00:00:00:00:00"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth dst is 02:00:00:00:00:00"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="ingress", pattern=["eth type is 0x0800"], actions=["drop"]),<br>
+        ]<br>
+        # verify reception with test packet<br>
+        packet = Ether() / IP() / Raw(load="xxxxx")<br>
+        with TestPmdShell() as testpmd:<br>
+            testpmd.start()<br>
+            received = self.send_packet_and_capture(packet)<br>
+            self.verify(received != [], "Test packet was never received.")<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            should_receive=False,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_drop_action_IP(self) -> None:<br>
+        """Validate flow rules with drop actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is dropped.<br>
+<br>
+        One packet will be sent as a confidence check, to ensure packets are being<br>
+        received under normal circumstances.<br>
+        """<br>
+        packet_list = [<br>
+            Ether() / IP(src="192.168.1.1") / Raw(load="xxxxx"),<br>
+            Ether() / IP(dst="192.168.1.1") / Raw(load="xxxxx"),<br>
+            Ether() / IP(ttl=64) / Raw(load="xxxxx"),<br>
+            Ether() / IPv6(src="2001:db8::1") / Raw(load="xxxxx"),<br>
+            Ether() / IPv6(dst="2001:db8::1") / Raw(load="xxxxx"),<br>
+            Ether() / IPv6() / UDP() / Raw(load="xxxxx"),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 src is 192.168.1.1"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 dst is 192.168.1.1"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="ingress", pattern=["eth / ipv4 ttl is 64"], actions=["drop"]),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv6 src is 2001:db8::1"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv6 dst is 2001:db8::2"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="ingress", pattern=["eth / ipv6 proto is 17"], actions=["drop"]),<br>
+        ]<br>
+        # verify reception with test packet<br>
+        packet = Ether() / IP() / Raw(load="xxxxx")<br>
+        with TestPmdShell() as testpmd:<br>
+            testpmd.start()<br>
+            received = self.send_packet_and_capture(packet)<br>
+            self.verify(received != [], "Test packet was never received.")<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            should_receive=False,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_drop_action_L4(self) -> None:<br>
+        """Validate flow rules with drop actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is dropped.<br>
+<br>
+        One packet will be sent as a confidence check, to ensure packets are being<br>
+        received under normal circumstances.<br>
+        """<br>
+        packet_list = [<br>
+            Ether() / IP() / TCP(sport=1234) / Raw(load="xxxxx"),<br>
+            Ether() / IP() / TCP(dport=80) / Raw(load="xxxxx"),<br>
+            Ether() / IP() / TCP(flags=0x02) / Raw(load="xxxxx"),<br>
+            Ether() / IP() / UDP(sport=5000) / Raw(load="xxxxx"),<br>
+            Ether() / IP() / UDP(dport=53) / Raw(load="xxxxx"),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 / tcp src is 1234"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="ingress", pattern=["eth / ipv4 / tcp dst is 80"], actions=["drop"]),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 / tcp flags is 0x02"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress", pattern=["eth / ipv4 / udp src is 5000"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="ingress", pattern=["eth / ipv4 / udp dst is 53"], actions=["drop"]),<br>
+        ]<br>
+        # verify reception with test packet<br>
+        packet = Ether() / IP() / Raw(load="xxxxx")<br>
+        with TestPmdShell() as testpmd:<br>
+            testpmd.start()<br>
+            received = self.send_packet_and_capture(packet)<br>
+            self.verify(received != [], "Test packet was never received.")<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            should_receive=False,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_drop_action_VLAN(self) -> None:<br>
+        """Validate flow rules with drop actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is dropped.<br>
+<br>
+        One packet will be sent as a confidence check, to ensure packets are being<br>
+        received under normal circumstances.<br>
+        """<br>
+        packet_list = [<br>
+            Ether() / Dot1Q(vlan=100) / Raw(load="xxxxx"),<br>
+            Ether() / Dot1Q(type=0x0800) / Raw(load="xxxxx"),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(direction="ingress", pattern=["eth / vlan"], actions=["drop"]),<br>
+            FlowRule(direction="ingress", pattern=["eth / vlan"], actions=["drop"]),<br>
+        ]<br>
+        # verify reception with test packet<br>
+        packet = Ether() / IP() / Raw(load="xxxxx")<br>
+        with TestPmdShell() as testpmd:<br>
+            testpmd.start()<br>
+            received = self.send_packet_and_capture(packet)<br>
+            self.verify(received != [], "Test packet was never received.")<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            should_receive=False,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_modify_actions(self) -> None:<br>
+        """Validate flow rules with actions that modify that packet during transmission.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Verify packet is received with the new attributes.<br>
+        """<br>
+        packet_list = [Ether() / IP(src="192.68.1.1"), Ether(src="02:00:00:00:00:00")]<br>
+        flow_list = [<br>
+            # rule to copy IPv4 src to IPv4 dst<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=1,<br>
+                pattern=["eth"],<br>
+                actions=[<br>
+                    "modify_field op set dst_type ipv4_dst src_type ipv4_src width 32"<br>
+                    " / queue index 0"<br>
+                ],<br>
+            ),<br>
+            # rule to copy src MAC to dst MAC<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=1,<br>
+                pattern=["eth"],<br>
+                actions=[<br>
+                    "modify_field op set dst_type mac_dst src_type mac_src width 48 / queue index 0"<br>
+                ],<br>
+            ),<br>
+        ]<br>
+        expected_packet_list = [Ether() / IP(dst="192.68.1.1"), Ether(dst="02:00:00:00:00:00")]<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify_modification,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=0,<br>
+            expected_packets=expected_packet_list,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_egress_rules(self) -> None:<br>
+        """Validate flow rules with egress directions.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is dropped.<br>
+<br>
+        One packet will be sent as a confidence check, to ensure packets are being<br>
+        received under normal circumstances.<br>
+        """<br>
+        packet_list = [<br>
+            Ether(src="02:00:00:00:00:00"),<br>
+            Ether() / IP(src="192.168.1.1"),<br>
+            IP() / TCP(sport=1234),<br>
+            IP() / UDP(sport=5000),<br>
+        ]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="egress", pattern=["eth src is 02:00:00:00:00:00"], actions=["drop"]<br>
+            ),<br>
+            FlowRule(direction="egress", pattern=["ipv4 src is 192.168.1.1"], actions=["drop"]),<br>
+            FlowRule(direction="egress", pattern=["tcp src is 1234"], actions=["drop"]),<br>
+            FlowRule(direction="egress", pattern=["udp src is 5000"], actions=["drop"]),<br>
+        ]<br>
+        # verify reception with test packet<br>
+        packet = Ether() / IP() / Raw(load="xxxxx")<br>
+        with TestPmdShell() as testpmd:<br>
+            testpmd.start()<br>
+            received = self.send_packet_and_capture(packet)<br>
+            self.verify(received != [], "Test packet was never received.")<br>
+        self.runner(<br>
+            verification_method=self.send_packet_and_verify,<br>
+            flows=flow_list,<br>
+            packets=packet_list,<br>
+            port_id=1,<br>
+            should_receive=False,<br>
+        )<br>
+<br>
+    @func_test<br>
+    def test_jump_action(self) -> None:<br>
+        """Validate flow rules with different group levels and jump actions.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd with the necessary configuration.<br>
+            Create each flow rule in testpmd.<br>
+            Send each packet in the list, check Rx queue ID.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate Rx queue.<br>
+        """<br>
+        packet_list = [Ether() / IP(), Ether() / IP() / TCP(), Ether() / IP() / UDP()]<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=0,<br>
+                pattern=["eth / ipv4 / tcp"],<br>
+                actions=["jump group 1"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=0,<br>
+                pattern=["eth / ipv4 / udp"],<br>
+                actions=["jump group 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=1,<br>
+                pattern=["eth / ipv4 / tcp"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=2,<br>
+                pattern=["eth / ipv4 / udp"],<br>
+                actions=["queue index 3"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                group_id=0,<br>
+                pattern=["eth / ipv4"],<br>
+                actions=["queue index 1"],<br>
+            ),<br>
+        ]<br>
+        expected_queue_list = [1, 2, 3]<br>
+        with TestPmdShell(rx_queues=4, tx_queues=4) as testpmd:<br>
+            self.send_packet_and_verify_jump(<br>
+                packets=packet_list,<br>
+                flow_rules=flow_list,<br>
+                test_queues=expected_queue_list,<br>
+                testpmd=testpmd,<br>
+            )<br>
+<br>
+    @func_test<br>
+    def test_priority_attribute(self) -> None:<br>
+        """Validate flow rules with queue actions and ethernet patterns.<br>
+<br>
+        Steps:<br>
+            Create a list of packets to test, with a corresponding flow list.<br>
+            Launch testpmd.<br>
+            Create first flow rule in flow list.<br>
+            Send first packet in packet list, capture verbose output.<br>
+            Delete flow rule, repeat for all flows/packets.<br>
+<br>
+        Verify:<br>
+            Check that each packet is received on the appropriate queue.<br>
+        """<br>
+        test_packet = Ether() / IP() / Raw()<br>
+        flow_list = [<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                priority_level=3,<br>
+                pattern=["eth / ipv4"],<br>
+                actions=["queue index 1"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                priority_level=2,<br>
+                pattern=["eth / ipv4"],<br>
+                actions=["queue index 2"],<br>
+            ),<br>
+            FlowRule(<br>
+                direction="ingress",<br>
+                priority_level=1,<br>
+                pattern=["eth / ipv4"],<br>
+                actions=["queue index 3"],<br>
+            ),<br>
+        ]<br>
+        expected_queue_list = [1, 2, 3]<br>
+        with TestPmdShell(rx_queues=4, tx_queues=4) as testpmd:<br>
+            testpmd.set_verbose(level=8)<br>
+            for flow, expected_queue in zip(flow_list, expected_queue_list):<br>
+                is_valid = testpmd.flow_validate(flow_rule=flow, port_id=0)<br>
+                self.verify_else_skip(is_valid, "flow rule failed validation.")<br>
+                try:<br>
+                    testpmd.flow_create(flow_rule=flow, port_id=0)<br>
+                except InteractiveCommandExecutionError:<br>
+                    self.log("Flow rule validation passed, but flow creation failed.")<br>
+                    self.verify(False, "Failed flow creation")<br>
+                testpmd.start()<br>
+                self.send_packet_and_capture(test_packet)<br>
+                verbose_output = testpmd.extract_verbose_output(testpmd.stop())<br>
+                received = False<br>
+                for testpmd_packet in verbose_output:<br>
+                    if testpmd_packet.queue_id == expected_queue:<br>
+                        received = True<br>
+                self.verify(received, f"Packet was not received on queue {expected_queue}")<br>
-- <br>
2.49.0<br>
<br>
</blockquote></div></div>