<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>