[PATCH v9 10/10] dts: add selective Rx tests
Thomas Monjalon
thomas at monjalon.net
Sat Jun 6 01:33:50 CEST 2026
Add TestSuite_rx_split with 7 test cases:
- 3 positive: headers only, payload only, two non-contiguous segments
- 4 negative: missing offload flag, out-of-range, overlap, all-discard
Add selective Rx capability detection via testpmd "show port info".
The test suite could be completed later for the basic buffer split
configuration based on offsets or protocols.
Signed-off-by: Thomas Monjalon <thomas at monjalon.net>
---
dts/api/capabilities.py | 2 +
dts/api/testpmd/__init__.py | 17 ++
dts/api/testpmd/types.py | 6 +
dts/framework/testbed_model/capability.py | 2 +
dts/tests/TestSuite_rx_split.py | 277 ++++++++++++++++++++++
5 files changed, 304 insertions(+)
create mode 100644 dts/tests/TestSuite_rx_split.py
diff --git a/dts/api/capabilities.py b/dts/api/capabilities.py
index 09bc538523..b0c1d81d36 100644
--- a/dts/api/capabilities.py
+++ b/dts/api/capabilities.py
@@ -136,6 +136,8 @@ class NicCapability(IntEnum):
#: Device supports all VLAN capabilities.
PORT_RX_OFFLOAD_VLAN = auto()
QUEUE_RX_OFFLOAD_VLAN = auto()
+ #: Device supports selective Rx.
+ SELECTIVE_RX = auto()
#: Device supports Rx queue setup after device started.
RUNTIME_RX_QUEUE_SETUP = auto()
#: Device supports Tx queue setup after device started.
diff --git a/dts/api/testpmd/__init__.py b/dts/api/testpmd/__init__.py
index e9187440bb..6973a64573 100644
--- a/dts/api/testpmd/__init__.py
+++ b/dts/api/testpmd/__init__.py
@@ -1409,6 +1409,23 @@ def get_capabilities_show_port_info(
self.ports[0].device_capabilities,
)
+ def get_capabilities_selective_rx(
+ self,
+ supported_capabilities: MutableSet["NicCapability"],
+ unsupported_capabilities: MutableSet["NicCapability"],
+ ) -> None:
+ """Get selective Rx capability from show port info.
+
+ Args:
+ supported_capabilities: Supported capabilities will be added to this set.
+ unsupported_capabilities: Unsupported capabilities will be added to this set.
+ """
+ port_info = self.show_port_info(self.ports[0].id)
+ if port_info.selective_rx:
+ supported_capabilities.add(NicCapability.SELECTIVE_RX)
+ else:
+ unsupported_capabilities.add(NicCapability.SELECTIVE_RX)
+
def get_capabilities_mcast_filtering(
self,
supported_capabilities: MutableSet["NicCapability"],
diff --git a/dts/api/testpmd/types.py b/dts/api/testpmd/types.py
index 0d322aece2..6f1eaf47cc 100644
--- a/dts/api/testpmd/types.py
+++ b/dts/api/testpmd/types.py
@@ -614,6 +614,12 @@ def _validate(info: str) -> str | None:
metadata=VLANOffloadFlag.make_parser(),
)
+ #: Selective Rx support
+ selective_rx: bool = field(
+ default=False,
+ metadata=TextParser.find(r"Selective Rx: supported"),
+ )
+
#: Maximum size of RX buffer
max_rx_bufsize: int | None = field(
default=None, metadata=TextParser.find_int(r"Maximum size of RX buffer: (\d+)")
diff --git a/dts/framework/testbed_model/capability.py b/dts/framework/testbed_model/capability.py
index 96e1cd449f..b10799ea4b 100644
--- a/dts/framework/testbed_model/capability.py
+++ b/dts/framework/testbed_model/capability.py
@@ -324,6 +324,8 @@ def mapping(cap: NicCapability) -> TestPmdNicCapability:
| NicCapability.FLOW_SHARED_OBJECT_KEEP
):
return (TestPmd.get_capabilities_show_port_info, None)
+ case NicCapability.SELECTIVE_RX:
+ return (TestPmd.get_capabilities_selective_rx, None)
case NicCapability.MCAST_FILTERING:
return (TestPmd.get_capabilities_mcast_filtering, None)
case NicCapability.FLOW_CTRL:
diff --git a/dts/tests/TestSuite_rx_split.py b/dts/tests/TestSuite_rx_split.py
new file mode 100644
index 0000000000..0c7913bbd8
--- /dev/null
+++ b/dts/tests/TestSuite_rx_split.py
@@ -0,0 +1,277 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2026 NVIDIA Corporation & Affiliates
+
+"""Rx split test suite.
+
+Test configuring a packet split on Rx,
+and discarding some segments (selective Rx) at NIC level.
+"""
+
+from collections.abc import Callable
+from typing import Any
+
+from scapy.layers.inet import IP
+from scapy.layers.l2 import Ether
+from scapy.packet import Packet, Raw
+
+from api.capabilities import (
+ NicCapability,
+ requires_nic_capability,
+)
+from api.packet import adjust_addresses, send_packet_and_capture
+from api.test import fail, verify
+from api.testpmd import TestPmd
+from api.testpmd.config import SimpleForwardingModes
+from api.testpmd.types import RxOffloadCapability, TxOffloadCapability
+from framework.exception import InteractiveCommandExecutionError
+from framework.test_suite import TestSuite, func_test
+
+PAYLOAD = bytes(range(256))
+ETHER_HDR_LEN = len(Ether())
+IP_HDR_LEN = len(IP())
+ETHER_IP_HDR_LEN = ETHER_HDR_LEN + IP_HDR_LEN
+ETHER_MIN_FRAME_LEN = 60
+
+
+ at requires_nic_capability(NicCapability.PORT_RX_OFFLOAD_BUFFER_SPLIT)
+ at requires_nic_capability(NicCapability.SELECTIVE_RX)
+class TestRxSplit(TestSuite):
+ """Rx split test suite.
+
+ Configure testpmd with various Rx segment offset/length combinations
+ and verify that only the requested portions of the packet are received
+ and forwarded.
+ """
+
+ def _create_testpmd(self, **kwargs: Any) -> TestPmd:
+ """Create a TestPmd instance with defaults overridden by kwargs."""
+ defaults: dict[str, Any] = {
+ "forward_mode": SimpleForwardingModes.io,
+ "rx_offloads": RxOffloadCapability.BUFFER_SPLIT | RxOffloadCapability.SCATTER,
+ }
+ return TestPmd(**{**defaults, **kwargs})
+
+ def _build_packet(self) -> Packet:
+ """Build a test packet with an incrementing byte pattern payload."""
+ packet = Ether() / IP() / Raw(load=PAYLOAD)
+ return adjust_addresses([packet])[0]
+
+ def _start_and_verify(self, testpmd: TestPmd, expected: Callable[[bytes], bytes]) -> None:
+ """Start testpmd, send the default packet, and verify received bytes."""
+ testpmd.start()
+ packet = self._build_packet()
+ self._send_and_verify(testpmd, packet, expected(bytes(packet)))
+
+ def _send_and_verify(self, testpmd: TestPmd, tg_packet: Packet, expected: bytes) -> None:
+ """Clear stats, send a packet, and verify received content and stats.
+
+ Args:
+ testpmd: The running testpmd instance.
+ tg_packet: The packet to send by Scapy on the TG.
+ expected: Expected raw bytes received by testpmd on the SUT.
+ """
+ testpmd.clear_port_stats_all(verify=False)
+
+ # TG send, SUT receive and forward back, then TG capture
+ sut_len = len(expected)
+ capture_len = max(sut_len, ETHER_MIN_FRAME_LEN)
+ received = send_packet_and_capture(tg_packet)
+ verify(
+ len(received) > 0,
+ "Did not receive any packets.",
+ )
+
+ recv_bytes = bytes(received[0])
+ verify(
+ len(recv_bytes) == capture_len,
+ f"Expected packet length {capture_len}, got {len(recv_bytes)}.",
+ )
+ verify(
+ recv_bytes[:sut_len] == expected,
+ "Received packet content does not match expected bytes.",
+ )
+
+ all_stats, _ = testpmd.show_port_stats_all()
+ total_rx_packets = sum(s.rx_packets for s in all_stats)
+ total_rx_bytes = sum(s.rx_bytes for s in all_stats)
+ verify(
+ total_rx_packets == 1,
+ f"Expected 1 Rx packet, got {total_rx_packets}.",
+ )
+ verify(
+ total_rx_bytes == sut_len,
+ f"Expected {sut_len} Rx bytes, got {total_rx_bytes}.",
+ )
+
+ def _verify_port_start_fails(self, testpmd: TestPmd, message: str) -> None:
+ """Verify that starting ports fails, then drain testpmd output."""
+ try:
+ testpmd.start_all_ports()
+ fail(message)
+ except InteractiveCommandExecutionError:
+ testpmd.stop(verify=False)
+
+ @func_test
+ def selective_rx_headers(self) -> None:
+ """Keep only the Ethernet + IP headers, discard the rest.
+
+ Steps:
+ Start testpmd with rxpkts, mbuf-size and buffer split enabled.
+ Configure the payload discard segment with length 0 (rest).
+ Send an Ether/IP/payload packet.
+
+ Verify:
+ Received packet has Ether + IP headers only.
+ Port stats show expected rx_packets and rx_bytes.
+ """
+ with self._create_testpmd(
+ rx_segments_length=[ETHER_IP_HDR_LEN, 0],
+ mbuf_size=[256, 0],
+ ) as testpmd:
+
+ def expected(packet: bytes) -> bytes:
+ return packet[:ETHER_IP_HDR_LEN]
+
+ self._start_and_verify(testpmd, expected)
+
+ @func_test
+ def selective_rx_headers_discard_length(self) -> None:
+ """Keep only the Ethernet + IP headers, discard the remaining length.
+
+ Steps:
+ Start testpmd with rxpkts, mbuf-size and buffer split enabled.
+ Configure the payload discard segment with an explicit length.
+ Send an Ether/IP/payload packet.
+
+ Verify:
+ Received packet has Ether + IP headers only.
+ Port stats show expected rx_packets and rx_bytes.
+ """
+ with self._create_testpmd(
+ rx_segments_length=[ETHER_IP_HDR_LEN, len(PAYLOAD)],
+ mbuf_size=[256, 0],
+ ) as testpmd:
+
+ def expected(packet: bytes) -> bytes:
+ return packet[:ETHER_IP_HDR_LEN]
+
+ self._start_and_verify(testpmd, expected)
+
+ @func_test
+ def selective_rx_payload_only(self) -> None:
+ """Skip the Ethernet + IP headers, keep only the payload.
+
+ Steps:
+ Start testpmd with rxpkts, mbuf-size and buffer split enabled.
+ Send an Ether/IP/payload packet.
+
+ Verify:
+ Received packet is matching the original payload.
+ Port stats show expected rx_packets and rx_bytes.
+ """
+ with self._create_testpmd(
+ rx_segments_length=[ETHER_IP_HDR_LEN, len(PAYLOAD)],
+ mbuf_size=[0, 512],
+ ) as testpmd:
+
+ def expected(_: bytes) -> bytes:
+ return PAYLOAD
+
+ self._start_and_verify(testpmd, expected)
+
+ @func_test
+ def selective_rx_two_segments(self) -> None:
+ """Keep the IP header and the middle of the payload, skip the rest.
+
+ Steps:
+ Start testpmd with rxpkts, mbuf-size, buffer split
+ and multi-segment Tx enabled.
+ Send an Ether/IP/payload packet.
+
+ Verify:
+ Received packet is matching the IP header and middle of payload.
+ Port stats show expected rx_packets and rx_bytes.
+ """
+ payload_offset = 100
+ payload_length = 100
+ with self._create_testpmd(
+ tx_offloads=TxOffloadCapability.MULTI_SEGS,
+ rx_segments_length=[ETHER_HDR_LEN, IP_HDR_LEN, payload_offset, payload_length, 0],
+ mbuf_size=[0, 256, 0, 256, 0],
+ ) as testpmd:
+
+ def expected(packet: bytes) -> bytes:
+ payload_start = ETHER_IP_HDR_LEN + payload_offset
+ return (
+ packet[ETHER_HDR_LEN:ETHER_IP_HDR_LEN]
+ + packet[payload_start : payload_start + payload_length]
+ )
+
+ self._start_and_verify(testpmd, expected)
+
+ @func_test
+ def selective_rx_no_offload(self) -> None:
+ """Configure selective Rx with buffer split disabled.
+
+ Steps:
+ Start testpmd with rxpkts, mbuf-size including a discard segment,
+ buffer split disabled, and device start disabled.
+ Attempt to start ports.
+
+ Verify:
+ Queue configuration fails.
+ """
+ with self._create_testpmd(
+ rx_offloads=None,
+ rx_segments_length=[ETHER_IP_HDR_LEN, 0],
+ mbuf_size=[256, 0],
+ disable_device_start=True,
+ ) as testpmd:
+ self._verify_port_start_fails(
+ testpmd,
+ "Expected configuration to fail with buffer split disabled.",
+ )
+
+ @func_test
+ def selective_rx_segment_exceeds_mbuf(self) -> None:
+ """Configure selective Rx with segment length exceeding mbuf capacity.
+
+ Steps:
+ Start testpmd with rxpkts larger than mbuf-size,
+ buffer split enabled, and device start disabled.
+ Attempt to start ports.
+
+ Verify:
+ Queue configuration fails.
+ """
+ with self._create_testpmd(
+ rx_segments_length=[4096, 0],
+ mbuf_size=[128, 0],
+ disable_device_start=True,
+ ) as testpmd:
+ self._verify_port_start_fails(
+ testpmd,
+ "Expected configuration to fail with segment > mbuf size.",
+ )
+
+ @func_test
+ def selective_rx_all_discard(self) -> None:
+ """Configure selective Rx with only discard segment.
+
+ Steps:
+ Start testpmd with only discard segment,
+ buffer split enabled, and device start disabled.
+ Attempt to start ports.
+
+ Verify:
+ Queue configuration fails.
+ """
+ with self._create_testpmd(
+ rx_segments_length=[0],
+ mbuf_size=[0],
+ disable_device_start=True,
+ ) as testpmd:
+ self._verify_port_start_fails(
+ testpmd,
+ "Expected configuration to fail with only discard segment.",
+ )
--
2.54.0
More information about the dev
mailing list